Implementing custom Pub/Sub

Bring Your Own Pub/Sub.

The Pub/Sub interface

To add support for a custom Pub/Sub, you have to implement both message.Publisher and message.Subscriber interfaces.

Full source: github.com/ThreeDotsLabs/watermill/message/pubsub.go

// ...
type Publisher interface {
    // Publish publishes provided messages to given topic.
   //
   // Publish can be synchronous or asynchronous - it depends on the implementation.
   //
   // Most publishers implementations don't support atomic publishing of messages.
   // This means that if publishing one of the messages fails, the next messages will not be published.
   //
   // Publish must be thread safe.
   Publish(topic string, messages ...*Message) error
    // Close should flush unsent messages, if publisher is async.
   Close() error
}

type Subscriber interface {
    // Subscribe returns output channel with messages from provided topic.
   // Channel is closed, when Close() was called on the subscriber.
   //
   // To receive the next message, `Ack()` must be called on the received message.
   // If message processing failed and message should be redelivered `Nack()` should be called.
   //
   // When provided ctx is cancelled, subscriber will close subscribe and close output channel.
   // Provided ctx is set to all produced messages.
   // When Nack or Ack is called on the message, context of the message is canceled.
   Subscribe(ctx context.Context, topic string) (<-chan *Message, error)
    // Close closes all subscriptions with their output channels and flush offsets etc. when needed.
   Close() error
}

type SubscribeInitializer interface {
// ...

TODO list

Here are a few things you shouldn’t forget about:

  1. Logging (good messages and proper levels).
  2. Replaceable and configurable messages marshaller.
  3. Close() implementation for the publisher and subscriber that is:
    • idempotent
    • working correctly even when the publisher or the subscriber is blocked (for example, waiting for an Ack).
    • working correctly when the subscriber output channel is blocked (because nothing is listening on it).
  4. Ack() and Nack() support for consumed messages.
  5. Redelivery on Nack() for a consumed message.
  6. Use Universal Pub/Sub tests
  7. Performance optimizations.
  8. GoDocs, Markdown docs and Getting Started examples.

We will also be thankful for submitting a pull requests with the new Pub/Sub implementation.

If anything is not clear, feel free to use any of our support channels to reach us, we will be glad to help.