replication

Documentation

Overview

Copyright 2016 The Transicator Authors

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright 2016 The Transicator Authors

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• func DropSlot(connect, sn string) error
• type Replicator
  • func CreateReplicator(connect, sn string) (*Replicator, error)
  • func (r *Replicator) Acknowledge(lsn uint64)
  • func (r *Replicator) Changes() <-chan *common.Change
  • func (r *Replicator) SetChangeFilter(f func(*common.Change) bool)
  • func (r *Replicator) Start()
  • func (r *Replicator) State() State
  • func (r *Replicator) Stop()
  • func (r *Replicator) StopAndDrop()
• type Snapshot
  • func MakeSnapshot(snap string) (*Snapshot, error)
  • func (s *Snapshot) Contains(txid uint64) bool
• type State
  • func (i State) String() string

Functions

func DropSlot

func DropSlot(connect, sn string) error

DropSlot deletes the logical replication slot created by "Start". "connect" is a postgres URL to be passed to the "pgclient" module. "sn" is the name of the replication slot to drop.

Types

type Replicator

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

A Replicator is a client for the logical replication protocol.

func CreateReplicator

func CreateReplicator(connect, sn string) (*Replicator, error)

CreateReplicator creates a new Replicator that will use the specified URL to reach Postgres, and the specified slot name. The user must call Start in order to start it up. "connect" is a postgres URL to be passed to the "pgclient" module. "sn" is the name of the replication slot to read from. This slot will be created if it does not already exist.

func (*Replicator) Acknowledge

func (r *Replicator) Acknowledge(lsn uint64)

Acknowledge acknowledges to the server that we have committed a change, and will result in a message being sent back to the database to the same effect. It is important to periodically acknowledge changes so that the database does not have to maintain its transaction log forever. However, changes that happened before the specified LSN might still be delivered on a reconnect, so it is important that consumers of this class be prepared to handle and ignore duplicates.

func (*Replicator) Changes

func (r *Replicator) Changes() <-chan *common.Change

Changes returns a channel that can be used to wait for changes. If an error is returned, then no more changes will be forthcoming. There is only one channel per Replicator -- any kind of "broadcast" needs to be handled by the client.

func (*Replicator) SetChangeFilter

func (r *Replicator) SetChangeFilter(f func(*common.Change) bool)

SetChangeFilter supplies a function that will be called before every change is passed on to the channel. This makes it easier to write clients, especially for tests. The specified filter function will run inside a critical goroutine, so it must make its own decision without blocking. A typical use case would be to look for a particular value of a field.

func (*Replicator) Start

func (r *Replicator) Start()

Start replication. Start will succeed even if the database cannot be reached.

func (*Replicator) State

func (r *Replicator) State() State

State returns the current state of the replication.

func (*Replicator) Stop

func (r *Replicator) Stop()

Stop stops the replication process and closes the channel. It does not remove the replication slot -- for that, use "DropSlot" after the channel is closed. This method does not return until replication has been stopped.

func (*Replicator) StopAndDrop

func (r *Replicator) StopAndDrop()

StopAndDrop stops the replication process and closes the channel, and then removes the replication slot. This method does not return until replication has been stopped.

type Snapshot

type Snapshot struct {
	Xmin uint64
	Xmax uint64
	Xips map[uint64]bool
}

Snapshot represents a Postgres snapshot. As per the Postgres docs, it is constructed from a string with the values "xmin:xmax:xip1,xip2...xipn". Each represents a transaction id, which is 32 bits and may roll over in the lifetime of the database.

func MakeSnapshot

func MakeSnapshot(snap string) (*Snapshot, error)

MakeSnapshot parses the snapshot specified in the form "xmin:xmax:xip1,xip2...xipn" into a Snapshot object.

func (*Snapshot) Contains

func (s *Snapshot) Contains(txid uint64) bool

Contains tells us whether a particular transaction's changes would be visible in the specified snapshot. It tests if they are within the range xmin:xmax and or they were not in the "xips" list. If this returns true, then for a given snapshot ID and TXID, the change would be visible at the time that the snapshot was made.

type State

type State int32

State represents whether the replicator is connected or reconnecting.

const (
	Connecting State = iota
	Running
	Stopped
)

Values of replication state

func (State) String

func (i State) String() string