cdc

type ByEntityIdRevAsc ¶ added in v0.11.0

type ByEntityIdRevAsc []etre.CDCEvent

ByEntityIdRevAsc sorts a slice of etre.CDCEvent by EntityId, Ts ascending. This is the correct way to sort CDC events. Timestamp are not reliable. This is used in Store.Find() instead of having MongoDB sort the results which can fail with too many CDC events because MongoDB has a max sort size. Also, it's better to offload sorting to Etre which can scale out more easily.

func (a ByEntityIdRevAsc) Len() int

func (a ByEntityIdRevAsc) Less(i, j int) bool

func (a ByEntityIdRevAsc) Swap(i, j int)

type ByTsAsc ¶ added in v0.11.0

type ByTsAsc []etre.CDCEvent

func (a ByTsAsc) Len() int

func (a ByTsAsc) Less(i, j int) bool

func (a ByTsAsc) Swap(i, j int)

type Filter ¶

type Filter struct {
	SinceTs int64 // Only read events that have a timestamp greater than or equal to this value.
	UntilTs int64 // Only read events that have a timestamp less than this value.
	Limit   int64
	Order   sort.Interface
}

Filter contains fields that are used to filter events that the CDC reads. Unset fields are ignored.

var NoFilter Filter

type RetryPolicy ¶

type RetryPolicy struct {
	RetryCount int
	RetryWait  int // milliseconds
}

RetryPolicy represents the retry policy that the Store uses when reading and writing events.

var NoRetryPolicy RetryPolicy

type Store ¶

type Store interface {
	// Write writes the CDC event to a persisitent data store. If writing
	// fails, it retries according to the RetryPolicy. If retrying fails,
	// the event is written to the fallbackFile. An error is returned if
	// writing to the persistent data store fails, even if writing to
	// fallback file succeeds.
	Write(context.Context, etre.CDCEvent) error

	// Read queries a persistent data store for events that satisfy the
	// given filter.
	Read(Filter) ([]etre.CDCEvent, error)
}

A Store reads and writes CDC events to/from a persistent data store.

func NewStore(coll *mongo.Collection, fallbackFile string, writeRetryPolicy RetryPolicy) Store