broker

Documentation

Overview

Package broker is an interface used for asynchronous messaging

Index

• Variables
• func Connect() error
• func Disconnect() error
• func Init(opts ...Option) error
• func Publish(channel string, msg *Message, opts ...PublishOption) error
• func Request(channel string, msg *RequestInput, opts ...PublishOption) (interface{}, error)
• func String() string
• type ActionHandle
• type Broker
• type Event
• type Handler
• type Message
• type NatsBroker
  • func NewBroker(opts ...Option) *NatsBroker
  • func (n *NatsBroker) Address() string
  • func (n *NatsBroker) Connect() error
  • func (n *NatsBroker) Disconnect() error
  • func (n *NatsBroker) Init(opts ...Option) error
  • func (n *NatsBroker) Options() Options
  • func (n *NatsBroker) Publish(channel string, msg *Message, opts ...PublishOption) error
  • func (n *NatsBroker) Request(channel string, msg *RequestInput, opts ...PublishOption) (interface{}, error)
  • func (n *NatsBroker) Respond(channel string, handler ActionHandle, opts ...SubscribeOption) (Subscriber, error)
  • func (n *NatsBroker) String() string
  • func (n *NatsBroker) Subscribe(channel string, handler Handler, opts ...SubscribeOption) (Subscriber, error)
• type Option
  • func Addrs(addrs ...string) Option
  • func Codec(c codec.Marshaller) Option
  • func Registry(r interface{}) Option
  • func Secure(b bool) Option
  • func TLSConfig(t *tls.Config) Option
• type Options
• type PublishOption
• type PublishOptions
• type RequestEvent
• type RequestInput
• type SubscribeOption
  • func DisableAutoAck() SubscribeOption
  • func Queue(name string) SubscribeOption
  • func SubscribeContext(ctx context.Context) SubscribeOption
• type SubscribeOptions
  • func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions
• type SubscribeService
• type Subscriber
  • func Respond(channel string, handler ActionHandle, opts ...SubscribeOption) (Subscriber, error)
  • func Subscribe(channel string, handler Handler, opts ...SubscribeOption) (Subscriber, error)

Variables

var (
	DefaultBroker = NewBroker()
)

Functions

func Connect

func Connect() error

func Disconnect

func Disconnect() error

func Init

func Init(opts ...Option) error

func Publish

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

func Request

func Request(channel string, msg *RequestInput, opts ...PublishOption) (interface{}, error)

func String

func String() string

Types

type ActionHandle

type ActionHandle func(RequestEvent) interface{}

type Broker

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

Broker is an interface used for asynchronous messaging.

type Event

type Event interface {
	Channel() string
	Message() *Message
	Ack() 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 channel. 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
}

type NatsBroker

type NatsBroker struct {
	sync.Once
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewBroker

func NewBroker(opts ...Option) *NatsBroker

func (*NatsBroker) Address

func (n *NatsBroker) Address() string

func (*NatsBroker) Connect

func (n *NatsBroker) Connect() error

func (*NatsBroker) Disconnect

func (n *NatsBroker) Disconnect() error

func (*NatsBroker) Init

func (n *NatsBroker) Init(opts ...Option) error

func (*NatsBroker) Options

func (n *NatsBroker) Options() Options

func (*NatsBroker) Publish

func (n *NatsBroker) Publish(channel string, msg *Message, opts ...PublishOption) error

func (*NatsBroker) Request

func (n *NatsBroker) Request(channel string, msg *RequestInput, opts ...PublishOption) (interface{}, error)

func (*NatsBroker) Respond

func (n *NatsBroker) Respond(channel string, handler ActionHandle, opts ...SubscribeOption) (Subscriber, error)

func (*NatsBroker) String

func (n *NatsBroker) String() string

func (*NatsBroker) Subscribe

func (n *NatsBroker) Subscribe(channel string, handler Handler, opts ...SubscribeOption) (Subscriber, error)

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.Marshaller) Option

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

func Registry

func Registry(r interface{}) 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 {
	Addresses []string
	Secure    bool
	Codec     codec.Marshaller
	TLSConfig *tls.Config
	// Registry used for clustering
	// Other options for implementations of the interface
	// can be stored in a context
	Context context.Context
}

type PublishOption

type PublishOption func(*PublishOptions)

type PublishOptions

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

type RequestEvent

type RequestEvent interface {
	Channel() string
	Message() *RequestInput
	Ack() error
}

type RequestInput

type RequestInput struct {
	Headers map[string]string
	Body    interface{}
	Params  map[string]interface{}
	Query   map[string]interface{}
}

*

• For requests

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 SubscribeService

type SubscribeService interface {
	SendMsg(interface{}) error
	RecvMsg(interface{}) error
	Close() error
	Recv() (*Message, error)
}

type Subscriber

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

Subscriber is a convenience return type for the Subscribe method

func Respond

func Respond(channel string, handler ActionHandle, opts ...SubscribeOption) (Subscriber, error)

func Subscribe

func Subscribe(channel string, handler Handler, opts ...SubscribeOption) (Subscriber, error)