eventbridge

Documentation

Index

• type Bridge
  • func New(s Source, m messaging.Sender, d Duplicate, o ...Option) *Bridge
  • func (b *Bridge) Close() error
  • func (b *Bridge) Run(ctx context.Context) error
• type Duplicate
• type Event
• type EventHandlerFunc
• type EventHandlerMiddleware
  • func NewPublishedEventsCollector(m *PublishedEventsMetrics) EventHandlerMiddleware
  • func WrapEventHandler(em EventHandlerMiddleware, ems ...EventHandlerMiddleware) EventHandlerMiddleware
• type Option
  • func SetLogger(l log.Logger) Option
  • func SetPublishedEventsCollector(c *PublishedEventsMetrics) Option
  • func SetUnpublishedStreamCollector(c *UnpublishedStreamMetrics) Option
• type PublishedEventsCollector
  • func (c *PublishedEventsCollector) EventHandler(ctx context.Context, e Event) (err error)
• type PublishedEventsMetrics
  • func (m *PublishedEventsMetrics) Observer(st stream.Type, et stream.EventType, start time.Time)
• type Source
• type UnpublishedStream
• type UnpublishedStreamCollector
  • func (c *UnpublishedStreamCollector) WalkFunc(ctx context.Context, s *UnpublishedStream) (err error)
• type UnpublishedStreamMetrics
  • func (m *UnpublishedStreamMetrics) Observer(st stream.Type, start time.Time)
• type WalkFunc
• type WalkUnpublishedEventsMiddleware
  • func NewUnpublishedStreamCollector(m *UnpublishedStreamMetrics) WalkUnpublishedEventsMiddleware
  • func WrapWalkUnpublishedEvents(em WalkUnpublishedEventsMiddleware, ems ...WalkUnpublishedEventsMiddleware) WalkUnpublishedEventsMiddleware

Types

type Bridge

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

func New

func New(s Source, m messaging.Sender, d Duplicate, o ...Option) *Bridge

func (*Bridge) Close

func (b *Bridge) Close() error

func (*Bridge) Run

func (b *Bridge) Run(ctx context.Context) error

type Duplicate

type Duplicate interface {
	Exists(Event) bool
	Ensure(Event)
}

type Event

type Event struct {
	StreamID   stream.ID
	StreamType stream.Type
	EventType  stream.EventType
	Version    stream.Version
	Envelope   []byte
}

type EventHandlerFunc

type EventHandlerFunc func(context.Context, Event) error

type EventHandlerMiddleware

type EventHandlerMiddleware func(EventHandlerFunc) EventHandlerFunc

func NewPublishedEventsCollector

func NewPublishedEventsCollector(m *PublishedEventsMetrics) EventHandlerMiddleware

func WrapEventHandler

func WrapEventHandler(em EventHandlerMiddleware, ems ...EventHandlerMiddleware) EventHandlerMiddleware

type Option

type Option func(*Bridge)

func SetLogger

func SetLogger(l log.Logger) Option

func SetPublishedEventsCollector

func SetPublishedEventsCollector(c *PublishedEventsMetrics) Option

func SetUnpublishedStreamCollector

func SetUnpublishedStreamCollector(c *UnpublishedStreamMetrics) Option

type PublishedEventsCollector

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

func (*PublishedEventsCollector) EventHandler

func (c *PublishedEventsCollector) EventHandler(ctx context.Context, e Event) (err error)

type PublishedEventsMetrics

type PublishedEventsMetrics struct {
	Counter   metrics.Counter
	Histogram metrics.Histogram
}

func (*PublishedEventsMetrics) Observer

func (m *PublishedEventsMetrics) Observer(st stream.Type, et stream.EventType, start time.Time)

type Source

type Source interface {
	WalkUnpublishedEvents(context.Context, WalkFunc) error
	UpdateVersion(context.Context, stream.ID, stream.Type, stream.Version) error
	OnEvent(EventHandlerFunc)
	OnConnect(func(context.Context) error)
	Connect(context.Context) error
	Close() error
}

type UnpublishedStream

type UnpublishedStream struct {
	SID                    stream.ID
	Type                   stream.Type
	LatestPublishedVersion stream.Version
	CurrentVersion         stream.Version
	Events                 []Event
}

type UnpublishedStreamCollector

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

func (*UnpublishedStreamCollector) WalkFunc

func (c *UnpublishedStreamCollector) WalkFunc(ctx context.Context, s *UnpublishedStream) (err error)

type UnpublishedStreamMetrics

type UnpublishedStreamMetrics struct {
	Counter   metrics.Counter
	Histogram metrics.Histogram
}

func (*UnpublishedStreamMetrics) Observer

func (m *UnpublishedStreamMetrics) Observer(st stream.Type, start time.Time)

type WalkFunc

type WalkFunc func(context.Context, *UnpublishedStream) error

type WalkUnpublishedEventsMiddleware

type WalkUnpublishedEventsMiddleware func(WalkFunc) WalkFunc

func NewUnpublishedStreamCollector

func NewUnpublishedStreamCollector(m *UnpublishedStreamMetrics) WalkUnpublishedEventsMiddleware

func WrapWalkUnpublishedEvents

func WrapWalkUnpublishedEvents(em WalkUnpublishedEventsMiddleware, ems ...WalkUnpublishedEventsMiddleware) WalkUnpublishedEventsMiddleware