source

Documentation

Index

• Constants
• type CDCMode
• type Config
  • func (c Config) Init() Config
  • func (Config) Parameters() map[string]config.Parameter
  • func (c Config) Validate() error
• type Iterator
• type SnapshotMode

Constants

const (
	ConfigCdcMode                = "cdcMode"
	ConfigLogreplAutoCleanup     = "logrepl.autoCleanup"
	ConfigLogreplPublicationName = "logrepl.publicationName"
	ConfigLogreplSlotName        = "logrepl.slotName"
	ConfigLogreplWithAvroSchema  = "logrepl.withAvroSchema"
	ConfigSnapshotFetchSize      = "snapshot.fetchSize"
	ConfigSnapshotMode           = "snapshotMode"
	ConfigTable                  = "table"
	ConfigTables                 = "tables"
	ConfigUrl                    = "url"
)

Types

type CDCMode

type CDCMode string

const (
	// CDCModeAuto tries to set up logical replication and falls back to long
	// polling if that is impossible.
	CDCModeAuto CDCMode = "auto"
	// CDCModeLogrepl uses logical replication to listen to changes.
	CDCModeLogrepl CDCMode = "logrepl"

	// AllTablesWildcard can be used if you'd like to listen to all tables.
	AllTablesWildcard = "*"
)

type Config

type Config struct {
	// URL is the connection string for the Postgres database.
	URL string `json:"url" validate:"required"`

	// Tables is a List of table names to read from, separated by a comma, e.g.:"table1,table2".
	// Use "*" if you'd like to listen to all tables.
	Tables []string `json:"tables"`
	// Deprecated: use `tables` instead.
	Table []string `json:"table"`

	// SnapshotMode is whether the plugin will take a snapshot of the entire table before starting cdc mode.
	SnapshotMode SnapshotMode `json:"snapshotMode" validate:"inclusion=initial|never" default:"initial"`

	// Snapshot fetcher size determines the number of rows to retrieve at a time.
	SnapshotFetchSize int `json:"snapshot.fetchSize" default:"50000"`

	// CDCMode determines how the connector should listen to changes.
	CDCMode CDCMode `json:"cdcMode" validate:"inclusion=auto|logrepl" default:"auto"`

	// LogreplPublicationName determines the publication name in case the
	// connector uses logical replication to listen to changes (see CDCMode).
	LogreplPublicationName string `json:"logrepl.publicationName" default:"conduitpub"`
	// LogreplSlotName determines the replication slot name in case the
	// connector uses logical replication to listen to changes (see CDCMode).
	LogreplSlotName string `json:"logrepl.slotName" default:"conduitslot"`

	// LogreplAutoCleanup determines if the replication slot and publication should be
	// removed when the connector is deleted.
	LogreplAutoCleanup bool `json:"logrepl.autoCleanup" default:"true"`

	// WithAvroSchema determines whether the connector should attach an avro schema on each
	// record.
	WithAvroSchema bool `json:"logrepl.withAvroSchema" default:"false"`
}

func (Config) Init

func (c Config) Init() Config

Init sets the desired value on Tables while Table is being deprecated.

func (Config) Parameters

func (Config) Parameters() map[string]config.Parameter

func (Config) Validate

func (c Config) Validate() error

Validate validates the provided config values.

type Iterator

type Iterator interface {
	// Next takes and returns the next record from the queue. Next is allowed to
	// block until either a record is available or the context gets canceled.
	Next(context.Context) (opencdc.Record, error)
	// Ack signals that a record at a specific position was successfully
	// processed.
	Ack(context.Context, opencdc.Position) error
	// Teardown attempts to gracefully teardown the iterator.
	Teardown(context.Context) error
}

Iterator is an object that can iterate over a queue of records.

type SnapshotMode

type SnapshotMode string

const (
	// SnapshotModeInitial creates a snapshot in the first run of the pipeline.
	SnapshotModeInitial SnapshotMode = "initial"
	// SnapshotModeNever skips snapshot creation altogether.
	SnapshotModeNever SnapshotMode = "never"
)