gorabbit

GoRabbit

GoRabbit provides additional capabilities for official RabbitMQ Go client amqp091-go.

Connection re-dialing

Add re-dialing capabilities when connection got closed, you can reliably open new channel:

conn, _ := connection.Dial("amqp://localhost:5672")
ch, _ := conn.Channel() // use as regular connection, but it will re-dial if connection is closed 
ch.Publish(...)

See full example at examples/connection.

Channel re-opening

Add channel re-opening capabilities on the top of a plain channel, so you can get reliably consume or publish, even is something bad happened over the network.

conn, _ := connection.Dial("amqp://localhost:5672")
ch, _ := channel.New(conn) // add re-opening capabilities
// get constant flow of deliveries, channel will be re-opened if something goes south
deliveries := ch.Consume("example-queue", "", false, false, false, false, nil)
for d := range deliveries {
    // process deliveries
}
// or publish reliably
ch.PublishWithContext(ctx, "example-exchange", "", false, false, amqp.Publishing{})

See full example at examples/channel.

Consumers

GoRabbit provides consumer, which simplifies creation of RabbitMQ consumer. It aims to provide sane defaults, but at the same time is fully configurable:

conn, _ := connection.Dial("amqp://localhost:5672")
ch, _ := channel.New(conn) 
consumer := gorabbit.NewConsumer(channel, "example-queue")
consumer.Start(context.Background(), gorabbit.ProcessorFunc(func(ctx context.Context, deliveries <-chan amqp.Delivery) error {
    for d := range deliveries {
        // process deliveries
    }
    return nil
}))

See full example at examples/consumer.

To simplify it even more, GoRabbit is providing process building blocks: ByOne and InBatches. All you need is to write a function for handling received messages: if it returns error, processor will automatically NACK message, ACK on success.

One

One consumer reads messages one-by-one and passes them to Transaction. Will ACK them on success, NACK on error.

Example of usage is in examples/one.

One Middlewares

Easily plug-in any middlewares and pass them to process.ByOne:

process.ByOne(myHandler, true, middleware.NewErrorLogging(zerolog.New(os.Stderr))) 

Or implement your own with simple API:

type Middleware func(DeliveryHandler) DeliveryHandler

Check out more examples at process/middleware/one.go.

Batch

If you want to process messages in batches, Batch processor is here for you.

Batch processor reads a batch of messages from a broker and passes them to BatchTransaction. Your code expected to return one-to-one errors for each passed message (len(messages) == len(errors)).

Consumer will ACK each of messages on success, NACK on error.

Example of usage is in examples/batch.

Batch Middlewares

Plug in any middlewares and pass them to the Batch consumer, implementing simple API:

process.InBatches(myHandler, true, middleware.NewBatchErrorLogging(zerolog.New(os.Stderr))) 

Or implement your own with simple API:

type BatchMiddleware func(BatchDeliveryHandler) BatchDeliveryHandler

Check out more examples at process/middleware/batch.go.

Testing

GoRabbit comes with testing helpers in a form of stretchr/testify/suite testing suite, with prepared RabbitMQ connection, channel or topology.

type TestSuite struct {
  rabbittest.ChannelSuite
}

func (s *TestSuite) TestMyFunction() {
  s.Channel.Publish(...)
}

Set up RABBITMQ_URL environment variable and tun tests:

RABBITMQ_URL=amqp://localhost:5672 go test -v  ./...