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)
• 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 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, ...)
  • 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) Logger() watermill.LoggerAdapter
  • func (r *Router) Run(ctx context.Context) (err 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
}

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.

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

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

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

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

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

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

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

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.