pubsub

README

go-pubsub

A simple abstraction for message publishing and consumption that presents a uniform API, regardless of the underlying implementation.

Current implementations and their status

Implementation	Status
Mock (in memory testing implementation)	incomplete
Apache Kafka	beta
Nats streaming	beta
AMQP	incomplete
AWS SQS	beta
AWS SNS	beta

The API is not yet guaranteed but changes should be minimal from now.

Documentation

Index

• type ConcurrentMessageSource
• type ConsumerErrorHandler
• type ConsumerMessage
• type ConsumerMessageHandler
• type MessageSink
• type MessageSource
• type ProducerMessage
• type SimpleProducerMessage
  • func (sm SimpleProducerMessage) Marshal() ([]byte, error)
• type Status
• type Statuser

Types

type ConcurrentMessageSource

type ConcurrentMessageSource interface {
	MessageSource
	// ConsumeMessagesConcurrently provides similar functionality to ConsumeMessages but utilises
	// multiple routines to achieve concurrency, the exact amount of routines that will
	// be created depends on the underlying technology
	ConsumeMessagesConcurrently(ctx context.Context, handler ConsumerMessageHandler, onError ConsumerErrorHandler) error
}

ConcurrentMessageSource concurrent message consumer

type ConsumerErrorHandler

type ConsumerErrorHandler func(ConsumerMessage, error) error

ConsumerErrorHandler is invoked when a message can not be processed. If an error handler returns an error itself, processing of messages is aborted

type ConsumerMessage

type ConsumerMessage struct {
	Data []byte
}

type ConsumerMessageHandler

type ConsumerMessageHandler func(ConsumerMessage) error

ConsumerMessageHandler processes messages, and should return an error if it is unable to do so.

type MessageSink

type MessageSink interface {
	io.Closer
	PutMessage(ProducerMessage) error
	Statuser
}

type MessageSource

type MessageSource interface {
	// Consume messages will block until error or until the context is done.
	ConsumeMessages(ctx context.Context, handler ConsumerMessageHandler, onError ConsumerErrorHandler) error
	Statuser
}

type ProducerMessage

type ProducerMessage interface {
	// Marshal returns the message in wire format
	Marshal() ([]byte, error)
}

ProducerMessage is an individual message that can be sent

type SimpleProducerMessage

type SimpleProducerMessage []byte

SimpleProducerMessage is a convenience type for simply sending byte slices.

func (SimpleProducerMessage) Marshal

func (sm SimpleProducerMessage) Marshal() ([]byte, error)

type Status

type Status struct {
	// Working indicates whether the source or sink is in a working state
	Working bool
	// Problems indicates and problems with the source or sink, whether or not they prevent it working.
	Problems []string
}

Status represents a snapshot of the state of a source or sink.

type Statuser

type Statuser interface {
	Status() (*Status, error)
}

Statuser is the interface that wraps the Status method.