broker

Documentation

Index

• Constants
• func AddSubscriber(subscription string, handler Handler, opts ...SubscribeOption) error
• func Start() error
• type Broker
  • func NewBroker(ctx context.Context, opts ...Option) Broker
• type Handler
• type Option
  • func ClientOption(c ...option.ClientOption) Option
  • func Context(ctx context.Context) Option
  • func ErrorHandler(h Handler) Option
  • func Name(n string) Option
  • func ProjectID(id string) Option
• type Options
• type PublishOption
  • func PublishAsync(b bool) PublishOption
  • func WithPublishSettings(publishSettings pubsub.PublishSettings) PublishOption
• type PublishOptions
• type Publisher
  • func NewPublisher(topic string, opts ...PublishOption) (Publisher, error)
• type RecoveryHandler
• type SubscribeOption
  • func WithReceiveSettings(receiveSettings pubsub.ReceiveSettings) SubscribeOption
  • func WithRecoveryHandler(r RecoveryHandler) SubscribeOption
• type SubscribeOptions

Constants

const (
	DefaultName = "mkit.broker.default"
)

Functions

func AddSubscriber

func AddSubscriber(subscription string, handler Handler, opts ...SubscribeOption) error

func Start

func Start() error

Types

type Broker

type Broker interface {
	NewPublisher(topic string, opts ...PublishOption) (pub Publisher, err error)
	AddSubscriber(subscription string, hdlr Handler, opts ...SubscribeOption) (err error)
	Start() error
}

Broker is an interface used for asynchronous messaging.

var DefaultBroker Broker

func NewBroker

func NewBroker(ctx context.Context, opts ...Option) Broker

NewBroker creates and returns a new Broker based on the packages within.

type Handler

type Handler func(context.Context, *pubsub.Message)

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 Option

type Option func(*Options)

func ClientOption

func ClientOption(c ...option.ClientOption) Option

ClientOption is a broker Option which allows google pubsub client options to be set for the client

func Context

func Context(ctx context.Context) Option

Context appContext to trigger terminate signal

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 Name

func Name(n string) Option

Name of the service

func ProjectID

func ProjectID(id string) Option

ProjectID provides an option which sets the google project id

type Options

type Options struct {
	Name          string
	ClientOptions []option.ClientOption
	ProjectID     string

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

	Context context.Context
}

type PublishOption

type PublishOption func(*PublishOptions)

func PublishAsync

func PublishAsync(b bool) PublishOption

func WithPublishSettings

func WithPublishSettings(publishSettings pubsub.PublishSettings) PublishOption

type PublishOptions

type PublishOptions struct {
	// pubsub PublishSettings
	PublishSettings pubsub.PublishSettings
	// publishes msg to the topic asynchronously if set to true.
	// Default false. i.e., publishes synchronously(blocking)
	Async bool
}

TODO support more pubsub.PublishSettings settings

type Publisher

type Publisher interface {
	Publish(ctx context.Context, msg *pubsub.Message) error
}

func NewPublisher

func NewPublisher(topic string, opts ...PublishOption) (Publisher, error)

type RecoveryHandler

type RecoveryHandler func(context.Context, *pubsub.Message, interface{})

RecoveryHandler is a function that is called when the recovery middleware recovers from a panic. The func takes the receive context, message and the return value from recover which reports whether the goroutine is panicking. Example usages of HandlerFunc could be to log panics or nack/ack messages which cause panics.

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

func WithReceiveSettings

func WithReceiveSettings(receiveSettings pubsub.ReceiveSettings) SubscribeOption

func WithRecoveryHandler

func WithRecoveryHandler(r RecoveryHandler) SubscribeOption

WithRecoveryHandler sets the function for recovering from a panic.

type SubscribeOptions

type SubscribeOptions struct {
	// pubsub ReceiveSettings
	ReceiveSettings pubsub.ReceiveSettings

	RecoveryHandler RecoveryHandler
}

SubscribeOptions TODO support more pubsub.ReceiveSettings settings