sink

Sep 16, 2023

Changes in this version

+ const UpdateSourceSQL

+ var LogLSNRegex = regexp.MustCompile(...)

+ var LogTxTimeRegex = regexp.MustCompile(`last completed transaction was at log time (.*)\.?$`)

+ func PGTime2Time(ts uint64) time.Time

+ func ScanCheckpointFromLog(f io.Reader) (lsn, ts string, err error)

+ type ApplyFn func(sourceRemaining int, message source.Change, committed chan cursor.Checkpoint) error

+ type BaseSink struct

+ CleanFn CleanFn

+ func (b *BaseSink) Apply(changes chan source.Change) (committed chan cursor.Checkpoint)

+ func (b *BaseSink) Error() error

+ func (b *BaseSink) Setup() (cp cursor.Checkpoint, err error)

+ func (b *BaseSink) Stop() error

+ type CleanFn func()

+ type PGXSink struct

+ BatchTXSize int

+ ConnStr string

+ LogReader io.Reader

+ Renice int64

+ SourceID string

+ func (p *PGXSink) Apply(changes chan source.Change) chan cursor.Checkpoint

+ func (p *PGXSink) ReplicationLagMilliseconds() int64

+ func (p *PGXSink) Setup() (cp cursor.Checkpoint, err error)

+ type PulsarSink struct

+ PulsarOption pulsar.ClientOptions

+ PulsarTopic string

+ ReplicatedClusters []string

+ SetupTracker SetupTracker

+ func (p *PulsarSink) Apply(changes chan source.Change) chan cursor.Checkpoint

+ func (p *PulsarSink) Setup() (cp cursor.Checkpoint, err error)

+ type SetupTracker func(client pulsar.Client, topic string) (cursor.Tracker, error)

+ type Sink interface

+ Apply func(changes chan source.Change) (committed chan cursor.Checkpoint)

+ Error func() error

+ Setup func() (cp cursor.Checkpoint, err error)

+ Stop func() error