broker

README

CloudEvents

Broker implementation provides queuing, backpressure, filtering

CloudEvents has multiple transports (http, pubsub, kafka...) available

Concepts

Actors

• Event Source
• Event Sink
• Broker

Messaging patterns

• Fire and Forgot
• Request and Response

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 Context(ctx context.Context) Option
  • func Name(n string) Option
• type Options
• type PublishOption
  • func PublishAsync(b bool) PublishOption
• type PublishOptions
• type Publisher
  • func NewPublisher(topic string, opts ...PublishOption) (Publisher, error)
• type RecoveryHandler
• type 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, event.Event) cloudevents.Result

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 Context

func Context(ctx context.Context) Option

Context appContext to trigger terminate signal

func Name

func Name(n string) Option

Name of the service

type Options

type Options struct {
	Name    string
	Context context.Context
}

type PublishOption

type PublishOption func(*PublishOptions)

func PublishAsync

func PublishAsync(b bool) PublishOption

type PublishOptions

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

type Publisher

type Publisher interface {
	Publish(ctx context.Context, event event.Event) error
}

func NewPublisher

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

type RecoveryHandler

type RecoveryHandler func(context.Context, event.Event, 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 WithRecoveryHandler

func WithRecoveryHandler(r RecoveryHandler) SubscribeOption

WithRecoveryHandler sets the function for recovering from a panic.

type SubscribeOptions

type SubscribeOptions struct {
	RecoveryHandler RecoveryHandler
}