broker

Documentation

Index

• func Publish(topic string, msg *pubsub.Message, opts ...PublishOption) error
• func Shutdown() error
• func String() string
• func Subscribe(topic string, handler Handler, opts ...SubscribeOption) error
• type Broker
  • func NewBroker(opts ...Option) Broker
• type Event
• type Handler
• type Message
• type Option
  • func ClientOption(c ...option.ClientOption) Option
  • func CreateSubscription(b bool) Option
  • func DeleteSubscription(b bool) Option
  • func ErrorHandler(h Handler) Option
  • func ProjectID(id string) Option
• type Options
• type PublishOption
  • func PublishContext(ctx context.Context) PublishOption
• type PublishOptions
• type SubscribeOption
  • func DisableAutoAck() SubscribeOption
  • func MaxExtension(d time.Duration) SubscribeOption
  • func MaxOutstandingMessages(max int) SubscribeOption
  • func Queue(name string) SubscribeOption
  • func SubscribeContext(ctx context.Context) SubscribeOption
• type SubscribeOptions
  • func NewSubscribeOptions(opts ...SubscribeOption) SubscribeOptions
• type Subscriber

Functions

func Publish

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

func Shutdown

func Shutdown() error

func String

func String() string

func Subscribe

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

Types

type Broker

type Broker interface {
	Options() Options
	Shutdown() error
	Publish(topic string, m *pubsub.Message, opts ...PublishOption) error
	Subscribe(topic string, h Handler, opts ...SubscribeOption) error
	String() string
}

Broker is an interface used for asynchronous messaging.

var DefaultBroker Broker

func NewBroker

func NewBroker(opts ...Option) Broker

NewBroker creates a new google pubsub broker

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(context.Context, *pubsub.Message) 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
}

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 CreateSubscription

func CreateSubscription(b bool) Option

CreateSubscription prevents the creation of the subscription if it not exists

func DeleteSubscription

func DeleteSubscription(b bool) Option

DeleteSubscription prevents the deletion of the subscription if it not exists

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 ProjectID

func ProjectID(id string) Option

ProjectID provides an option which sets the google project id

type Options

type Options struct {

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

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

func DisableAutoAck() SubscribeOption

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

func MaxExtension

func MaxExtension(d time.Duration) SubscribeOption

MaxExtension is the maximum period for which the Subscription should automatically extend the ack deadline for each message.

func MaxOutstandingMessages

func MaxOutstandingMessages(max int) SubscribeOption

MaxOutstandingMessages sets the maximum number of unprocessed messages (unacknowledged but not yet expired) to receive.

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