replicator

package

Go to main page

Versions in this module

v0

v0.0.15

Aug 24, 2025

Changes in this version

+ const DefaultSchema

type Config

+ RetryConfig utils.RetryConfig

+ func (c Config) GetRetryConfig() utils.RetryConfig

type PostgresReplicationConnection

+ func (rc *PostgresReplicationConnection) IsHealthy(ctx context.Context) bool

+ func (rc *PostgresReplicationConnection) Reconnect(ctx context.Context) error

type ReplicationConnection

+ IsHealthy func(ctx context.Context) bool

+ Reconnect func(ctx context.Context) error

v0.0.14

Aug 14, 2025

v0.0.13

Feb 6, 2025

v0.0.12

Nov 19, 2024

Changes in this version

+ var ErrReplicatorAlreadyStarted = errors.New("replicator already started")

+ var ErrReplicatorAlreadyStopped = errors.New("replicator already stopped")

+ var ErrReplicatorNotStarted = errors.New("replicator not started")

+ type BaseFactory struct

+ func (f *BaseFactory) CreateConnections(config Config) (ReplicationConnection, StandardConnection, error)

type BaseReplicator

+ DDLReplicator *DDLReplicator

+ func (r *BaseReplicator) CurrentTxBuffer() []utils.CDCMessage

+ func (r *BaseReplicator) SetCurrentTxBuffer(messages []utils.CDCMessage)

+ func (r *BaseReplicator) Start(ctx context.Context) error

+ func (r *BaseReplicator) Stop(ctx context.Context) error

type CopyAndStreamReplicator

+ func NewCopyAndStreamReplicator(base *BaseReplicator, maxWorkers int, copyOnly bool) *CopyAndStreamReplicator

+ func (r *CopyAndStreamReplicator) Start(ctx context.Context) error

+ func (r *CopyAndStreamReplicator) Stop(ctx context.Context) error

+ type CopyAndStreamReplicatorFactory struct

+ CopyOnly bool

+ MaxCopyWorkersPerTable int

+ func (f *CopyAndStreamReplicatorFactory) CreateReplicator(config Config, natsClient NATSClient) (Replicator, error)

+ type Factory interface

+ CreateReplicator func(config Config, natsClient NATSClient) (Replicator, error)

type Replicator

+ Start func(ctx context.Context) error

+ Stop func(ctx context.Context) error

type StreamReplicator

+ func NewStreamReplicator(base *BaseReplicator) *StreamReplicator

+ func (r *StreamReplicator) Start(ctx context.Context) error

+ func (r *StreamReplicator) Stop(ctx context.Context) error

+ type StreamReplicatorFactory struct

+ func (f *StreamReplicatorFactory) CreateReplicator(config Config, natsClient NATSClient) (Replicator, error)

v0.0.11

Nov 16, 2024

Changes in this version

+ func GeneratePublicationName(group string) string

+ func InitializeOIDMap(ctx context.Context, conn StandardConnection) error

+ type BaseReplicator struct

+ Config Config

+ LastLSN pglogrepl.LSN

+ Logger zerolog.Logger

+ NATSClient NATSClient

+ Relations map[uint32]*pglogrepl.RelationMessage

+ ReplicationConn ReplicationConnection

+ StandardConn StandardConnection

+ TableDetails map[string][]string

+ TableReplicationKeys map[string]utils.ReplicationKey

+ func NewBaseReplicator(config Config, replicationConn ReplicationConnection, ...) *BaseReplicator

+ func (r *BaseReplicator) AddPrimaryKeyInfo(message *utils.CDCMessage, table string)

+ func (r *BaseReplicator) CheckReplicationSlotExists(slotName string) (bool, error)

+ func (r *BaseReplicator) CheckReplicationSlotStatus(ctx context.Context) error

+ func (r *BaseReplicator) CreatePublication() error

+ func (r *BaseReplicator) CreateReplicationSlot(ctx context.Context) error

+ func (r *BaseReplicator) GetConfiguredTables(ctx context.Context) ([]string, error)

+ func (r *BaseReplicator) GetLastState() (pglogrepl.LSN, error)

+ func (r *BaseReplicator) GracefulShutdown(ctx context.Context) error

+ func (r *BaseReplicator) HandleBeginMessage(_ *pglogrepl.BeginMessage) error

+ func (r *BaseReplicator) HandleCommitMessage(msg *pglogrepl.CommitMessage) error

+ func (r *BaseReplicator) HandleDeleteMessage(msg *pglogrepl.DeleteMessage, lsn pglogrepl.LSN) error

+ func (r *BaseReplicator) HandleInsertMessage(msg *pglogrepl.InsertMessage, lsn pglogrepl.LSN) error

+ func (r *BaseReplicator) HandleUpdateMessage(msg *pglogrepl.UpdateMessage, lsn pglogrepl.LSN) error

+ func (r *BaseReplicator) InitializePrimaryKeyInfo() error

+ func (r *BaseReplicator) ProcessNextMessage(ctx context.Context, lastStatusUpdate *time.Time, ...) error

+ func (r *BaseReplicator) PublishToNATS(data utils.CDCMessage) error

+ func (r *BaseReplicator) SaveState(lsn pglogrepl.LSN) error

+ func (r *BaseReplicator) SendStandbyStatusUpdate(ctx context.Context) error

+ func (r *BaseReplicator) StartReplicationFromLSN(ctx context.Context, startLSN pglogrepl.LSN, stopChan <-chan struct{}) error

+ func (r *BaseReplicator) StreamChanges(ctx context.Context, stopChan <-chan struct{}) error

+ type Buffer struct

+ func NewBuffer(maxRows int, flushTimeout time.Duration) *Buffer

+ func (b *Buffer) Add(item interface{}) bool

+ func (b *Buffer) Flush() []interface{}

+ type Config struct

+ Database string

+ Group string

+ Host string

+ Password string

+ Port uint16

+ Schema string

+ Tables []string

+ TrackDDL bool

+ User string

+ func (c Config) ConnectionString() string

+ type CopyAndStreamReplicator struct

+ CopyOnly bool

+ DDLReplicator DDLReplicator

+ MaxCopyWorkersPerTable int

+ func (r *CopyAndStreamReplicator) CopyTable(ctx context.Context, tableName, snapshotID string) error

+ func (r *CopyAndStreamReplicator) CopyTableRange(ctx context.Context, tableName string, startPage, endPage uint32, ...) (int64, error)

+ func (r *CopyAndStreamReplicator) CopyTableWorker(ctx context.Context, wg *sync.WaitGroup, errChan chan<- error, ...)

+ func (r *CopyAndStreamReplicator) CopyTables(ctx context.Context, tables []string, snapshotID string) error

+ func (r *CopyAndStreamReplicator) NewBaseReplicator() *BaseReplicator

+ func (r *CopyAndStreamReplicator) ParallelCopy(ctx context.Context) error

+ func (r *CopyAndStreamReplicator) StartReplication() error

+ type DDLReplicator struct

+ BaseRepl *BaseReplicator

+ Config Config

+ DDLConn StandardConnection

+ func NewDDLReplicator(config Config, BaseRepl *BaseReplicator, ddlConn StandardConnection) (*DDLReplicator, error)

+ func (d *DDLReplicator) Close(ctx context.Context) error

+ func (d *DDLReplicator) HasPendingDDLEvents(ctx context.Context) (bool, error)

+ func (d *DDLReplicator) ProcessDDLEvents(ctx context.Context) error

+ func (d *DDLReplicator) SetupDDLTracking(ctx context.Context) error

+ func (d *DDLReplicator) Shutdown(ctx context.Context) error

+ func (d *DDLReplicator) StartDDLReplication(ctx context.Context)

+ type NATSClient interface

+ Close func() error

+ GetState func() (pgflonats.State, error)

+ JetStream func() nats.JetStreamContext

+ PublishMessage func(subject string, data []byte) error

+ SaveState func(state pgflonats.State) error

+ type PgxPoolConn interface

+ BeginTx func(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)

+ Exec func(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)

+ Query func(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

+ QueryRow func(ctx context.Context, sql string, args ...interface{}) pgx.Row

+ Release func()

+ type PgxPoolConnWrapper struct

+ type PostgresReplicationConnection struct

+ Config Config

+ Conn *pgconn.PgConn

+ func (rc *PostgresReplicationConnection) Close(ctx context.Context) error

+ func (rc *PostgresReplicationConnection) Connect(ctx context.Context) error

+ func (rc *PostgresReplicationConnection) CreateReplicationSlot(ctx context.Context, slotName string) (pglogrepl.CreateReplicationSlotResult, error)

+ func (rc *PostgresReplicationConnection) ReceiveMessage(ctx context.Context) (pgproto3.BackendMessage, error)

+ func (rc *PostgresReplicationConnection) SendStandbyStatusUpdate(ctx context.Context, status pglogrepl.StandbyStatusUpdate) error

+ func (rc *PostgresReplicationConnection) StartReplication(ctx context.Context, slotName string, startLSN pglogrepl.LSN, ...) error

+ type ReplicationConnection interface

+ Close func(ctx context.Context) error

+ Connect func(ctx context.Context) error

+ CreateReplicationSlot func(ctx context.Context, slotName string) (pglogrepl.CreateReplicationSlotResult, error)

+ ReceiveMessage func(ctx context.Context) (pgproto3.BackendMessage, error)

+ SendStandbyStatusUpdate func(ctx context.Context, status pglogrepl.StandbyStatusUpdate) error

+ StartReplication func(ctx context.Context, slotName string, startLSN pglogrepl.LSN, ...) error

+ func NewReplicationConnection(config Config) ReplicationConnection

+ type ReplicationError struct

+ Err error

+ Op string

+ func (e *ReplicationError) Error() string

+ type Replicator interface

+ StartReplication func() error

+ func NewReplicator(config Config, natsClient *pgflonats.NATSClient, copyAndStream bool, ...) (Replicator, error)

+ type StandardConnection interface

+ Acquire func(ctx context.Context) (PgxPoolConn, error)

+ BeginTx func(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)

+ Close func(ctx context.Context) error

+ Connect func(ctx context.Context) error

+ Exec func(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)

+ Query func(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

+ QueryRow func(ctx context.Context, sql string, args ...interface{}) pgx.Row

+ type StandardConnectionImpl struct

+ func NewStandardConnection(config Config) (*StandardConnectionImpl, error)

+ func (s *StandardConnectionImpl) Acquire(ctx context.Context) (PgxPoolConn, error)

+ func (s *StandardConnectionImpl) BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)

+ func (s *StandardConnectionImpl) Close(_ context.Context) error

+ func (s *StandardConnectionImpl) Connect(ctx context.Context) error

+ func (s *StandardConnectionImpl) Exec(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)

+ func (s *StandardConnectionImpl) Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

+ func (s *StandardConnectionImpl) QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row

+ type StreamReplicator struct

+ DDLReplicator DDLReplicator

+ func (r *StreamReplicator) StartReplication() error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL