broker

Documentation

Overview

Package broker is the micro broker

Index

• func Publish(topic string, m *Message) error
• type Broker
• type ErrorHandler
• type Handler
• type Message
• type Option
  • func Addrs(addrs ...string) Option
  • func Codec(c codec.Marshaler) Option
  • func Registry(r registry.Registry) Option
  • func Secure(b bool) Option
  • func TLSConfig(t *tls.Config) Option
• type Options
• type PublishOption
  • func PublishContext(ctx context.Context) PublishOption
• type PublishOptions
• type SubscribeOption
  • func HandleError(h ErrorHandler) SubscribeOption
  • func Queue(name string) SubscribeOption
  • func SubscribeContext(ctx context.Context) SubscribeOption
• type SubscribeOptions
  • func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions
• type Subscriber
  • func Subscribe(topic string, h Handler) (Subscriber, error)

Functions

func Publish

func Publish(topic string, m *Message) error

Publish a message to a topic

Types

type Broker

type Broker interface {
	Init(...Option) error
	Options() Options
	Address() string
	Connect() error
	Disconnect() error
	Publish(topic string, m *Message, opts ...PublishOption) error
	Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
	String() string
}

Broker is an interface used for asynchronous messaging.

var (
	// DefaultBroker implementation
	DefaultBroker Broker
)

type ErrorHandler

type ErrorHandler func(*Message, error)

type Handler

type Handler func(*Message) error

Handler is used to process messages via a subscription of a topic.

type Message

type Message struct {
	Header map[string]string
	Body   []byte
}

type Option

type Option func(*Options)

func Addrs

func Addrs(addrs ...string) Option

Addrs sets the host addresses to be used by the broker

func Codec

func Codec(c codec.Marshaler) Option

Codec sets the codec used for encoding/decoding used where a broker does not support headers

func Registry

func Registry(r registry.Registry) Option

func Secure

func Secure(b bool) Option

Secure communication with the broker

func TLSConfig

func TLSConfig(t *tls.Config) Option

Specify TLS Config

type Options

type Options struct {
	Addrs  []string
	Secure bool
	Codec  codec.Marshaler

	TLSConfig *tls.Config
	// Registry used for clustering
	Registry registry.Registry
	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type PublishOption

type PublishOption func(*PublishOptions)

func PublishContext

func PublishContext(ctx context.Context) PublishOption

PublishContext set context

type PublishOptions

type PublishOptions struct {
	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

func HandleError

func HandleError(h ErrorHandler) SubscribeOption

ErrorHandler will catch all broker errors that cant be handled in normal way, for example Codec errors

func Queue

func Queue(name string) SubscribeOption

Queue sets the name of the queue to share messages on

func SubscribeContext

func SubscribeContext(ctx context.Context) SubscribeOption

SubscribeContext set context

type SubscribeOptions

type SubscribeOptions struct {
	// Handler executed when errors occur processing messages
	ErrorHandler ErrorHandler

	// Subscribers with the same queue name
	// will create a shared subscription where each
	// receives a subset of messages.
	Queue string

	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

func NewSubscribeOptions

func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions

type Subscriber

type Subscriber interface {
	Options() SubscribeOptions
	Topic() string
	Unsubscribe() error
}

Subscriber is a convenience return type for the Subscribe method

func Subscribe

func Subscribe(topic string, h Handler) (Subscriber, error)

Subscribe to a topic