Middlewares

Add functionality to handlers.

Introduction

Middlewares wrap handlers with functionality that is important, but not relevant for the primary handler’s logic.Examples include retrying the handler after an error was returned, or recovering from panic in the handlerand capturing the stacktrace.

Middlewares wrap the handler function like this:

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

  1. // ...
  2. // HandlerMiddleware allows us to write something like decorators to HandlerFunc.
  3. // It can execute something before handler (for example: modify consumed message)
  4. // or after (modify produced messages, ack/nack on consumed message, handle errors, logging, etc.).
  5. //
  6. // It can be attached to the router by using `AddMiddleware` method.
  7. //
  8. // Example:
  9. //        func ExampleMiddleware(h message.HandlerFunc) message.HandlerFunc {
  10. //            return func(message *message.Message) ([]*message.Message, error) {
  11. //                fmt.Println("executed before handler")
  12. //                producedMessages, err := h(message)
  13. //                fmt.Println("executed after handler")
  14. //
  15. //                return producedMessages, err
  16. //            }
  17. //        }
  18. type HandlerMiddleware func(h HandlerFunc) HandlerFunc
  19. // ...

Available middlewares

Below are the middlewares provided by Watermill and ready to use. You can also easily implement your own.For example, if you’d like to store every received message in some kind of log, it’s the best way to do it.

Randomfail

  1. // RandomFail makes the handler fail with an error based on random chance. Error probability should be in the range (0,1).
  2. func RandomFail(errorProbability float32) message.HandlerMiddleware {
  3.     return func(h message.HandlerFunc) message.HandlerFunc {
  4.         return func(message *message.Message) ([]*message.Message, error) {
  5.             if shouldFail(errorProbability) {
  6.                 return nil, errors.New("random fail occurred")
  7.             }
  9.             return h(message)
  10.         }
  11.     }
  12. }
  1. // RandomPanic makes the handler panic based on random chance. Panic probability should be in the range (0,1).
  2. func RandomPanic(panicProbability float32) message.HandlerMiddleware {
  3.     return func(h message.HandlerFunc) message.HandlerFunc {
  4.         return func(message *message.Message) ([]*message.Message, error) {
  5.             if shouldFail(panicProbability) {
  6.                 panic("random panic occurred")
  7.             }
  9.             return h(message)
  10.         }
  11.     }
  12. }

Retry

  1. // Retry provides a middleware that retries the handler if errors are returned.
  2. // The retry behaviour is configurable, with exponential backoff and maximum elapsed time.
  3. type Retry struct {
  4.     // MaxRetries is maximum number of times a retry will be attempted.
  5.     MaxRetries int
  7.     // InitalInterval is the first interval between retries. Subsequent intervals will be scaled by Multiplier.
  8.     InitialInterval time.Duration
  9.     // MaxInterval sets the limit for the exponential backoff of retries. The interval will not be increased beyond MaxInterval.
  10.     MaxInterval time.Duration
  11.     // Multiplier is the factor by which the waiting interval will be multiplied between retries.
  12.     Multiplier float64
  13.     // MaxElapsedTime sets the time limit of how long retries will be attempted. Disabled if 0.
  14.     MaxElapsedTime time.Duration
  15.     // RandomizationFactor randomizes the spread of the backoff times within the interval of:
  16.     // [currentInterval * (1 - randomization_factor), currentInterval * (1 + randomization_factor)].
  17.     RandomizationFactor float64
  19.     // OnRetryHook is an optional function that will be executed on each retry attempt.
  20.     // The number of the current retry is passed as retryNum,
  21.     OnRetryHook func(retryNum int, delay time.Duration)
  23.     Logger watermill.LoggerAdapter
  24. }

Instant Ack

  1. // InstantAck makes the handler instantly acknowledge the incoming message, regardless of any errors.
  2. // It may be used to gain throughput, but at a cost:
  3. // If you had exactly-once delivery, you may expect at-least-once instead.
  4. // If you had ordered messages, the ordering might be broken.
  5. func InstantAck(h message.HandlerFunc) message.HandlerFunc {
  6.     return func(message *message.Message) ([]*message.Message, error) {
  7.         message.Ack()
  8.         return h(message)
  9.     }
  10. }

Poison

  1. // PoisonQueue provides a middleware that salvages unprocessable messages and published them on a separate topic.
  2. // The main middleware chain then continues on, business as usual.
  3. func PoisonQueue(pub message.Publisher, topic string) (message.HandlerMiddleware, error) {
  4.     if topic == "" {
  5.         return nil, ErrInvalidPoisonQueueTopic
  6.     }
  8.     pq := poisonQueue{
  9.         topic: topic,
  10.         pub:   pub,
  11.         shouldGoToPoisonQueue: func(err error) bool {
  12.             return true
  13.         },
  14.     }
  16.     return pq.Middleware, nil
  17. }
  1. // PoisonQueueWithFilter is just like PoisonQueue, but accepts a function that decides which errors qualify for the poison queue.
  2. func PoisonQueueWithFilter(pub message.Publisher, topic string, shouldGoToPoisonQueue func(err error) bool) (message.HandlerMiddleware, error) {
  3.     if topic == "" {
  4.         return nil, ErrInvalidPoisonQueueTopic
  5.     }
  7.     pq := poisonQueue{
  8.         topic: topic,
  9.         pub:   pub,
  11.         shouldGoToPoisonQueue: shouldGoToPoisonQueue,
  12.     }
  14.     return pq.Middleware, nil
  15. }

Ignore Errors

  1. // IgnoreErrors provides a middleware that makes the handler ignore some explicitly whitelisted errors.
  2. type IgnoreErrors struct {
  3.     ignoredErrors map[string]struct{}
  4. }
  1. // NewIgnoreErrors creates a new IgnoreErrors middleware.
  2. func NewIgnoreErrors(errs []error) IgnoreErrors {
  3.     errsMap := make(map[string]struct{}, len(errs))
  5.     for _, err := range errs {
  6.         errsMap[err.Error()] = struct{}{}
  7.     }
  9.     return IgnoreErrors{errsMap}
  10. }

Recoverer

  1. // Recoverer recovers from any panic in the handler and appends RecoveredPanicError with the stacktrace
  2. // to any error returned from the handler.
  3. func Recoverer(h message.HandlerFunc) message.HandlerFunc {
  4.     return func(event *message.Message) (events []*message.Message, err error) {
  5.         defer func() {
  6.             if r := recover(); r != nil {
  7.                 panicErr := errors.WithStack(RecoveredPanicError{V: r, Stacktrace: string(debug.Stack())})
  8.                 err = multierror.Append(err, panicErr)
  9.             }
  10.         }()
  12.         return h(event)
  13.     }
  14. }

Throttle

  1. // Throttle provides a middleware that limits the amount of messages processed per unit of time.
  2. // This may be done e.g. to prevent excessive load caused by running a handler on a long queue of unprocessed messages.
  3. type Throttle struct {
  4.     throttle <-chan time.Time
  5. }
  1. // NewThrottle creates a new Throttle middleware.
  2. // Example duration and count: NewThrottle(10, time.Second) for 10 messages per second
  3. func NewThrottle(count int64, duration time.Duration) *Throttle {
  4.     return &Throttle{time.Tick(duration / time.Duration(count))}
  5. }

Correlation

  1. // SetCorrelationID sets a correlation ID for the message.
  2. //
  3. // SetCorrelationID should be called when the message enters the system.
  4. // When message is produced in a request (for example HTTP),
  5. // message correlation ID should be the same as the request's correlation ID.
  6. func SetCorrelationID(id string, msg *message.Message) {
  7.     if MessageCorrelationID(msg) != "" {
  8.         return
  9.     }
  11.     msg.Metadata.Set(CorrelationIDMetadataKey, id)
  12. }
  1. // MessageCorrelationID returns correlation ID from the message.
  2. func MessageCorrelationID(message *message.Message) string {
  3.     return message.Metadata.Get(CorrelationIDMetadataKey)
  4. }
  1. // CorrelationID adds correlation ID to all messages produced by the handler.
  2. // ID is based on ID from message received by handler.
  3. //
  4. // To make CorrelationID working correctly, SetCorrelationID must be called to first message entering the system.
  5. func CorrelationID(h message.HandlerFunc) message.HandlerFunc {
  6.     return func(message *message.Message) ([]*message.Message, error) {
  7.         producedMessages, err := h(message)
  9.         correlationID := MessageCorrelationID(message)
  10.         for _, msg := range producedMessages {
  11.             SetCorrelationID(correlationID, msg)
  12.         }
  14.         return producedMessages, err
  15.     }
  16. }

Timeout

  1. // Timeout makes the handler cancel the incoming message's context after a specified time.
  2. // Any timeout-sensitive functionality of the handler should listen on msg.Context().Done() to know when to fail.
  3. func Timeout(timeout time.Duration) func(message.HandlerFunc) message.HandlerFunc {
  4.     return func(h message.HandlerFunc) message.HandlerFunc {
  5.         return func(msg *message.Message) ([]*message.Message, error) {
  6.             ctx, cancel := context.WithTimeout(msg.Context(), timeout)
  7.             defer func() {
  8.                 cancel()
  9.             }()
  11.             msg.SetContext(ctx)
  12.             return h(msg)
  13.         }
  14.     }
  15. }