logrepl

Documentation

Index

• func Cleanup(ctx context.Context, c CleanupConfig) error
• type CDCConfig
• type CDCHandler
  • func NewCDCHandler(rs *internal.RelationSet, tableKeys map[string]string, ...) *CDCHandler
  • func (h *CDCHandler) Handle(ctx context.Context, m pglogrepl.Message, lsn pglogrepl.LSN) (pglogrepl.LSN, error)
• type CDCIterator
  • func NewCDCIterator(ctx context.Context, pool *pgxpool.Pool, c CDCConfig) (*CDCIterator, error)
  • func (i *CDCIterator) Ack(_ context.Context, sdkPos opencdc.Position) error
  • func (i *CDCIterator) Next(ctx context.Context) (opencdc.Record, error)
  • func (i *CDCIterator) StartSubscriber(ctx context.Context) error
  • func (i *CDCIterator) TXSnapshotID() string
  • func (i *CDCIterator) Teardown(ctx context.Context) error
• type CleanupConfig
• type CombinedIterator
  • func NewCombinedIterator(ctx context.Context, pool *pgxpool.Pool, conf Config) (*CombinedIterator, error)
  • func (c *CombinedIterator) Ack(ctx context.Context, p opencdc.Position) error
  • func (c *CombinedIterator) Next(ctx context.Context) (opencdc.Record, error)
  • func (c *CombinedIterator) Teardown(ctx context.Context) error
• type Config
  • func (c Config) Validate() error

Functions

func Cleanup

func Cleanup(ctx context.Context, c CleanupConfig) error

Cleanup drops the provided replication slot and publication. It will terminate any backends consuming the replication slot before deletion.

Types

type CDCConfig

type CDCConfig struct {
	LSN             pglogrepl.LSN
	SlotName        string
	PublicationName string
	Tables          []string
	TableKeys       map[string]string
	WithAvroSchema  bool
}

CDCConfig holds configuration values for CDCIterator.

type CDCHandler

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

CDCHandler is responsible for handling logical replication messages, converting them to a record and sending them to a channel.

func NewCDCHandler

func NewCDCHandler(rs *internal.RelationSet, tableKeys map[string]string, out chan<- opencdc.Record) *CDCHandler

func (*CDCHandler) Handle

func (h *CDCHandler) Handle(ctx context.Context, m pglogrepl.Message, lsn pglogrepl.LSN) (pglogrepl.LSN, error)

Handle is the handler function that receives all logical replication messages. Returns non-zero LSN when a record was emitted for the message.

type CDCIterator

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

CDCIterator asynchronously listens for events from the logical replication slot and returns them to the caller through Next.

func NewCDCIterator

func NewCDCIterator(ctx context.Context, pool *pgxpool.Pool, c CDCConfig) (*CDCIterator, error)

NewCDCIterator initializes logical replication by creating the publication and subscription manager.

func (*CDCIterator) Ack

func (i *CDCIterator) Ack(_ context.Context, sdkPos opencdc.Position) error

Ack forwards the acknowledgment to the subscription.

func (*CDCIterator) Next

func (i *CDCIterator) Next(ctx context.Context) (opencdc.Record, error)

Next returns the next record retrieved from the subscription. This call will block until either a record is returned from the subscription, the subscription stops because of an error or the context gets canceled. Returns error when the subscription has been started.

func (*CDCIterator) StartSubscriber

func (i *CDCIterator) StartSubscriber(ctx context.Context) error

StartSubscriber starts the logical replication service in the background. Blocks until the subscription becomes ready.

func (*CDCIterator) TXSnapshotID

func (i *CDCIterator) TXSnapshotID() string

TXSnapshotID returns the transaction snapshot which is received when the replication slot is created. The value can be empty, when the iterator is resuming.

func (*CDCIterator) Teardown

func (i *CDCIterator) Teardown(ctx context.Context) error

Teardown stops the CDC subscription and blocks until the subscription is done or the context gets canceled. If the subscription stopped with an unexpected error, the error is returned.

type CleanupConfig

type CleanupConfig struct {
	URL             string
	SlotName        string
	PublicationName string
}

type CombinedIterator

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

func NewCombinedIterator

func NewCombinedIterator(ctx context.Context, pool *pgxpool.Pool, conf Config) (*CombinedIterator, error)

NewCombinedIterator will initialize and start the Snapshot and CDC iterators. Failure to parse the position or validate the config will return an error.

func (*CombinedIterator) Ack

func (c *CombinedIterator) Ack(ctx context.Context, p opencdc.Position) error

func (*CombinedIterator) Next

func (c *CombinedIterator) Next(ctx context.Context) (opencdc.Record, error)

Next provides the next available record from the snapshot or CDC stream. If the end of the snapshot is reached, next will switch to the CDC iterator and retrive the next available record. Failure to switch the iterator will return an error.

func (*CombinedIterator) Teardown

func (c *CombinedIterator) Teardown(ctx context.Context) error

Teardown will stop and teardown the CDC and Snapshot iterators.

type Config

type Config struct {
	Position          opencdc.Position
	SlotName          string
	PublicationName   string
	Tables            []string
	TableKeys         map[string]string
	WithSnapshot      bool
	SnapshotFetchSize int
}

func (Config) Validate

func (c Config) Validate() error

Validate performs validation tasks on the config.