broker

Documentation

Overview

Package broker is an interface used for asynchronous messaging

Package broker provides a http based message broker

Package memory provides a memory broker

Index

• Variables
• func Connect() error
• func Disconnect() error
• func Init(opts ...Option) error
• func Publish(topic string, msg *Message, opts ...PublishOption) error
• func String() string
• type Broker
  • func NewBroker(opts ...Option) Broker
  • func NewMemoryBroker(opts ...Option) Broker
  • func Provide(_ DiConfig, config *Config) (Broker, error)
• type BrokerConfig
• type Config
  • func NewConfig() *Config
  • func (d *Config) Merge(src *Config) error
• type DiConfig
  • func ProvideConfig(_ di.DiConfig, flags *DiFlags, config *Config, cliConfig *cli.Config, ...) (DiConfig, error)
  • func ProvideConfigNoFlags(config *Config, configor config.Config) (DiConfig, error)
• type DiFlags
  • func ProvideFlags(config *Config, cliConfig *cli.Config, c cli.Cli) (*DiFlags, error)
• type Event
• type Handler
• type Message
• type Option
  • func Addrs(addrs ...string) Option
  • func Codec(c codec.Marshaler) Option
  • func ErrorHandler(h Handler) Option
  • func Logger(l logger.Logger) Option
  • func Registry(r registry.Registry) Option
  • func Secure(b bool) Option
  • func TLSConfig(t *tls.Config) Option
• type Options
  • func NewOptions(opts ...Option) *Options
• type PublishOption
  • func PublishContext(ctx context.Context) PublishOption
• type PublishOptions
• type SubscribeOption
  • func DisableAutoAck() 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, handler Handler, opts ...SubscribeOption) (Subscriber, error)

Variables

var (
	DefaultPath    = "/"
	DefaultAddress = "127.0.0.1:0"
)

var (
	// DefaultBroker is the default Broker.
	DefaultBroker = NewBroker()
)

var DiNoCliSet = wire.NewSet(ProvideConfigNoFlags, Provide)

var DiSet = wire.NewSet(ProvideFlags, ProvideConfig, Provide)

var Plugins = generic.NewContainer(func(opts ...Option) Broker { return nil })

Functions

func Connect

func Connect() error

func Disconnect

func Disconnect() error

func Init

func Init(opts ...Option) error

func Publish

func Publish(topic string, msg *Message, opts ...PublishOption) error

func String

func String() string

String returns the name of the Broker.

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.

func NewBroker

func NewBroker(opts ...Option) Broker

NewBroker returns a new http broker.

func NewMemoryBroker

func NewMemoryBroker(opts ...Option) Broker

func Provide

func Provide(

	_ DiConfig,

	config *Config,
) (Broker, error)

type BrokerConfig

type BrokerConfig struct {
	Enabled   bool     `json:"enabled" yaml:"Enabled"`
	Plugin    string   `json:"plugin,omitempty" yaml:"Plugin,omitempty"`
	Addresses []string `json:"addresses,omitempty" yaml:"Addresses,omitempty"`
}

type Config

type Config struct {
	Broker BrokerConfig `json:"broker" yaml:"Broker"`
}

func NewConfig

func NewConfig() *Config

func (*Config) Merge

func (d *Config) Merge(src *Config) error

type DiConfig

type DiConfig struct{}

DiConfig is marker that DiFlags has been parsed into Config

func ProvideConfig

func ProvideConfig(
	_ di.DiConfig,
	flags *DiFlags,
	config *Config,
	cliConfig *cli.Config,
	configor config.Config,
) (DiConfig, error)

func ProvideConfigNoFlags

func ProvideConfigNoFlags(
	config *Config,
	configor config.Config,
) (DiConfig, error)

type DiFlags

type DiFlags struct {
	Plugin    string
	Addresses string
}

func ProvideFlags

func ProvideFlags(
	config *Config,
	cliConfig *cli.Config,
	c cli.Cli,
) (*DiFlags, error)

type Event

type Event interface {
	Topic() string
	Message() *Message
	Ack() error
	Error() error
}

Event is given to a subscription handler for processing.

type Handler

type Handler func(Event) error

Handler is used to process messages via a subscription of a topic. The handler is passed a publication interface which contains the message and optional Ack method to acknowledge receipt of the message.

type Message

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

Message is a message send/received from the broker.

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 ErrorHandler

func ErrorHandler(h Handler) Option

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

func Logger

func Logger(l logger.Logger) Option

Logger sets the underline logger.

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

	// Logger is the underlying logger
	Logger logger.Logger

	// Handler executed when error happens in broker mesage
	// processing
	ErrorHandler Handler

	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
}

func NewOptions

func NewOptions(opts ...Option) *Options

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 DisableAutoAck

func DisableAutoAck() SubscribeOption

DisableAutoAck will disable auto acking of messages after they have been handled.

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 {
	// AutoAck defaults to true. When a handler returns
	// with a nil error the message is acked.
	AutoAck bool
	// 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, handler Handler, opts ...SubscribeOption) (Subscriber, error)