replication

Documentation

Index

• func DropSlot(dsn, slot string) error
• func PublicationTables(dsn string) string
• func RevertChangeSet(dsn string, changeset []Change) error
• type Change
  • func (c Change) RevertSQL() (string, []any)
• type CheckpointLoader
• type Config
• type HandleChanges
• type SlotInfo
  • func CreateSlot(dsn, slot string) (*SlotInfo, error)
• type Subscription
  • func Subscribe(cfg Config, handler HandleChanges, useNamespace bool) (*Subscription, error)
  • func (s *Subscription) DSN() string
  • func (s *Subscription) LastError() error
  • func (s *Subscription) PublicationName() string
  • func (s *Subscription) SlotName() string
  • func (s *Subscription) Start(logger *slog.Logger, loadCheckpoint CheckpointLoader, retry bool) error
  • func (s *Subscription) Stop()
• type SystemInfo
  • func Identify(dsn string) (*SystemInfo, error)

Functions

func DropSlot

func DropSlot(dsn, slot string) error

func PublicationTables

func PublicationTables(dsn string) string

func RevertChangeSet

func RevertChangeSet(dsn string, changeset []Change) error

Types

type Change

type Change struct {
	ServerTime   time.Time `json:"servertime"`
	Kind         string    `json:"kind"`
	Schema       string    `json:"schema"`
	Table        string    `json:"table"`
	ColumnNames  []string  `json:"columnnames"`
	ColumnValues []any     `json:"columnvalues"`
	OldKeys      struct {
		KeyNames  []string `json:"keynames,omitempty"`
		KeyValues []any    `json:"keyvalues,omitempty"`
	} `json:"oldkeys"`
	SQL string `json:"sql"`
}

func (Change) RevertSQL

func (c Change) RevertSQL() (string, []any)

type CheckpointLoader

type CheckpointLoader func() (pglogrepl.LSN, error)

type Config

type Config struct {
	DSN             string
	PublicationName string
	SlotName        string
	Timeout         time.Duration
	// contains filtered or unexported fields
}

type HandleChanges

type HandleChanges func(changeset []Change, currentPosition pglogrepl.LSN) error

type SlotInfo

type SlotInfo struct {
	Name         string `json:"name"`
	Plugin       string `json:"plugin"`
	Database     string `json:"database"`
	SnapshotName string `json:"snapshotName"`
	RestartLSN   string `json:"restartLSN"`
}

func CreateSlot

func CreateSlot(dsn, slot string) (*SlotInfo, error)

type Subscription

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

func Subscribe

func Subscribe(cfg Config, handler HandleChanges, useNamespace bool) (*Subscription, error)

func (*Subscription) DSN

func (s *Subscription) DSN() string

func (*Subscription) LastError

func (s *Subscription) LastError() error

func (*Subscription) PublicationName

func (s *Subscription) PublicationName() string

func (*Subscription) SlotName

func (s *Subscription) SlotName() string

func (*Subscription) Start

func (s *Subscription) Start(logger *slog.Logger, loadCheckpoint CheckpointLoader, retry bool) error

func (*Subscription) Stop

func (s *Subscription) Stop()

type SystemInfo

type SystemInfo struct {
	SystemID string `json:"systemid"`
	Timeline int32  `json:"timeline"`
	XLogPos  string `json:"xlogpos"`
	DBName   string `json:"dbname"`
}

func Identify

func Identify(dsn string) (*SystemInfo, error)