replication

Documentation

Index

• Constants
• Variables
• type Delayer
• type Options
  • func DefaultOptions() *Options
  • func (opts *Options) Validate() error
  • func (o *Options) WithAllowTxDiscarding(allowTxDiscarding bool) *Options
  • func (o *Options) WithDelayer(delayer Delayer) *Options
  • func (o *Options) WithPrefetchTxBufferSize(prefetchTxBufferSize int) *Options
  • func (o *Options) WithPrimaryDatabase(primaryDatabase string) *Options
  • func (o *Options) WithPrimaryHost(primaryHost string) *Options
  • func (o *Options) WithPrimaryPassword(primaryPassword string) *Options
  • func (o *Options) WithPrimaryPort(primaryPort int) *Options
  • func (o *Options) WithPrimaryUsername(primaryUsername string) *Options
  • func (o *Options) WithReplicationCommitConcurrency(replicationCommitConcurrency int) *Options
  • func (o *Options) WithSkipIntegrityCheck(skipIntegrityCheck bool) *Options
  • func (o *Options) WithStreamChunkSize(streamChunkSize int) *Options
  • func (o *Options) WithWaitForIndexing(waitForIndexing bool) *Options
• type TxReplicator
  • func NewTxReplicator(uuid xid.ID, db database.DB, opts *Options, logger logger.Logger) (*TxReplicator, error)
  • func (txr *TxReplicator) Start() error
  • func (txr *TxReplicator) Stop() error

Constants

const DefaultAllowTxDiscarding = false

const DefaultChunkSize int = 64 * 1024 // 64 * 1024 64 KiB

const DefaultPrefetchTxBufferSize int = 100

const DefaultReplicationCommitConcurrency int = 10

const DefaultSkipIntegrityCheck = false

const DefaultWaitForIndexing = false

Variables

var ErrAlreadyRunning = errors.New("already running")

var ErrAlreadyStopped = errors.New("already stopped")

var ErrIllegalArguments = errors.New("illegal arguments")

var ErrInvalidOptions = fmt.Errorf("%w: invalid options", ErrIllegalArguments)

var ErrInvalidReplicationMetadata = errors.New("invalid replication metadata retrieved")

var ErrNoSynchronousReplicationOnPrimary = errors.New("primary is not running with synchronous replication")

var ErrReplicaDivergedFromPrimary = errors.New("replica diverged from primary")

Types

type Delayer

type Delayer interface {
	DelayAfter(retries int) time.Duration
}

type Options

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

func DefaultOptions

func DefaultOptions() *Options

func (*Options) Validate

func (opts *Options) Validate() error

func (*Options) WithAllowTxDiscarding

func (o *Options) WithAllowTxDiscarding(allowTxDiscarding bool) *Options

WithAllowTxDiscarding enable auto discarding of precommitted transactions

func (*Options) WithDelayer

func (o *Options) WithDelayer(delayer Delayer) *Options

WithDelayer sets delayer used to pause re-attempts

func (*Options) WithPrefetchTxBufferSize

func (o *Options) WithPrefetchTxBufferSize(prefetchTxBufferSize int) *Options

WithPrefetchTxBufferSize sets tx buffer size

func (*Options) WithPrimaryDatabase

func (o *Options) WithPrimaryDatabase(primaryDatabase string) *Options

WithPrimaryDatabase sets the source database name

func (*Options) WithPrimaryHost

func (o *Options) WithPrimaryHost(primaryHost string) *Options

WithPrimaryHost sets the source database address

func (*Options) WithPrimaryPassword

func (o *Options) WithPrimaryPassword(primaryPassword string) *Options

WithPrimaryPassword sets password used for replication

func (*Options) WithPrimaryPort

func (o *Options) WithPrimaryPort(primaryPort int) *Options

WithPrimaryPort sets the source database port

func (*Options) WithPrimaryUsername

func (o *Options) WithPrimaryUsername(primaryUsername string) *Options

WithPrimaryUsername sets username used for replication

func (*Options) WithReplicationCommitConcurrency

func (o *Options) WithReplicationCommitConcurrency(replicationCommitConcurrency int) *Options

WithReplicationCommitConcurrency sets the number of goroutines doing replication

func (*Options) WithSkipIntegrityCheck

func (o *Options) WithSkipIntegrityCheck(skipIntegrityCheck bool) *Options

WithSkipIntegrityCheck disable integrity checks when reading data during replication

func (*Options) WithStreamChunkSize

func (o *Options) WithStreamChunkSize(streamChunkSize int) *Options

WithStreamChunkSize sets streaming chunk size

func (*Options) WithWaitForIndexing

func (o *Options) WithWaitForIndexing(waitForIndexing bool) *Options

WithWaitForIndexing wait for indexing to be up to date during replication

type TxReplicator

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

func NewTxReplicator

func NewTxReplicator(uuid xid.ID, db database.DB, opts *Options, logger logger.Logger) (*TxReplicator, error)

func (*TxReplicator) Start

func (txr *TxReplicator) Start() error

func (*TxReplicator) Stop

func (txr *TxReplicator) Stop() error