Outboxer

Outboxer is a go library that implements the outbox pattern.

Getting Started

Outboxer simplifies the challenging work of orchestrating message reliability. Essentially we are trying to solve this question:

How can producers reliably send messages when the broker/consumer is unavailable?

If you have a distributed system architecture and especially mainly deal with Event Driven Architecture, you might want to use outboxer.

The first thing to do is include the package in your project.

go get github.com/italolelis/outboxer

Usage

Let's setup a simple example where you are using Google's PubSub and Postgres as your outbox pattern components:

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

db, err := sql.Open("postgres", os.Getenv("DS_DSN"))
if err != nil {
    fmt.Printf("could not connect to amqp: %s", err)
    return
}

// we need to create a data store instance first
ds, err := postgres.WithInstance(ctx, db)
if err != nil {
    fmt.Printf("could not setup the data store: %s", err)
    return
}
defer ds.Close()

// we create an event stream passing the pusub connection
client, err := pubsub.NewClient(ctx, os.Getenv("GCP_PROJECT_ID"))
if err != nil {
    fmt.Printf("failed to connect to gcp: %s", err)
    return
}

es := pubsubOut.New(client)

// now we create an outboxer instance passing the data store and event stream
o, err := outboxer.New(
    outboxer.WithDataStore(ds),
    outboxer.WithEventStream(es),
    outboxer.WithCheckInterval(1*time.Second),
    outboxer.WithCleanupInterval(5*time.Second),
    outboxer.WithCleanUpBefore(time.Now().AddDate(0, 0, -5)),
)
if err != nil {
    fmt.Printf("could not create an outboxer instance: %s", err)
    return
}

// here we initialize the outboxer checks and cleanup go rotines
o.Start(ctx)
defer o.Stop()

// finally we are ready to send messages
if err = o.Send(ctx, &outboxer.OutboxMessage{
    Payload: []byte("test payload"),
    Options: map[string]interface{}{
        amqpOut.ExchangeNameOption: "test",
        amqpOut.ExchangeTypeOption: "topic",
        amqpOut.RoutingKeyOption:   "test.send",
    },
}); err != nil {
    fmt.Printf("could not send message: %s", err)
    return
}

// we can also listen for errors and ok messages that were send
for {
    select {
    case err := <-o.ErrChan():
        fmt.Printf("could not send message: %s", err)
    case <-o.OkChan():
        fmt.Printf("message received")
        return
    }
}

Features

Outboxer comes with a few implementations of Data Stores and Event Streams.

Data Stores

• Postgres DataStore
• MySQL DataStore
• SQLServer DataStore

Event Streams

• AMQP EventStream
• Kinesis EventStream
• SQS EventStream
• GCP PubSub

Contributing

Please read CONTRIBUTING.md for details on our code of conduct and the process for submitting pull requests to us.

License

This project is licensed under the MIT License - see the LICENSE file for details