internal

Documentation

Index

• Variables
• func CreatePublication(ctx context.Context, conn *pgxpool.Pool, name string, ...) error
• func DropPublication(ctx context.Context, conn *pgxpool.Pool, name string, ...) error
• func IsPgDuplicateErr(err error) bool
• type CreatePublicationOptions
• type DropPublicationOptions
• type Handler
• type ReadReplicationSlotResult
  • func ReadReplicationSlot(ctx context.Context, conn *pgxpool.Pool, name string) (ReadReplicationSlotResult, error)
• type RelationSet
  • func NewRelationSet() *RelationSet
  • func (rs *RelationSet) Add(r *pglogrepl.RelationMessage)
  • func (rs *RelationSet) Get(id uint32) (*pglogrepl.RelationMessage, error)
  • func (rs *RelationSet) Values(id uint32, row *pglogrepl.TupleData) (map[string]any, error)
• type Subscription
  • func CreateSubscription(ctx context.Context, pool *pgxpool.Pool, slotName, publication string, ...) (*Subscription, error)
  • func (s *Subscription) Ack(lsn pglogrepl.LSN)
  • func (s *Subscription) Done() <-chan struct{}
  • func (s *Subscription) Err() error
  • func (s *Subscription) Ready() <-chan struct{}
  • func (s *Subscription) Run(ctx context.Context) error
  • func (s *Subscription) Stop()
  • func (s *Subscription) Teardown(ctx context.Context) error
  • func (s *Subscription) Wait(ctx context.Context, timeout time.Duration) error

Variables

var ErrMissingSlot = errors.New("replication slot missing")

Functions

func CreatePublication

func CreatePublication(ctx context.Context, conn *pgxpool.Pool, name string, opts CreatePublicationOptions) error

CreatePublication creates a publication.

func DropPublication

func DropPublication(ctx context.Context, conn *pgxpool.Pool, name string, opts DropPublicationOptions) error

DropPublication drops a publication.

func IsPgDuplicateErr

func IsPgDuplicateErr(err error) bool

Types

type CreatePublicationOptions

type CreatePublicationOptions struct {
	Tables            []string
	PublicationParams []string
}

CreatePublicationOptions contains additional options for creating a publication. If AllTables and Tables are both true and not empty at the same time, publication creation will fail.

type DropPublicationOptions

type DropPublicationOptions struct {
	IfExists bool
}

DropPublicationOptions contains additional options for dropping a publication.

type Handler

type Handler func(context.Context, pglogrepl.Message, pglogrepl.LSN) (pglogrepl.LSN, error)

type ReadReplicationSlotResult

type ReadReplicationSlotResult struct {
	Name              string
	ConfirmedFlushLSN pglogrepl.LSN
	RestartLSN        pglogrepl.LSN
}

func ReadReplicationSlot

func ReadReplicationSlot(ctx context.Context, conn *pgxpool.Pool, name string) (ReadReplicationSlotResult, error)

ReadReplicationSlot returns state of an existing replication slot.

type RelationSet

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

RelationSet can be used to build a cache of relations returned by logical replication.

func NewRelationSet

func NewRelationSet() *RelationSet

NewRelationSet creates a new relation set.

func (*RelationSet) Add

func (rs *RelationSet) Add(r *pglogrepl.RelationMessage)

func (*RelationSet) Get

func (rs *RelationSet) Get(id uint32) (*pglogrepl.RelationMessage, error)

func (*RelationSet) Values

func (rs *RelationSet) Values(id uint32, row *pglogrepl.TupleData) (map[string]any, error)

type Subscription

type Subscription struct {
	SlotName      string
	Publication   string
	Tables        []string
	StartLSN      pglogrepl.LSN
	Handler       Handler
	StatusTimeout time.Duration
	TXSnapshotID  string
	// contains filtered or unexported fields
}

Subscription manages a subscription to a logical replication slot.

func CreateSubscription

func CreateSubscription(
	ctx context.Context,
	pool *pgxpool.Pool,
	slotName,
	publication string,
	tables []string,
	startLSN pglogrepl.LSN,
	h Handler,
) (*Subscription, error)

CreateSubscription initializes the logical replication subscriber by creating the replication slot.

func (*Subscription) Ack

func (s *Subscription) Ack(lsn pglogrepl.LSN)

Ack stores the LSN as flushed. Next time WAL positions are flushed, Postgres will know it can purge WAL logs up to this LSN.

func (*Subscription) Done

func (s *Subscription) Done() <-chan struct{}

Done returns a channel that is closed when the subscription is done.

func (*Subscription) Err

func (s *Subscription) Err() error

Err returns an error that might have happened when the subscription stopped running.

func (*Subscription) Ready

func (s *Subscription) Ready() <-chan struct{}

Ready returns a channel that is closed when the subscription is ready and receiving messages.

func (*Subscription) Run

func (s *Subscription) Run(ctx context.Context) error

Run logical replication listener and block until error or ctx is canceled.

func (*Subscription) Stop

func (s *Subscription) Stop()

Stop signals to the subscription it should stop. Call Wait to block until the subscription actually stops running.

func (*Subscription) Teardown

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

func (*Subscription) Wait

func (s *Subscription) Wait(ctx context.Context, timeout time.Duration) error

Wait will block until the subscription is stopped. If the context gets cancelled in the meantime it will return the context error, otherwise nil is returned.