outbox

Documentation

Index

• Constants
• Variables
• type Broker
• type Client
• type Execer
• type Logger
• type Option
  • func EnableDebugMode() Option
  • func WithIterationRate(dur time.Duration) Option
  • func WithLogger(logger Logger) Option
  • func WithPublishTimeout(dur time.Duration) Option
• type Outbox
  • func NewOutbox(broker Broker, conn *sql.DB, opts ...Option) *Outbox
  • func (o *Outbox) Start(ctx context.Context) error
  • func (o *Outbox) Writer() Client
• type Record
  • func NewRecord(id uuid.UUID, eventType string, payload json.Marshaler) *Record
  • func (r *Record) Done()
  • func (r *Record) Fail()
  • func (r *Record) Null()
• type Status

Constants

const (
	// DefaultIterationRate is the timeout after which all outbox events
	// in the outbox table are sent to the broker.
	//
	// Default: 5 * time.Second.
	DefaultIterationRate = 5 * time.Second
	// DefaultPublishTimeout is the timeout after which the publication
	// of the current event is canceled. The current event marked as 'not yet published', and
	// processing continues.
	//
	// Default: 2 * time.Second.
	DefaultPublishTimeout = 2 * time.Second
	// DebugMode enables additional logs for debug outbox process.
	// Now, this option do nothing.
	//
	// Default: false.
	DebugMode = false
)

Variables

var DefaultLogger = log.Default()

var ErrNoRecrods = errors.New("no records in the outbox")

Types

type Broker

type Broker interface {
	Publish(ctx context.Context, subject string, payload []byte) error
}

type Client

type Client interface {
	WriteOutbox(context.Context, Execer, *Record) error
}

Client provides possibility to set outbox record to the outbox table. Insertion must be in the same transaction as the produced action.

type Execer

type Execer interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
}

type Logger

type Logger interface {
	Print(...any)
	Printf(string, ...any)
}

type Option

type Option func(config) config

Option sets specific configuration to the Outbox.

func EnableDebugMode

func EnableDebugMode() Option

func WithIterationRate

func WithIterationRate(dur time.Duration) Option

WithIterationRate sets new interval for sending events from the outbox table.

func WithLogger

func WithLogger(logger Logger) Option

WithLogger sets custom implementation of Logger.

func WithPublishTimeout

func WithPublishTimeout(dur time.Duration) Option

WithPublishTimeout sets a custom timeout for publishing next event.

type Outbox

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

Outbox is struct that implement outbox pattern.

Writing all outgoing events in a temporary table in the same transaction in which we process the action associated with this event. Then we try to publish the event to the broker with specific timeout until the event is sent.

More about outbox pattern you can read at https://microservices.io/patterns/data/transactional-outbox.html.

func NewOutbox

func NewOutbox(broker Broker, conn *sql.DB, opts ...Option) *Outbox

NewOutbox creates new outbox implementation.

func (*Outbox) Start

func (o *Outbox) Start(ctx context.Context) error

Start initialize outbox table and start worker process. Worker is process that send outgoing messages to broker.

func (*Outbox) Writer

func (o *Outbox) Writer() Client

Writer creates new Client to write outgoing events to the temporary table.

type Record

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

Record is event that should be processed by outbox worker.

func NewRecord

func NewRecord(id uuid.UUID, eventType string, payload json.Marshaler) *Record

NewRecord creates new record that can be processed by outbox worker.

Parameters:

id - is a unique id for outbox table. ID should be unique or storage
		will ignore all duplicate ids.
eventType - is a topic to which event will be published.
payload - the body to be published.

func (*Record) Done

func (r *Record) Done()

Done sets Done status to current Record. Status will be ignored on first save to the outbox table.

func (*Record) Fail

func (r *Record) Fail()

Fail sets Failed status to current Record. Status will be ignored on first save to the outbox table.

func (*Record) Null

func (r *Record) Null()

Null sets Null status to current Record.

type Status

type Status string

Status defines current status of Record.

const (
	// Progress means the current Record is processed by outbox worker.
	Progress Status = "progress"
	// Failed means the current Record not processed by worker by specific
	// reason.
	Failed Status = "failed"
	// Done means the current Record is successfully processed.
	Done Status = "done"
	// Null means the current Record is not processed yet.
	Null Status = ""
)