outbox

README

Outbox

[!IMPORTANT] This submodule is couple with github.com/jackc/pgx/v5 and therefore works for Postgres databases only. This is the reason why this is shipped in a submodule instead of being included in the base module.

This submodule implements the outbox pattern on top of the Bunnify publisher.

How does it work

Whenever you call the Publish function the event is stored in a database table created only if not exists on the creation of the Publisher. This function takes a transaction, which makes your entities + creation of the event atomic.

There is a go routine that every certain duration will try to fetch the outbox events pending for publishing if any. Each one of them will be published in the same way that the bunnify Publisher does. All published events will be marked as published in the database table (or deleted if configured).

Installation

go get github.com/pmorelli92/bunnify/outbox

Examples

Setup Publisher

https://github.com/pmorelli92/bunnify/blob/985913450b7b9a21219b96f23843288bef7eac74/outbox/tests/consumer_publish_test.go#L87-L104

Publish

https://github.com/pmorelli92/bunnify/blob/985913450b7b9a21219b96f23843288bef7eac74/outbox/tests/consumer_publish_test.go#L107-L127

Configuration

https://github.com/pmorelli92/bunnify/blob/985913450b7b9a21219b96f23843288bef7eac74/outbox/publisherOption.go#L17-L38

Documentation

Index

• func WithDeleteAfterPublish() func(*publisherOption)
• func WithLoopingInterval(interval time.Duration) func(*publisherOption)
• func WithNoficationChannel(notificationCh chan<- bunnify.Notification) func(*publisherOption)
• type Publisher
  • func NewPublisher(ctx context.Context, db *pgxpool.Pool, inner bunnify.Publisher, ...) (*Publisher, error)
  • func (p *Publisher) Close()
  • func (p *Publisher) Publish(ctx context.Context, tx pgx.Tx, exchange string, routingKey string, ...) error

Functions

func WithDeleteAfterPublish

func WithDeleteAfterPublish() func(*publisherOption)

WithDeleteAfterPublish specifies that a published event from the outbox will be deleted instead of marked as published as it is not interesting to have the event as a timelog history.

func WithLoopingInterval

func WithLoopingInterval(interval time.Duration) func(*publisherOption)

WithLoopingInterval specifies the interval on which the loop to check the pending to publish events is executed

func WithNoficationChannel

func WithNoficationChannel(notificationCh chan<- bunnify.Notification) func(*publisherOption)

WithNoficationChannel specifies a go channel to receive messages such as connection established, reconnecting, event published, consumed, etc.

Types

type Publisher

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

Publisher is used for publishing events.

func NewPublisher

func NewPublisher(
	ctx context.Context,
	db *pgxpool.Pool,
	inner bunnify.Publisher,
	opts ...func(*publisherOption)) (*Publisher, error)

NewPublisher creates a publisher using a database connection and acts as a wrapper for bunnify publisher.

func (*Publisher) Close

func (p *Publisher) Close()

func (*Publisher) Publish

func (p *Publisher) Publish(
	ctx context.Context,
	tx pgx.Tx,
	exchange string,
	routingKey string,
	event bunnify.PublishableEvent) error

Publish publishes an event to the outbox database table. Then the loop will pick this up and try to use the inner bunnify Publisher do an AQMP publish.