logrepl

package

v0.3.0 Latest Latest Go to latest Published: Sep 27, 2022 License: Apache-2.0 Imports: 10 Imported by: 0

Documentation ¶

Index ¶

Constants
Variables
func LSNToPosition(lsn pglogrepl.LSN) sdk.Position
func PositionToLSN(pos sdk.Position) (pglogrepl.LSN, error)
type CDCHandler
- func NewCDCHandler(rs *internal.RelationSet, keyColumn string, columns []string, ...) *CDCHandler
- func (h *CDCHandler) Handle(ctx context.Context, m pglogrepl.Message, lsn pglogrepl.LSN) error
type CDCIterator
- func NewCDCIterator(ctx context.Context, conn *pgx.Conn, config Config) (*CDCIterator, error)
type Config
type SnapshotConfig
type SnapshotIterator
- func NewSnapshotIterator(ctx context.Context, conn *pgx.Conn, cfg SnapshotConfig) (*SnapshotIterator, error)

Constants ¶

const (
	// TODO same constant is defined in packages longpoll, logrepl and destination
	//  use same constant everywhere
	MetadataPostgresTable = "postgres.table"
)

Variables ¶

View Source

var ErrSnapshotComplete = errors.New("snapshot complete")

ErrSnapshotComplete is returned by Next when a snapshot is finished

View Source

var ErrSnapshotInterrupt = errors.New("snapshot interrupted")

ErrSnapshotInterrupt is returned by Teardown when a snapshot is interrupted

Functions ¶

func LSNToPosition ¶

func LSNToPosition(lsn pglogrepl.LSN) sdk.Position

LSNToPosition converts a Postgres LSN to a Conduit position.

func PositionToLSN ¶

func PositionToLSN(pos sdk.Position) (pglogrepl.LSN, error)

PositionToLSN converts a Conduit position to a Postgres LSN.

Types ¶

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,
	keyColumn string,
	columns []string,
	out chan<- sdk.Record,
) *CDCHandler

func (*CDCHandler) Handle ¶

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

Handle is the handler function that receives all logical replication messages.

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, conn *pgx.Conn, config Config) (*CDCIterator, error)

NewCDCIterator sets up the subscription to a logical replication slot and starts a goroutine that listens to events. The goroutine will keep running until either the context is canceled or Teardown is called.

func (*CDCIterator) Ack ¶

func (i *CDCIterator) Ack(ctx context.Context, pos sdk.Position) error

Ack forwards the acknowledgment to the subscription.

func (*CDCIterator) Next ¶

func (i *CDCIterator) Next(ctx context.Context) (sdk.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.

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 Config ¶

type Config struct {
	Position        sdk.Position
	SlotName        string
	PublicationName string
	TableName       string
	KeyColumnName   string
	Columns         []string
}

Config holds configuration values for CDCIterator.

type SnapshotConfig ¶ added in v0.3.0

type SnapshotConfig struct {
	SnapshotName string
	Table        string
	Columns      []string
	KeyColumn    string
}

type SnapshotIterator ¶ added in v0.3.0

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

func NewSnapshotIterator ¶ added in v0.3.0

func NewSnapshotIterator(ctx context.Context, conn *pgx.Conn, cfg SnapshotConfig) (*SnapshotIterator, error)

func (*SnapshotIterator) Ack ¶ added in v0.3.0

func (s *SnapshotIterator) Ack(ctx context.Context, pos sdk.Position) error

Ack is a noop for snapshots

func (*SnapshotIterator) Next ¶ added in v0.3.0

func (s *SnapshotIterator) Next(ctx context.Context) (sdk.Record, error)

func (*SnapshotIterator) Teardown ¶ added in v0.3.0

func (s *SnapshotIterator) Teardown(ctx context.Context) error

Teardown attempts to gracefully teardown the iterator.

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
internal

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL