cdc

Documentation

Overview

Package cdc provides interfaces for reading and writing change data capture (CDC) events.

Index

• type ByEntityIdRevAsc
  • func (a ByEntityIdRevAsc) Len() int
  • func (a ByEntityIdRevAsc) Less(i, j int) bool
  • func (a ByEntityIdRevAsc) Swap(i, j int)
• type ByTsAsc
  • func (a ByTsAsc) Len() int
  • func (a ByTsAsc) Less(i, j int) bool
  • func (a ByTsAsc) Swap(i, j int)
• type Filter
• type RetryPolicy
• type Store
  • func NewStore(coll *mongo.Collection, fallbackFile string, writeRetryPolicy RetryPolicy) Store

Types

type ByEntityIdRevAsc

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 (ByEntityIdRevAsc) Len

func (a ByEntityIdRevAsc) Len() int

func (ByEntityIdRevAsc) Less

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

func (ByEntityIdRevAsc) Swap

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

type ByTsAsc

type ByTsAsc []etre.CDCEvent

func (ByTsAsc) Len

func (a ByTsAsc) Len() int

func (ByTsAsc) Less

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

func (ByTsAsc) Swap

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

NoFilter is a convenience var for calls like Read(cdc.NoFilter). Other packages must not modify this var.

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

NoRetryPolicy is a convenience var for calls like NewStore(cdc.NoRetryPolicy). Other packages must not modify this var.

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

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