poller

Documentation

Index

• type Option
  • func WithAggregateKinds[K eventsourcing.ID](at ...eventsourcing.Kind) Option[K]
  • func WithLimit[K eventsourcing.ID](limit int) Option[K]
  • func WithMetadata[K eventsourcing.ID](metadata store.MetadataFilter) Option[K]
  • func WithMetadataKV[K eventsourcing.ID](key string, values ...string) Option[K]
  • func WithPollInterval[K eventsourcing.ID](pollInterval time.Duration) Option[K]
  • func WithSplits[K eventsourcing.ID](splits uint32, splitIDs []uint32) Option[K]
• type Poller
  • func New[K eventsourcing.ID](logger *slog.Logger, repository Repository[K], options ...Option[K]) Poller[K]
  • func (p *Poller[K]) Feed(ctx context.Context, sinker sink.Sinker[K]) error
• type Repository

Types

type Option

type Option[K eventsourcing.ID] func(*Poller[K])

func WithAggregateKinds

func WithAggregateKinds[K eventsourcing.ID](at ...eventsourcing.Kind) Option[K]

func WithLimit

func WithLimit[K eventsourcing.ID](limit int) Option[K]

func WithMetadata

func WithMetadata[K eventsourcing.ID](metadata store.MetadataFilter) Option[K]

func WithMetadataKV

func WithMetadataKV[K eventsourcing.ID](key string, values ...string) Option[K]

func WithPollInterval

func WithPollInterval[K eventsourcing.ID](pollInterval time.Duration) Option[K]

func WithSplits

func WithSplits[K eventsourcing.ID](splits uint32, splitIDs []uint32) Option[K]

type Poller

type Poller[K eventsourcing.ID] struct {
	// contains filtered or unexported fields
}

func New

func New[K eventsourcing.ID](logger *slog.Logger, repository Repository[K], options ...Option[K]) Poller[K]

func (*Poller[K]) Feed

func (p *Poller[K]) Feed(ctx context.Context, sinker sink.Sinker[K]) error

Feed forwards the handling to a sink. eg: a message queue

type Repository

type Repository[K eventsourcing.ID] interface {
	PendingEvents(ctx context.Context, batchSize int, filter store.Filter) ([]*eventsourcing.Event[K], error)
	AfterSink(ctx context.Context, eID eventid.EventID) error
}