outbox

Documentation

Index

• Constants
• Variables
• type Broker
• type Client
• type Logger
• type Option
  • func EnableDebugMode() Option
  • func WithIterationRate(dur time.Duration) Option
  • func WithLogger(logger Logger) Option
• type Outbox
  • func NewOutbox(broker Broker, conn SQLConn, 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()
• type SQLConn
• 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
	// 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, SQLConn, *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 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.

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 SQLConn, 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.

type SQLConn

type SQLConn interface {
	ExecContext(ctx context.Context, sql string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, sql string, args ...any) (*sql.Rows, error)
}

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"
)