binlog

Documentation

Index

• Variables
• func KeyRangeFilterFunc(unused key.KeyspaceIdType, keyrange *pb.KeyRange, ...) sendTransactionFunc
• func TablesFilterFunc(tables []string, sendReply sendTransactionFunc) sendTransactionFunc
• type BinlogStreamer
  • func NewBinlogStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, clientCharset *mproto.Charset, ...) *BinlogStreamer
  • func (bls *BinlogStreamer) Stream(ctx *sync2.ServiceContext) (err error)
• type EventStreamer
  • func NewEventStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, ...) *EventStreamer
  • func (evs *EventStreamer) Stream(ctx *sync2.ServiceContext) error
• type RegisterUpdateStreamServiceFunc
• type UpdateStream
  • func NewUpdateStream(mysqld mysqlctl.MysqlDaemon, dbname string) *UpdateStream
  • func (updateStream *UpdateStream) Disable()
  • func (updateStream *UpdateStream) Enable()
  • func (updateStream *UpdateStream) HandlePanic(err *error)
  • func (updateStream *UpdateStream) IsEnabled() bool
  • func (updateStream *UpdateStream) RegisterService()
  • func (updateStream *UpdateStream) ServeUpdateStream(position string, sendReply func(reply *proto.StreamEvent) error) (err error)
  • func (updateStream *UpdateStream) StreamKeyRange(position string, keyspaceIDType key.KeyspaceIdType, keyRange *pb.KeyRange, ...) (err error)
  • func (updateStream *UpdateStream) StreamTables(position string, tables []string, charset *mproto.Charset, ...) (err error)
• type UpdateStreamControl
• type UpdateStreamControlMock
  • func NewUpdateStreamControlMock() *UpdateStreamControlMock
  • func (m *UpdateStreamControlMock) Disable()
  • func (m *UpdateStreamControlMock) Enable()
  • func (m *UpdateStreamControlMock) IsEnabled() bool

Variables

var (

	// ErrClientEOF is returned by BinlogStreamer 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 BinlogStreamer 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 (
	STREAM_COMMENT = "/* _stream "
	SPACE          = " "
)

var RegisterUpdateStreamServices []RegisterUpdateStreamServiceFunc

RegisterUpdateStreamServices is the list of all registration callbacks to invoke

Functions

func KeyRangeFilterFunc

func KeyRangeFilterFunc(unused key.KeyspaceIdType, keyrange *pb.KeyRange, sendReply sendTransactionFunc) sendTransactionFunc

KeyRangeFilterFunc returns a function that calls sendReply only if statements in the transaction match the specified keyrange. The resulting function can be passed into the BinlogStreamer: bls.Stream(file, pos, sendTransaction) -> bls.Stream(file, pos, KeyRangeFilterFunc(keyrange, sendTransaction)) TODO(erez): Remove 'KeyspaceIdType' from here: it's no longer used.

func TablesFilterFunc

func TablesFilterFunc(tables []string, sendReply sendTransactionFunc) sendTransactionFunc

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

Types

type BinlogStreamer

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

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

func NewBinlogStreamer

func NewBinlogStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, clientCharset *mproto.Charset, startPos myproto.ReplicationPosition, sendTransaction sendTransactionFunc) *BinlogStreamer

NewBinlogStreamer creates a BinlogStreamer.

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. sendTransaction is called each time a transaction is committed or rolled back.

func (*BinlogStreamer) Stream

func (bls *BinlogStreamer) Stream(ctx *sync2.ServiceContext) (err error)

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

type EventStreamer

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

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

func NewEventStreamer

func NewEventStreamer(dbname string, mysqld mysqlctl.MysqlDaemon, startPos myproto.ReplicationPosition, sendEvent sendEventFunc) *EventStreamer

NewEventStreamer returns a new EventStreamer on top of a BinlogStreamer

func (*EventStreamer) Stream

func (evs *EventStreamer) Stream(ctx *sync2.ServiceContext) error

Stream starts streaming updates

type RegisterUpdateStreamServiceFunc

type RegisterUpdateStreamServiceFunc func(proto.UpdateStream)

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

type UpdateStream

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

UpdateStream is the real implementation of proto.UpdateStream and UpdateStreamControl

func NewUpdateStream

func NewUpdateStream(mysqld mysqlctl.MysqlDaemon, dbname string) *UpdateStream

NewUpdateStream returns a new UpdateStream object

func (*UpdateStream) Disable

func (updateStream *UpdateStream) Disable()

Disable will disallow any connection to the service

func (*UpdateStream) Enable

func (updateStream *UpdateStream) Enable()

Enable will allow connections to the service

func (*UpdateStream) HandlePanic

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

HandlePanic is part of the proto.UpdateStream interface

func (*UpdateStream) IsEnabled

func (updateStream *UpdateStream) IsEnabled() bool

IsEnabled returns true if UpdateStream is enabled

func (*UpdateStream) RegisterService

func (updateStream *UpdateStream) 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 (*UpdateStream) ServeUpdateStream

func (updateStream *UpdateStream) ServeUpdateStream(position string, sendReply func(reply *proto.StreamEvent) error) (err error)

ServeUpdateStream is part of the proto.UpdateStream interface

func (*UpdateStream) StreamKeyRange

func (updateStream *UpdateStream) StreamKeyRange(position string, keyspaceIDType key.KeyspaceIdType, keyRange *pb.KeyRange, charset *mproto.Charset, sendReply func(reply *proto.BinlogTransaction) error) (err error)

StreamKeyRange is part of the proto.UpdateStream interface

func (*UpdateStream) StreamTables

func (updateStream *UpdateStream) StreamTables(position string, tables []string, charset *mproto.Charset, sendReply func(reply *proto.BinlogTransaction) error) (err error)

StreamTables is part of the proto.UpdateStream interface

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