Skip to content
A library that implements the outboxer pattern in go
Branch: master
Clone or download
Latest commit 1c3aa2c Feb 21, 2019
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
amqp Added more tests Feb 11, 2019
build Added MySQL support Feb 15, 2019
lock Added MySQL support Feb 15, 2019
mysql Added MySQL support Feb 15, 2019
postgres Added MySQL support Feb 15, 2019
.gitignore Added go.sum Feb 2, 2019
.travis.yml Added MySQL support Feb 15, 2019
CONTRIBUTING.md Initial commit Feb 1, 2019
LICENSE
README.md Updated readme Feb 15, 2019
go.mod Added MySQL support Feb 15, 2019
go.sum Added MySQL support Feb 15, 2019
options.go Added batch sizes Feb 1, 2019
outbox_message.go Fixed tests for outbox message Feb 8, 2019
outbox_message_test.go Fixing spelling Feb 21, 2019
outboxer.go Fixed lint findings Feb 2, 2019
outboxer_example_test.go Added a few more checks Feb 2, 2019
outboxer_test.go Added MySQL support Feb 15, 2019

README.md

Outboxer

Build Status Coverage Status Go Report Card GoDoc

Outboxer is a go library that implements the outbox pattern.

Getting Started

Outboxer was designed to simplify the tough work of orchestrating message reliabilty. 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 is dealing 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

Initial Configuration

Let's setup a simple example where you are using RabbitMQ 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
}

conn, err := amqp.Dial(os.Getenv("ES_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 amqp connection
es := amqpOut.NewAMQP(conn)

// 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.

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

You can’t perform that action at this time.