store

Documentation

Index

• func InsertEvents(ctx context.Context, db EventStore, msgs chan *model.BatchUpdate, ...) chan []*pulsarutils.ConsumerMessageId
• func IsRetryableRedisError(err error, nonRetryableErrors []*regexp.Regexp) bool
• func WithRetry(executeDb func() error, nonRetryableErrors []*regexp.Regexp) error
• type EventStore
• type RedisEventStore
  • func NewRedisEventStore(db redis.UniversalClient, eventRetention configuration.EventRetentionPolicy) *RedisEventStore
  • func (repo *RedisEventStore) ReportEvents(update []*model.Event) error

Functions

func InsertEvents

func InsertEvents(
	ctx context.Context,
	db EventStore,
	msgs chan *model.BatchUpdate,
	bufferSize int,
	maxSize int,
	maxRows int,
	fatalErrors []*regexp.Regexp,
	metrics *commonmetrics.Metrics,
) chan []*pulsarutils.ConsumerMessageId

InsertEvents takes a channel of armada events and inserts them into the event db the events are republished to an output channel for further processing (e.g. Ackking)

func IsRetryableRedisError

func IsRetryableRedisError(err error, nonRetryableErrors []*regexp.Regexp) bool

IsRetryableRedisError returns true if the error doesn't match the list of nonRetryableErrors

func WithRetry

func WithRetry(executeDb func() error, nonRetryableErrors []*regexp.Regexp) error

Types

type EventStore

type EventStore interface {
	ReportEvents(update []*model.Event) error
}

type RedisEventStore

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

func NewRedisEventStore

func NewRedisEventStore(db redis.UniversalClient, eventRetention configuration.EventRetentionPolicy) *RedisEventStore

func (*RedisEventStore) ReportEvents

func (repo *RedisEventStore) ReportEvents(update []*model.Event) error