broker

Documentation

Index

• func MustPublish(ctx context.Context, topic string, message proto.Message, ...)
• func Publish(ctx context.Context, topic string, message proto.Message, ...) error
• func PublishRaw(ctx context.Context, topic string, body []byte, header map[string]string, ...) error
• func Register(b Broker)
• func SubscribeCancellable(ctx context.Context, topic string, handler SubscriberHandler, ...) error
• type Broker
  • func Default() Broker
  • func NewBroker(s string, opts ...Option) Broker
• type Message
• type Option
  • func BeforeDisconnect(f func() error) Option
  • func WithContext(ctx context.Context) Option
• type Options
• type PublishOption
  • func PublishContext(ctx context.Context) PublishOption
• type PublishOptions
• type SubscribeOpener
• type SubscribeOption
  • func HandleError(h func(error)) SubscribeOption
  • func Queue(name string) SubscribeOption
  • func SubscribeContext(ctx context.Context) SubscribeOption
• type SubscribeOptions
  • func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions
• type SubscriberHandler
• type TopicOpener
• type UnSubscriber
  • func Subscribe(ctx context.Context, topic string, handler SubscriberHandler, ...) (UnSubscriber, error)

Functions

func MustPublish

func MustPublish(ctx context.Context, topic string, message proto.Message, opts ...PublishOption)

MustPublish publishes a message ignoring the error

func Publish

func Publish(ctx context.Context, topic string, message proto.Message, opts ...PublishOption) error

Publish sends a message to standard broker. For the moment, forward message to client.Publish

func PublishRaw

func PublishRaw(ctx context.Context, topic string, body []byte, header map[string]string, opts ...PublishOption) error

PublishRaw sends a message to standard broker. For the moment, forward message to client.Publish

func Register

func Register(b Broker)

func SubscribeCancellable

func SubscribeCancellable(ctx context.Context, topic string, handler SubscriberHandler, opts ...SubscribeOption) error

Types

type Broker

type Broker interface {
	PublishRaw(context.Context, string, []byte, map[string]string, ...PublishOption) error
	Publish(context.Context, string, proto.Message, ...PublishOption) error
	Subscribe(context.Context, string, SubscriberHandler, ...SubscribeOption) (UnSubscriber, error)
}

func Default

func Default() Broker

func NewBroker

func NewBroker(s string, opts ...Option) Broker

NewBroker wraps a standard broker but prevents it from disconnecting while there still is a service running

type Message

type Message interface {
	Unmarshal(target proto.Message) (context.Context, error)
	RawData() (map[string]string, []byte)
}

type Option

type Option func(*Options)

Option definition

func BeforeDisconnect

func BeforeDisconnect(f func() error) Option

BeforeDisconnect registers all functions to be triggered before the broker disconnect

func WithContext

func WithContext(ctx context.Context) Option

type Options

type Options struct {
	Context context.Context
	// contains filtered or unexported fields
}

Options to the broker

type PublishOption

type PublishOption func(options *PublishOptions)

func PublishContext

func PublishContext(ctx context.Context) PublishOption

type PublishOptions

type PublishOptions struct {
	Context context.Context
}

type SubscribeOpener

type SubscribeOpener func(string, ...SubscribeOption) (*pubsub.Subscription, error)

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

func HandleError

func HandleError(h func(error)) SubscribeOption

HandleError sets an ErrorHandler to 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 func(error)

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

type SubscriberHandler func(Message) error

type TopicOpener

type TopicOpener func(context.Context, string) (*pubsub.Topic, error)

type UnSubscriber

type UnSubscriber func() error

func Subscribe

func Subscribe(ctx context.Context, topic string, handler SubscriberHandler, opts ...SubscribeOption) (UnSubscriber, error)