kafka

Documentation

Index

• type EventBus
  • func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error)
  • func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.EventHandler) error
  • func (b *EventBus) Close() error
  • func (b *EventBus) Errors() <-chan error
  • func (b *EventBus) HandleEvent(ctx context.Context, event eh.Event) error
  • func (b *EventBus) HandlerType() eh.EventHandlerType
• type Option
  • func WithCodec(codec eh.EventCodec) Option
  • func WithStartOffset(startOffset int64) Option
  • func WithTopic(topic string) Option

Types

type EventBus

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus is a local event bus that delegates handling of published events to all matching registered handlers, in order of registration.

func NewEventBus

func NewEventBus(addressList, appID string, options ...Option) (*EventBus, error)

NewEventBus creates an EventBus, with optional GCP connection settings.

func (*EventBus) AddHandler

func (b *EventBus) AddHandler(ctx context.Context, m eh.EventMatcher, h eh.EventHandler) error

AddHandler implements the AddHandler method of the eventhorizon.EventBus interface.

func (*EventBus) Close

func (b *EventBus) Close() error

Close implements the Close method of the eventhorizon.EventBus interface.

func (*EventBus) Errors

func (b *EventBus) Errors() <-chan error

Errors implements the Errors method of the eventhorizon.EventBus interface.

func (*EventBus) HandleEvent

func (b *EventBus) HandleEvent(ctx context.Context, event eh.Event) error

HandleEvent implements the HandleEvent method of the eventhorizon.EventHandler interface.

func (*EventBus) HandlerType

func (b *EventBus) HandlerType() eh.EventHandlerType

HandlerType implements the HandlerType method of the eventhorizon.EventHandler interface.

type Option

type Option func(*EventBus) error

Option is an option setter used to configure creation.

func WithCodec

func WithCodec(codec eh.EventCodec) Option

WithCodec uses the specified codec for encoding events.

func WithStartOffset

func WithStartOffset(startOffset int64) Option

WithStartOffset sets the consumer group's offset to start at Defaults to: LastOffset Per the kafka client documentation

StartOffset determines from whence the consumer group should begin
consuming when it finds a partition without a committed offset.  If
non-zero, it must be set to one of FirstOffset or LastOffset.

func WithTopic

func WithTopic(topic string) Option

WithTopic uses the specified topic for the event bus topic name

Defaults to: appID + "_events"