binlog

Documentation

Index

• Variables
• func KeyRangeFilterFunc(keyrange *topodatapb.KeyRange, ...) sendTransactionFunc
• func TablesFilterFunc(tables []string, callback func(*binlogdatapb.BinlogTransaction) error) sendTransactionFunc
• type EventStreamer
  • func NewEventStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, se *schema.Engine, ...) *EventStreamer
  • func (evs *EventStreamer) Stream(ctx context.Context) error
• type FullBinlogStatement
• type RegisterUpdateStreamServiceFunc
• type StreamList
  • func (sl *StreamList) Add(c context.CancelFunc) int
  • func (sl *StreamList) Delete(i int)
  • func (sl *StreamList) Init()
  • func (sl *StreamList) Stop()
• type Streamer
  • func NewStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, se *schema.Engine, ...) *Streamer
  • func (bls *Streamer) Stream(ctx context.Context) (err error)
• type UpdateStream
• type UpdateStreamControl
• type UpdateStreamControlMock
  • func NewUpdateStreamControlMock() *UpdateStreamControlMock
  • func (m *UpdateStreamControlMock) Disable()
  • func (m *UpdateStreamControlMock) Enable()
  • func (m *UpdateStreamControlMock) IsEnabled() bool
• type UpdateStreamImpl
  • func NewUpdateStream(ts topo.Server, keyspace string, cell string, mysqld mysqlctl.MysqlDaemon, ...) *UpdateStreamImpl
  • func (updateStream *UpdateStreamImpl) Disable()
  • func (updateStream *UpdateStreamImpl) Enable()
  • func (updateStream *UpdateStreamImpl) HandlePanic(err *error)
  • func (updateStream *UpdateStreamImpl) IsEnabled() bool
  • func (updateStream *UpdateStreamImpl) RegisterService()
  • func (updateStream *UpdateStreamImpl) StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, ...) (err error)
  • func (updateStream *UpdateStreamImpl) StreamTables(ctx context.Context, position string, tables []string, ...) (err error)

Variables

var (

	// ErrClientEOF is returned by Streamer if the stream ended because the
	// consumer of the stream indicated it doesn't want any more events.
	ErrClientEOF = fmt.Errorf("binlog stream consumer ended the reply stream")
	// ErrServerEOF is returned by Streamer if the stream ended because the
	// connection to the mysqld server was lost, or the stream was terminated by
	// mysqld.
	ErrServerEOF = fmt.Errorf("binlog stream connection was closed by mysqld")
)

var RegisterUpdateStreamServices []RegisterUpdateStreamServiceFunc

RegisterUpdateStreamServices is the list of all registration callbacks to invoke

Functions

func KeyRangeFilterFunc

func KeyRangeFilterFunc(keyrange *topodatapb.KeyRange, callback func(*binlogdatapb.BinlogTransaction) error) sendTransactionFunc

KeyRangeFilterFunc returns a function that calls callback only if statements in the transaction match the specified keyrange. The resulting function can be passed into the Streamer: bls.Stream(file, pos, sendTransaction) -> bls.Stream(file, pos, KeyRangeFilterFunc(keyrange, sendTransaction))

func TablesFilterFunc

func TablesFilterFunc(tables []string, callback func(*binlogdatapb.BinlogTransaction) error) sendTransactionFunc

TablesFilterFunc returns a function that calls callback only if statements in the transaction match the specified tables. The resulting function can be passed into the Streamer: bls.Stream(file, pos, sendTransaction) -> bls.Stream(file, pos, TablesFilterFunc(sendTransaction))

Types

type EventStreamer

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

EventStreamer is an adapter on top of a binlog Streamer that convert the events into StreamEvent objects.

func NewEventStreamer

func NewEventStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, se *schema.Engine, startPos replication.Position, timestamp int64, sendEvent sendEventFunc) *EventStreamer

NewEventStreamer returns a new EventStreamer on top of a Streamer

func (*EventStreamer) Stream

func (evs *EventStreamer) Stream(ctx context.Context) error

Stream starts streaming updates

type FullBinlogStatement

type FullBinlogStatement struct {
	Statement  *binlogdatapb.BinlogTransaction_Statement
	Table      string
	KeyspaceID []byte
	PKNames    []*querypb.Field
	PKValues   []sqltypes.Value
}

FullBinlogStatement has all the information we can gather for an event. Some fields are only set if asked for, and if RBR is used. Otherwise we'll revert back to using the SQL comments, for SBR.

type RegisterUpdateStreamServiceFunc

type RegisterUpdateStreamServiceFunc func(UpdateStream)

RegisterUpdateStreamServiceFunc is the type to use for delayed registration of RPC servers until we have all the objects

type StreamList

type StreamList struct {
	sync.Mutex
	// contains filtered or unexported fields
}

StreamList is a map of context.CancelFunc to mass-interrupt ongoing calls.

func (*StreamList) Add

func (sl *StreamList) Add(c context.CancelFunc) int

Add adds a CancelFunc to the map.

func (*StreamList) Delete

func (sl *StreamList) Delete(i int)

Delete removes a CancelFunc from the list.

func (*StreamList) Init

func (sl *StreamList) Init()

Init must be called before using the list.

func (*StreamList) Stop

func (sl *StreamList) Stop()

Stop stops all the current streams.

type Streamer

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

Streamer streams binlog events from MySQL by connecting as a slave. A Streamer should only be used once. To start another stream, call NewStreamer() again.

func NewStreamer

func NewStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, se *schema.Engine, clientCharset *binlogdatapb.Charset, startPos replication.Position, timestamp int64, sendTransaction sendTransactionFunc) *Streamer

NewStreamer creates a binlog Streamer.

dbname specifes the database to stream events for. mysqld is the local instance of mysqlctl.Mysqld. charset is the default character set on the BinlogPlayer side. startPos is the position to start streaming at. Incompatible with timestamp. timestamp is the timestamp to start streaming at. Incompatible with startPos. sendTransaction is called each time a transaction is committed or rolled back.

func (*Streamer) Stream

func (bls *Streamer) Stream(ctx context.Context) (err error)

Stream starts streaming binlog events using the settings from NewStreamer().

type UpdateStream

type UpdateStream interface {
	// StreamKeyRange streams events related to a KeyRange only
	StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) error

	// StreamTables streams events related to a set of Tables only
	StreamTables(ctx context.Context, position string, tables []string, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) error

	// HandlePanic should be called in a defer,
	// first thing in the RPC implementation.
	HandlePanic(*error)
}

UpdateStream is the interface for the binlog server

type UpdateStreamControl

type UpdateStreamControl interface {
	// Enable will allow any new RPC calls
	Enable()

	// Disable will interrupt all current calls, and disallow any new call
	Disable()

	// IsEnabled returns true iff the service is enabled
	IsEnabled() bool
}

UpdateStreamControl is the interface an UpdateStream service implements to bring it up or down.

type UpdateStreamControlMock

type UpdateStreamControlMock struct {
	sync.Mutex
	// contains filtered or unexported fields
}

UpdateStreamControlMock is an implementation of UpdateStreamControl to be used in tests

func NewUpdateStreamControlMock

func NewUpdateStreamControlMock() *UpdateStreamControlMock

NewUpdateStreamControlMock creates a new UpdateStreamControlMock

func (*UpdateStreamControlMock) Disable

func (m *UpdateStreamControlMock) Disable()

Disable is part of UpdateStreamControl

func (*UpdateStreamControlMock) Enable

func (m *UpdateStreamControlMock) Enable()

Enable is part of UpdateStreamControl

func (*UpdateStreamControlMock) IsEnabled

func (m *UpdateStreamControlMock) IsEnabled() bool

IsEnabled is part of UpdateStreamControl

type UpdateStreamImpl

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

UpdateStreamImpl is the real implementation of UpdateStream and UpdateStreamControl

func NewUpdateStream

func NewUpdateStream(ts topo.Server, keyspace string, cell string, mysqld mysqlctl.MysqlDaemon, se *schema.Engine, dbname string) *UpdateStreamImpl

NewUpdateStream returns a new UpdateStreamImpl object

func (*UpdateStreamImpl) Disable

func (updateStream *UpdateStreamImpl) Disable()

Disable will disallow any connection to the service

func (*UpdateStreamImpl) Enable

func (updateStream *UpdateStreamImpl) Enable()

Enable will allow connections to the service

func (*UpdateStreamImpl) HandlePanic

func (updateStream *UpdateStreamImpl) HandlePanic(err *error)

HandlePanic is part of the UpdateStream interface

func (*UpdateStreamImpl) IsEnabled

func (updateStream *UpdateStreamImpl) IsEnabled() bool

IsEnabled returns true if UpdateStreamImpl is enabled

func (*UpdateStreamImpl) RegisterService

func (updateStream *UpdateStreamImpl) RegisterService()

RegisterService needs to be called to publish stats, and to start listening to clients. Only once instance can call this in a process.

func (*UpdateStreamImpl) StreamKeyRange

func (updateStream *UpdateStreamImpl) StreamKeyRange(ctx context.Context, position string, keyRange *topodatapb.KeyRange, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) (err error)

StreamKeyRange is part of the UpdateStream interface

func (*UpdateStreamImpl) StreamTables

func (updateStream *UpdateStreamImpl) StreamTables(ctx context.Context, position string, tables []string, charset *binlogdatapb.Charset, callback func(trans *binlogdatapb.BinlogTransaction) error) (err error)

StreamTables is part of the UpdateStream interface