replication

Documentation

Index

• Constants
• Variables
• func AppendUint64(buf []byte, n uint64) []byte
• func SendStandbyStatusUpdate(_ context.Context, conn pg.Connection, walWritePosition uint64) error
• type ListenerContext
• type ListenerFunc
• type Message
• type Replication
  • func New(conn pg.Connection) *Replication
  • func (r *Replication) Start(publicationName, slotName string, startLSN pg.LSN) error
  • func (r *Replication) Test(ctx context.Context) error
• type Streamer
  • func NewStream(dsn string, cfg config.Config, listenerFunc ListenerFunc) Streamer
• type XLogData
  • func ParseXLogData(buf []byte) (XLogData, error)

Constants

const (
	StandbyStatusUpdateByteID = 'r'
)

Variables

var (
	ErrSlotInUse    = errors.New("replication slot in use")
	ErrNotConnected = errors.New("stream is not connected")
)

Functions

func AppendUint64

func AppendUint64(buf []byte, n uint64) []byte

func SendStandbyStatusUpdate

func SendStandbyStatusUpdate(_ context.Context, conn pg.Connection, walWritePosition uint64) error

Types

type ListenerContext

type ListenerContext struct {
	Message  any
	Relation *format.Relation
	Ack      func() error
}

type ListenerFunc

type ListenerFunc func(ctx *ListenerContext)

type Message

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

type Replication

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

func New

func New(conn pg.Connection) *Replication

func (*Replication) Start

func (r *Replication) Start(publicationName, slotName string, startLSN pg.LSN) error

func (*Replication) Test

func (r *Replication) Test(ctx context.Context) error

type Streamer

type Streamer interface {
	Connect(ctx context.Context) error
	Open(ctx context.Context) error
	Close(ctx context.Context)
	GetSystemInfo() *pg.IdentifySystemResult
	OpenFromSnapshotLSN()
}

func NewStream

func NewStream(dsn string, cfg config.Config, listenerFunc ListenerFunc) Streamer

type XLogData

type XLogData struct {
	ServerTime   time.Time
	WALData      []byte
	WALStart     pg.LSN
	ServerWALEnd pg.LSN
}

func ParseXLogData

func ParseXLogData(buf []byte) (XLogData, error)