Version: v1.1.1 Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: Feb 19, 2020 License: MIT Imports: 10 Imported by: 256




This section is empty.


View Source
var (
	// ErrOutputInNoPublisherHandler happens when a handler func returned some messages in a no-publisher handler.
	// todo: maybe change the handler func signature in no-publisher handler so that there's no possibility for this
	ErrOutputInNoPublisherHandler = errors.New("returned output messages in a handler without publisher")


func HandlerNameFromCtx added in v0.3.0

func HandlerNameFromCtx(ctx context.Context) string

HandlerNameFromCtx returns the name of the message handler in the router that consumed the message.

func PublishTopicFromCtx added in v1.0.3

func PublishTopicFromCtx(ctx context.Context) string

PublishTopicFromCtx returns the topic to which message will be published by the router.

func PublisherNameFromCtx added in v0.3.0

func PublisherNameFromCtx(ctx context.Context) string

PublisherNameFromCtx returns the name of the message publisher type that published the message in the router. For example, for Kafka it will be `kafka.Publisher`.

func SubscribeTopicFromCtx added in v1.0.3

func SubscribeTopicFromCtx(ctx context.Context) string

SubscribeTopicFromCtx returns the topic from which message was received in the router.

func SubscriberNameFromCtx added in v0.3.0

func SubscriberNameFromCtx(ctx context.Context) string

SubscriberNameFromCtx returns the name of the message subscriber type that subscribed to the message in the router. For example, for Kafka it will be `kafka.Subscriber`.


type DuplicateHandlerNameError added in v0.3.0

type DuplicateHandlerNameError struct {
	HandlerName string

DuplicateHandlerNameError is sent in a panic when you try to add a second handler with the same name.

func (DuplicateHandlerNameError) Error added in v0.3.0

type Handler added in v1.0.1

type Handler struct {
	// contains filtered or unexported fields

func (*Handler) AddMiddleware added in v1.0.1

func (h *Handler) AddMiddleware(m ...HandlerMiddleware)

AddMiddleware adds new middleware to the specified handler in the router.

The order of middleware matters. Middleware added at the beginning is executed first.

type HandlerFunc

type HandlerFunc func(msg *Message) ([]*Message, error)

HandlerFunc is function called when message is received.

msg.Ack() is called automatically when HandlerFunc doesn't return error. When HandlerFunc returns error, msg.Nack() is called. When msg.Ack() was called in handler and HandlerFunc returns error, msg.Nack() will be not sent because Ack was already sent.

HandlerFunc's are executed parallel when multiple messages was received (because msg.Ack() was sent in HandlerFunc or Subscriber supports multiple consumers).

var PassthroughHandler HandlerFunc = func(msg *Message) ([]*Message, error) {
	return []*Message{msg}, nil

PassthroughHandler is a handler that passes the message unchanged from the subscriber to the publisher.

type HandlerMiddleware

type HandlerMiddleware func(h HandlerFunc) HandlerFunc

HandlerMiddleware allows us to write something like decorators to HandlerFunc. It can execute something before handler (for example: modify consumed message) or after (modify produced messages, ack/nack on consumed message, handle errors, logging, etc.).

It can be attached to the router by using `AddMiddleware` method.


func ExampleMiddleware(h message.HandlerFunc) message.HandlerFunc {
	return func(message *message.Message) ([]*message.Message, error) {
		fmt.Println("executed before handler")
		producedMessages, err := h(message)
		fmt.Println("executed after handler")

		return producedMessages, err

type Message

type Message struct {
	// UUID is an unique identifier of message.
	// It is only used by Watermill for debugging.
	// UUID can be empty.
	UUID string

	// Metadata contains the message metadata.
	// Can be used to store data which doesn't require unmarshaling entire payload.
	// It is something similar to HTTP request's headers.
	// Metadata is marshaled and will be saved to PubSub.
	Metadata Metadata

	// Payload is message's payload.
	Payload Payload
	// contains filtered or unexported fields

func NewMessage

func NewMessage(uuid string, payload Payload) *Message

func (*Message) Ack

func (m *Message) Ack() bool

Ack sends message's acknowledgement.

Ack is not blocking. Ack is idempotent. False is returned, if Nack is already sent.

func (*Message) Acked

func (m *Message) Acked() <-chan struct{}

Acked returns channel which is closed when acknowledgement is sent.


select {
case <-message.Acked():
	// ack received
case <-message.Nacked():
	// nack received

func (*Message) Context added in v0.2.0

func (m *Message) Context() context.Context

Context returns the message's context. To change the context, use SetContext.

The returned context is always non-nil; it defaults to the background context.

func (*Message) Copy added in v0.2.0

func (m *Message) Copy() *Message

Copy copies all message without Acks/Nacks. The context is not propagated to the copy.

func (*Message) Equals added in v0.2.0

func (m *Message) Equals(toCompare *Message) bool

Equals compare, that two messages are equal. Acks/Nacks are not compared.

func (*Message) Nack

func (m *Message) Nack() bool

Nack sends message's negative acknowledgement.

Nack is not blocking. Nack is idempotent. False is returned, if Ack is already sent.

func (*Message) Nacked

func (m *Message) Nacked() <-chan struct{}

Nacked returns channel which is closed when negative acknowledgement is sent.


select {
case <-message.Acked():
	// ack received
case <-message.Nacked():
	// nack received

func (*Message) SetContext added in v0.2.0

func (m *Message) SetContext(ctx context.Context)

SetContext sets provided context to the message.

type Messages

type Messages []*Message

func (Messages) IDs

func (m Messages) IDs() []string

type Metadata

type Metadata map[string]string

Metadata is sent with every message to provide extra context without unmarshaling the message payload.

func (Metadata) Get

func (m Metadata) Get(key string) string

func (Metadata) Set

func (m Metadata) Set(key, value string)

type NoPublishHandlerFunc added in v1.0.0

type NoPublishHandlerFunc func(msg *Message) error

NoPublishHandlerFunc is HandlerFunc alternative, which doesn't produce any messages.

type Payload

type Payload []byte

type Publisher

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 PublisherDecorator added in v0.3.0

type PublisherDecorator func(pub Publisher) (Publisher, error)

PublisherDecorator wraps the underlying Publisher, adding some functionality.

func MessageTransformPublisherDecorator added in v0.3.0

func MessageTransformPublisherDecorator(transform func(*Message)) PublisherDecorator

MessageTransformPublisherDecorator creates a publisher decorator that calls transform on each message that passes through the publisher.

type Router

type Router struct {
	// contains filtered or unexported fields

Router is responsible for handling messages from subscribers using provided handler functions.

If the handler function returns a message, the message is published with the publisher. You can use middlewares to wrap handlers with common logic like logging, instrumentation, etc.

func NewRouter

func NewRouter(config RouterConfig, logger watermill.LoggerAdapter) (*Router, error)

func (*Router) AddHandler

func (r *Router) AddHandler(
	handlerName string,
	subscribeTopic string,
	subscriber Subscriber,
	publishTopic string,
	publisher Publisher,
	handlerFunc HandlerFunc,
) *Handler

AddHandler adds a new handler.

handlerName must be unique. For now, it is used only for debugging.

subscribeTopic is a topic from which handler will receive messages.

publishTopic is a topic to which router will produce messages returned by handlerFunc. When handler needs to publish to multiple topics, it is recommended to just inject Publisher to Handler or implement middleware which will catch messages and publish to topic based on metadata for example.

func (*Router) AddMiddleware

func (r *Router) AddMiddleware(m ...HandlerMiddleware)

AddMiddleware adds a new middleware to the router.

The order of middleware matters. Middleware added at the beginning is executed first.

func (*Router) AddNoPublisherHandler

func (r *Router) AddNoPublisherHandler(
	handlerName string,
	subscribeTopic string,
	subscriber Subscriber,
	handlerFunc NoPublishHandlerFunc,

AddNoPublisherHandler adds a new handler. This handler cannot return messages. When message is returned it will occur an error and Nack will be sent.

handlerName must be unique. For now, it is used only for debugging.

subscribeTopic is a topic from which handler will receive messages.

subscriber is Subscriber from which messages will be consumed.

func (*Router) AddPlugin

func (r *Router) AddPlugin(p ...RouterPlugin)

AddPlugin adds a new plugin to the router. Plugins are executed during startup of the router.

A plugin can, for example, close the router after SIGINT or SIGTERM is sent to the process (SignalsHandler plugin).

func (*Router) AddPublisherDecorators added in v0.3.0

func (r *Router) AddPublisherDecorators(dec ...PublisherDecorator)

AddPublisherDecorators wraps the router's Publisher. The first decorator is the innermost, i.e. calls the original publisher.

func (*Router) AddSubscriberDecorators added in v0.3.0

func (r *Router) AddSubscriberDecorators(dec ...SubscriberDecorator)

AddSubscriberDecorators wraps the router's Subscriber. The first decorator is the innermost, i.e. calls the original subscriber.

func (*Router) Close

func (r *Router) Close() error

Close gracefully closes the router with a timeout provided in the configuration.

func (*Router) Logger

func (r *Router) Logger() watermill.LoggerAdapter

func (*Router) Run

func (r *Router) Run(ctx context.Context) (err error)

Run runs all plugins and handlers and starts subscribing to provided topics. This call is blocking while the router is running.

When all handlers have stopped (for example, because subscriptions were closed), the router will also stop.

To stop Run() you should call Close() on the router.

ctx will be propagated to all subscribers.

When all handlers are stopped (for example: because of closed connection), Run() will be also stopped.

func (*Router) Running

func (r *Router) Running() chan struct{}

Running is closed when router is running. In other words: you can wait till router is running using

fmt.Println("Starting router")
go r.Run(ctx)
<- r.Running()
fmt.Println("Router is running")

type RouterConfig

type RouterConfig struct {
	// CloseTimeout determines how long router should work for handlers when closing.
	CloseTimeout time.Duration

func (RouterConfig) Validate

func (c RouterConfig) Validate() error

type RouterPlugin

type RouterPlugin func(*Router) error

RouterPlugin is function which is executed on Router start.

type SubscribeInitializer added in v0.3.0

type SubscribeInitializer interface {
	// SubscribeInitialize can be called to initialize subscribe before consume.
	// When calling Subscribe before Publish, SubscribeInitialize should be not required.
	// Not every Pub/Sub requires this initialize and it may be optional for performance improvements etc.
	// For detailed SubscribeInitialize functionality, please check Pub/Subs godoc.
	// Implementing SubscribeInitialize is not obligatory.
	SubscribeInitialize(topic string) error

type Subscriber

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 SubscriberDecorator added in v0.3.0

type SubscriberDecorator func(sub Subscriber) (Subscriber, error)

SubscriberDecorator wraps the underlying Subscriber, adding some functionality.

func MessageTransformSubscriberDecorator added in v0.3.0

func MessageTransformSubscriberDecorator(transform func(*Message)) SubscriberDecorator

MessageTransformSubscriberDecorator creates a subscriber decorator that calls transform on each message that passes through the subscriber.


Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL