messenger

README

Messenger

Outbox pattern implementation in Go.

Abstracts from handling the retrieve of the messages from many sources and publishing to many queues. Just store the messages you want to publish.

   

Documentation

Index

• Variables
• type ErrorLogger
• type GenericMessage
  • func NewMessage(payload []byte) (*GenericMessage, error)
  • func (m *GenericMessage) At() time.Time
  • func (m *GenericMessage) ID() string
  • func (m *GenericMessage) Metadata() map[string]string
  • func (m *GenericMessage) Payload() []byte
  • func (m *GenericMessage) Published() bool
  • func (m *GenericMessage) SetMetadata(key, value string) *GenericMessage
• type Message
• type Messenger
  • func NewMessenger(store Store, publisher Publisher, opts ...Option) *Messenger
  • func (w *Messenger) Clean(ctx context.Context) error
  • func (w *Messenger) Publish(ctx context.Context) error
  • func (w *Messenger) Start(ctx context.Context) error
• type Option
  • func WithCleanUp(expiration time.Duration) Option
  • func WithErrorLogger(l ErrorLogger) Option
  • func WithInterval(p time.Duration) Option
  • func WithPublishBatchSize(bs int) Option
• type Publisher
• type Store

Variables

var ErrEmptyMessagePayload = errors.New("empty message payload")

ErrEmptyMessagePayload is the error returned when the message payload is empty.

Types

type ErrorLogger

type ErrorLogger interface {
	Error(err error)
}

ErrorLogger is the interface that wraps the basic message publishing.

type GenericMessage

type GenericMessage struct {
	// Unique identifier for the message.
	MsgID string
	// Contains the message header to be sent by the messenger to the message queue.
	MsgMetadata map[string]string
	// Payload is the message payload. Must not be empty
	MsgPayload []byte
	// Message is published to broker or not.
	MsgPublished bool
	// At represent the moment of the message creation.
	MsgAt time.Time
}

GenericMessage represents a message to be sent to message message queue. It implements the Message interface.

func NewMessage

func NewMessage(payload []byte) (*GenericMessage, error)

NewMessage returns a new Message given a payload.

func (*GenericMessage) At

func (m *GenericMessage) At() time.Time

At returns message creation moment.

func (*GenericMessage) ID

func (m *GenericMessage) ID() string

ID returns the unique identifier of the message.

func (*GenericMessage) Metadata

func (m *GenericMessage) Metadata() map[string]string

Metadata returns the message metadata.

func (*GenericMessage) Payload

func (m *GenericMessage) Payload() []byte

Payload returns the message payload.

func (*GenericMessage) Published

func (m *GenericMessage) Published() bool

Published returns if message is published to broker.

func (*GenericMessage) SetMetadata

func (m *GenericMessage) SetMetadata(key, value string) *GenericMessage

SetMetadata sets the given key-value pair to the message metadata. If the key already exists, it replaces the value.

type Message

type Message interface {
	ID() string
	Metadata() map[string]string
	Payload() []byte
	Published() bool
	At() time.Time
}

A Message represents a message to be sent to message message queue.

type Messenger

type Messenger struct {
	// contains filtered or unexported fields
}

Messenger is responsible of publishing messages from datastore to publisher, and cleaning already published messages.

func NewMessenger

func NewMessenger(store Store, publisher Publisher, opts ...Option) *Messenger

NewMessenger returns a `Messenger` instance with defaults.

• Publish batch size: 100
• Publish period: 1s
• Golang standard error logger.

func (*Messenger) Clean

func (w *Messenger) Clean(ctx context.Context) error

Clean runs once the message cleaning process given a message expiration time.

func (*Messenger) Publish

func (w *Messenger) Publish(ctx context.Context) error

Publish runs once publishing process.

func (*Messenger) Start

func (w *Messenger) Start(ctx context.Context) error

Start runs the process of publishing/cleaning messages every period. In case there is a publish error, it will call to error handler without stopping the process. If a fatal error happens, ex, cant connect to datastore it will stop the process.

type Option

type Option func(*Messenger)

Option defines the optional parameters for messenger.

func WithCleanUp

func WithCleanUp(expiration time.Duration) Option

WithCleanUp enables cleanup process setting an expiration time for messages.

func WithErrorLogger

func WithErrorLogger(l ErrorLogger) Option

WithErrorLogger replaces the default error logger.

func WithInterval

func WithInterval(p time.Duration) Option

WithInterval replaces the default interval duration.

func WithPublishBatchSize

func WithPublishBatchSize(bs int) Option

WithPublishBatchSize replaces the default published batch size.

type Publisher

type Publisher interface {
	// Sends the message to broker.
	Publish(ctx context.Context, msg Message) error
}

Publisher is the interface that wraps the basic message publishing.

type Store

type Store interface {
	// List unpublished messages with a batch size
	Messages(ctx context.Context, batch int) ([]Message, error)
	// Mark as published the given messages, if one of the messages fails will not update any of the messages
	Published(ctx context.Context, msg ...Message) error
	// Deletes messages marked as published and older than expiration period from datastore.
	DeletePublishedByExpiration(ctx context.Context, exp time.Duration) error
}

Store is the interface that wraps the message retrieval and update methods.