logrepl

package

v0.7.0 Latest Latest Go to latest Published: Apr 24, 2024 License: Apache-2.0 Imports: 11 Imported by: 0

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/conduitio/conduit-connector-postgres

Links

Open Source Insights

Documentation ¶

Index ¶

Variables
func LSNToPosition(lsn pglogrepl.LSN) sdk.Position
func PositionToLSN(pos sdk.Position) (pglogrepl.LSN, error)
type CDCHandler
- func NewCDCHandler(rs *internal.RelationSet, tableKeys map[string]string, out chan<- sdk.Record) *CDCHandler
- func (h *CDCHandler) Handle(ctx context.Context, m pglogrepl.Message, lsn pglogrepl.LSN) error
type CDCIterator
- func NewCDCIterator(ctx context.Context, connPool *pgxpool.Pool, config Config) (*CDCIterator, error)
type Config
- func (c Config) LSN() (pglogrepl.LSN, error)
type SnapshotConfig
type SnapshotIterator
- func NewSnapshotIterator(ctx context.Context, conn *pgx.Conn, cfg SnapshotConfig) (*SnapshotIterator, error)

Constants ¶

This section is empty.

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,
	tableKeys map[string]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, connPool *pgxpool.Pool, 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(_ 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
	Tables          []string
	TableKeys       map[string]string
}

Config holds configuration values for CDCIterator.

func (Config) LSN ¶ added in v0.7.0

func (c Config) LSN() (pglogrepl.LSN, error)

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(context.Context, 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