internal

Jun 14, 2024

Changes in this version

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

+ type ReadReplicationSlotResult struct

+ ConfirmedFlushLSN pglogrepl.LSN

+ Name string

+ RestartLSN pglogrepl.LSN

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

type Subscription

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

May 23, 2024

Changes in this version

+ func IsPgDuplicateErr(err error) bool

type Subscription

+ TXSnapshotID string

+ func CreateSubscription(ctx context.Context, conn *pgconn.PgConn, slotName, publication string, ...) (*Subscription, error)

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

Apr 5, 2022

Changes in this version

+ func CreatePublication(ctx context.Context, conn *pgconn.PgConn, publicationName string, ...) error

+ func DropPublication(ctx context.Context, conn *pgconn.PgConn, publicationName string, ...) error

+ type CreatePublicationOptions struct

+ AllTables bool

+ PublicationParams []string

+ Tables []string

+ type DropPublicationOptions struct

+ IfExists bool

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

+ type RelationSet struct

+ func NewRelationSet(ci *pgtype.ConnInfo) *RelationSet

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

+ func (rs *RelationSet) Get(id pgtype.OID) (*pglogrepl.RelationMessage, error)

+ func (rs *RelationSet) Values(id pgtype.OID, row *pglogrepl.TupleData) (map[string]pgtype.Value, error)

+ type Subscription struct

+ ConnConfig pgconn.Config

+ Handler Handler

+ Publication string

+ SlotName string

+ StartLSN pglogrepl.LSN

+ StatusTimeout time.Duration

+ Tables []string

+ func NewSubscription(config pgconn.Config, slotName, publication string, tables []string, ...) *Subscription

+ 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) Start(ctx context.Context) (err error)

+ func (s *Subscription) Stop()

+ func (s *Subscription) Wait(ctx context.Context) error