eventx

Dec 19, 2023

Changes in this version

type Option

+ func WithErrorLogger(fn func(err error)) Option

Nov 29, 2023

Changes in this version

+ type GetSequenceFunc func(ctx context.Context) (sql.NullInt64, error)

+ type RetryConsumer struct

+ func NewRetryConsumer[E EventConstraint](runner *Runner[E], repo Repository[E], getSequence GetSequenceFunc, ...) *RetryConsumer[E]

+ func (c *RetryConsumer[E]) RunConsumer(ctx context.Context)

+ type RetryConsumerOption func(conf *retryConsumerConfig)

+ func WithConsumerRetryDuration(d time.Duration) RetryConsumerOption

+ func WithRetryConsumerErrorLogger(logger func(err error)) RetryConsumerOption

+ func WithRetryConsumerFetchLimit(limit uint64) RetryConsumerOption

+ type SetSequenceFunc func(ctx context.Context, seq uint64) error

May 17, 2023

Changes in this version

type EventConstraint

+ GetID func() uint64

+ type RetentionJob struct

+ func NewRetentionJob(runner *Runner[E], repo RetentionRepository, options ...RetentionOption) *RetentionJob[E]

+ func (j *RetentionJob[E]) RunJob(ctx context.Context)

+ type RetentionOption func(opts *retentionOptions)

+ func WithDeleteBatchSize(size uint64) RetentionOption

+ func WithMaxTotalEvents(maxSize uint64) RetentionOption

+ func WithRetentionErrorLogger(logger func(err error)) RetentionOption

+ func WithRetentionErrorRetryDuration(d time.Duration) RetentionOption

+ type RetentionRepository interface

+ DeleteEventsBefore func(ctx context.Context, beforeSeq uint64) error

+ GetMinSequence func(ctx context.Context) (sql.NullInt64, error)

Feb 14, 2023

Changes in this version

+ type EventConstraint interface

+ GetSequence func() uint64

+ GetSize func() uint64

Aug 22, 2021

Changes in this version

+ var ErrEventNotFound = errors.New("not found any events from a sequence")

Jul 5, 2021

Changes in this version

+ func MergeContext(a, b context.Context) context.Context

+ type Event struct

+ CreatedAt time.Time

+ Data string

+ ID uint64

+ Seq uint64

+ type Option func(opts *eventxOptions)

+ func WithCoreStoredEventsSize(size uint64) Option

+ func WithDBProcessorErrorRetryTimer(d time.Duration) Option

+ func WithDBProcessorRetryTimer(d time.Duration) Option

+ func WithGetLastEventsLimit(limit uint64) Option

+ func WithGetUnprocessedEventsLimit(limit uint64) Option

+ func WithLogger(logger *zap.Logger) Option

+ type Repository interface

+ GetEventsFrom func(ctx context.Context, from uint64, limit uint64) ([]Event, error)

+ GetLastEvents func(ctx context.Context, limit uint64) ([]Event, error)

+ GetUnprocessedEvents func(ctx context.Context, limit uint64) ([]Event, error)

+ UpdateSequences func(ctx context.Context, events []Event) error

+ type Runner struct

+ func NewRunner(repo Repository, unmarshal UnmarshalEvent, options ...Option) *Runner

+ func (r *Runner) NewSubscriber(from uint64, fetchLimit uint64, options ...SubscriberOption) *Subscriber

+ func (r *Runner) Run(ctx context.Context)

+ func (r *Runner) Signal()

+ type Subscriber struct

+ func (s *Subscriber) Fetch(ctx context.Context) ([]UnmarshalledEvent, error)

+ type SubscriberOption func(opts *subscriberOptions)

+ func WithSubscriberSizeLimit(sizeLimit uint64) SubscriberOption

+ type Timer interface

+ Chan func() <-chan time.Time

+ Reset func()

+ ResetAfterChan func()

+ type UnmarshalEvent func(e Event) UnmarshalledEvent

+ type UnmarshalledEvent interface

+ GetSequence func() uint64