events

Documentation

Index

• type EventRouter
  • func NewEventRouter(options ...EventRouterOption) (*EventRouter, error)
  • func (e *EventRouter) AddHandler(name string, topic string, f func(msg *message.Message) error)
  • func (e *EventRouter) Close() error
  • func (e *EventRouter) DumpRawEvents(msg *message.Message) error
  • func (e *EventRouter) Run(ctx context.Context) error
  • func (e *EventRouter) RunHandlers(ctx context.Context) error
• type EventRouterOption
  • func WithLogger(logger watermill.LoggerAdapter) EventRouterOption
  • func WithPublisher(publisher message.Publisher) EventRouterOption
  • func WithSubscriber(subscriber message.Subscriber) EventRouterOption
  • func WithVerbose(verbose bool) EventRouterOption
• type PublisherManager
  • func NewPublisherManager() *PublisherManager
  • func (s *PublisherManager) Publish(payload interface{}) error
  • func (s *PublisherManager) PublishBlind(payload interface{})
  • func (s *PublisherManager) RegisterPublisher(topic string, sub message.Publisher)

Types

type EventRouter

type EventRouter struct {
	Publisher  message.Publisher
	Subscriber message.Subscriber
	// contains filtered or unexported fields
}

func NewEventRouter

func NewEventRouter(options ...EventRouterOption) (*EventRouter, error)

func (*EventRouter) AddHandler

func (e *EventRouter) AddHandler(name string, topic string, f func(msg *message.Message) error)

func (*EventRouter) Close

func (e *EventRouter) Close() error

func (*EventRouter) DumpRawEvents

func (e *EventRouter) DumpRawEvents(msg *message.Message) error

func (*EventRouter) Run

func (e *EventRouter) Run(ctx context.Context) error

func (*EventRouter) RunHandlers

func (e *EventRouter) RunHandlers(ctx context.Context) error

type EventRouterOption

type EventRouterOption func(*EventRouter)

func WithLogger

func WithLogger(logger watermill.LoggerAdapter) EventRouterOption

func WithPublisher

func WithPublisher(publisher message.Publisher) EventRouterOption

func WithSubscriber

func WithSubscriber(subscriber message.Subscriber) EventRouterOption

func WithVerbose

func WithVerbose(verbose bool) EventRouterOption

type PublisherManager

type PublisherManager struct {
	Publishers map[string][]message.Publisher
	// contains filtered or unexported fields
}

PublisherManager is used to distribute messages to a set of Publishers. As such, you "register" a publisher to the given topic. When you Publish a message to a topic, it will get distributed to all publishers on the channel they were registered with that topic.

The Manager also keeps a sequence number for each outgoing message, in the order they are handled by Publish.

func NewPublisherManager

func NewPublisherManager() *PublisherManager

func (*PublisherManager) Publish

func (s *PublisherManager) Publish(payload interface{}) error

Publish distributes a message to all Publishers across all topics. Serializing the payload to JSON is done by Publish itself.

Returns an error for any processing or distribution issues.

func (*PublisherManager) PublishBlind

func (s *PublisherManager) PublishBlind(payload interface{})

func (*PublisherManager) RegisterPublisher

func (s *PublisherManager) RegisterPublisher(topic string, sub message.Publisher)