Pub/Sub

Publishers and Subscribers.

Publisher

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

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

Publishing multiple messages

Most publishers implementations don’t support atomic publishing of messages.This means that if publishing one of the messages fails, the next messages won’t be published.

Async publish

Publish can be synchronous or asynchronous - it depends on the implementation.

Close()

Close should flush unsent messages if the publisher is asynchronous.It is important to not forget to close the subscriber. Otherwise you may lose some of the messages.

Subscriber

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

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

Ack/Nack mechanism

It is the Subscriber’s responsibility to handle an Ack and a Nack from a message.A proper implementation should wait for an Ack or a Nack before consuming the next message.

Important Subscriber’s implementation notice:Ack/offset to message’s storage/broker must be sent after Ack from Watermill’s message.Otherwise there is a chance to lose messages if the process dies before the messages have been processed.

Close()

Close closes all subscriptions with their output channels and flushes offsets, etc. when needed.

At-least-once delivery

Watermill is built with at-least-once delivery semantics.That means when some error occurs when processing a message and an Ack cannot be sent, the message will be redelivered.

You need to keep it in mind and build your application to be idempotent or implement a deduplication mechanism.

Unfortunately, it’s not possible to create an universal middleware for deduplication, so we encourage you to build your own.

Universal tests

Every Pub/Sub is similar in most aspects.To avoid implementing separate tests for every Pub/Sub, we’ve created a test suite which should be passed by any Pub/Subimplementation.

These tests can be found in pubsub/tests/test_pubsub.go.

Built-in implementations

To check available Pub/Sub implementations, see Supported Pub/Subs.

Implementing custom Pub/Sub

See Implementing custom Pub/Sub for instructions on how to introduce support fora new Pub/Sub.

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

You can also request a new Pub/Sub implementation by submitting a new issue.

Keep going!

Now that you already know how a Pub/Sub is working, we recommend learning about the Message Router component.