message

Documentation

Index

• Variables
• func HandlerNameFromCtx(ctx context.Context) string
• func PublishTopicFromCtx(ctx context.Context) string
• func PublisherNameFromCtx(ctx context.Context) string
• func SubscribeTopicFromCtx(ctx context.Context) string
• func SubscriberNameFromCtx(ctx context.Context) string
• type DuplicateHandlerNameError
  • func (d DuplicateHandlerNameError) Error() string
• type Handler
  • func (h *Handler) AddMiddleware(m ...HandlerMiddleware)
  • func (h *Handler) Started() chan struct{}
  • func (h *Handler) Stop()
  • func (h *Handler) Stopped() chan struct{}
• type HandlerFunc
• type HandlerMiddleware
• type Message
  • func NewMessage(uuid string, payload Payload) *Message
  • func (m *Message) Ack() bool
  • func (m *Message) Acked() <-chan struct{}
  • func (m *Message) Context() context.Context
  • func (m *Message) Copy() *Message
  • func (m *Message) Equals(toCompare *Message) bool
  • func (m *Message) Nack() bool
  • func (m *Message) Nacked() <-chan struct{}
  • func (m *Message) SetContext(ctx context.Context)
• type Messages
  • func (m Messages) IDs() []string
• type Metadata
  • func (m Metadata) Get(key string) string
  • func (m Metadata) Set(key, value string)
• type NoPublishHandlerFunc
• type Payload
• type Publisher
• type PublisherDecorator
  • func MessageTransformPublisherDecorator(transform func(*Message)) PublisherDecorator
• type Router
  • func NewDefaultRouter(logger watermill.LoggerAdapter) *Router
  • func NewRouter(config RouterConfig, logger watermill.LoggerAdapter) (*Router, error)
  • func (r *Router) AddHandler(handlerName string, subscribeTopic string, subscriber Subscriber, ...) *Handler
  • func (r *Router) AddMiddleware(m ...HandlerMiddleware)
  • func (r *Router) AddNoPublisherHandler(handlerName string, subscribeTopic string, subscriber Subscriber, ...) *Handler
  • func (r *Router) AddPlugin(p ...RouterPlugin)
  • func (r *Router) AddPublisherDecorators(dec ...PublisherDecorator)
  • func (r *Router) AddSubscriberDecorators(dec ...SubscriberDecorator)
  • func (r *Router) Close() error
  • func (r *Router) Handlers() map[string]HandlerFunc
  • func (r *Router) IsClosed() bool
  • func (r *Router) IsRunning() bool
  • func (r *Router) Logger() watermill.LoggerAdapter
  • func (r *Router) Run(ctx context.Context) (err error)
  • func (r *Router) RunHandlers(ctx context.Context) error
  • func (r *Router) Running() chan struct{}
• type RouterConfig
  • func (c RouterConfig) Validate() error
• type RouterPlugin
• type SubscribeInitializer
• type Subscriber
• type SubscriberDecorator
  • func MessageTransformSubscriberDecorator(transform func(*Message)) SubscriberDecorator

Variables

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")
)

Functions

func HandlerNameFromCtx

func HandlerNameFromCtx(ctx context.Context) string

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

func PublishTopicFromCtx

func PublishTopicFromCtx(ctx context.Context) string

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

func PublisherNameFromCtx

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

func SubscribeTopicFromCtx(ctx context.Context) string

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

func SubscriberNameFromCtx

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`.

Types

type DuplicateHandlerNameError

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

func (d DuplicateHandlerNameError) Error() string

type Handler

type Handler struct {
	// contains filtered or unexported fields
}

Handler handles Messages.

func (*Handler) AddMiddleware

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.

func (*Handler) Started

func (h *Handler) Started() chan struct{}

Started returns channel which is stopped when handler is running.

func (*Handler) Stop

func (h *Handler) Stop()

Stop stops the handler. Stop is asynchronous. You can check if handler was stopped with Stopped() function.

func (*Handler) Stopped

func (h *Handler) Stopped() chan struct{}

Stopped returns channel which is stopped when handler did stop.

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.

Example:

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 a unique identifier of the 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 unmarshalling the entire payload.
	// It is something similar to HTTP request's headers.
	//
	// Metadata is marshaled and will be saved to the PubSub.
	Metadata Metadata

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

Message is the basic transfer unit. Messages are emitted by Publishers and received by Subscribers.

A publisher can modify the message during publishing, e.g. can alter the metadata. Avoid modifying the message in parallel with publishing, as it can lead to data races. In general, a message should be passed to a single Publish and then considered immutable. If needed, use the Copy method to create a new message.

func NewMessage

func NewMessage(uuid string, payload Payload) *Message

NewMessage creates a new Message with given uuid and payload.

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.

Usage:

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

func (*Message) Context

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

func (m *Message) Copy() *Message

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

func (*Message) Equals

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.

Usage:

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

func (*Message) SetContext

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

SetContext sets provided context to the message.

type Messages

type Messages []*Message

Messages is a slice of messages.

func (Messages) IDs

func (m Messages) IDs() []string

IDs returns a slice of Messages' IDs.

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

Get returns the metadata value for the given key. If the key is not found, an empty string is returned.

func (Metadata) Set

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

Set sets the metadata key to value.

type NoPublishHandlerFunc

type NoPublishHandlerFunc func(msg *Message) error

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

type Payload

type Payload []byte

Payload is the Message's payload.

type Publisher

type Publisher interface {
	// Publish publishes provided messages to the given topic.
	//
	// Publish can be synchronous or asynchronous - it depends on the implementation.
	//
	// Most publisher 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 does not work with a single Context.
	// Use the Context() method of each message instead.
	//
	// Publish must be thread safe.
	Publish(topic string, messages ...*Message) error
	// Close should flush unsent messages if publisher is async.
	Close() error
}

Publisher is the emitting part of a Pub/Sub.

type PublisherDecorator

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

PublisherDecorator wraps the underlying Publisher, adding some functionality.

func MessageTransformPublisherDecorator

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 NewDefaultRouter

func NewDefaultRouter(logger watermill.LoggerAdapter) *Router

NewDefaultRouter creates a new Router with default configuration.

func NewRouter

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

NewRouter creates a new Router with given configuration.

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.

If handler is added while router is already running, you need to explicitly call RunHandlers().

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,
) *Handler

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.

If handler is added while router is already running, you need to explicitly call RunHandlers().

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

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

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) Handlers

func (r *Router) Handlers() map[string]HandlerFunc

Handlers returns all registered handlers.

func (*Router) IsClosed

func (r *Router) IsClosed() bool

func (*Router) IsRunning

func (r *Router) IsRunning() bool

IsRunning returns true when router is running.

Warning: for historical reasons, this method is not aware of router closing. If you want to know if the router was closed, use IsClosed.

func (*Router) Logger

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

Logger returns the Router's logger.

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) RunHandlers

func (r *Router) RunHandlers(ctx context.Context) error

RunHandlers runs all handlers that were added after Run(). RunHandlers is idempotent, so can be called multiple times safely.

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")

Warning: for historical reasons, this channel is not aware of router closing - the channel will be closed if the router has been running and closed.

type RouterConfig

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

RouterConfig holds the Router's configuration options.

func (RouterConfig) Validate

func (c RouterConfig) Validate() error

Validate returns Router configuration error, if any.

type RouterPlugin

type RouterPlugin func(*Router) error

RouterPlugin is function which is executed on Router start.

type SubscribeInitializer

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 initialization, 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
}

SubscribeInitializer is used to initialize subscribers.

type Subscriber

type Subscriber interface {
	// Subscribe returns an output channel with messages from the provided topic.
	// The channel is closed after Close() is called on the subscriber.
	//
	// To receive the next message, `Ack()` must be called on the received message.
	// If message processing fails and the message should be redelivered `Nack()` should be called instead.
	//
	// When the provided ctx is canceled, the subscriber closes the subscription and the output channel.
	// The provided ctx is passed to all produced messages.
	// When Nack or Ack is called on the message, the context of the message is canceled.
	Subscribe(ctx context.Context, topic string) (<-chan *Message, error)
	// Close closes all subscriptions with their output channels and flushes offsets etc. when needed.
	Close() error
}

Subscriber is the consuming part of the Pub/Sub.

type SubscriberDecorator

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

SubscriberDecorator wraps the underlying Subscriber, adding some functionality.

func MessageTransformSubscriberDecorator

func MessageTransformSubscriberDecorator(transform func(*Message)) SubscriberDecorator

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