Adding replication protocol support to mysql server implementation

replication/binlogstreamer.go

@@ -84,11 +84,32 @@ func (s *BinlogStreamer) closeWithError(err error) {
 }
 }
 
-func newBinlogStreamer() *BinlogStreamer {
+func NewBinlogStreamer() *BinlogStreamer {
 s := new(BinlogStreamer)
 
 s.ch = make(chan *BinlogEvent, 10240)
 s.ech = make(chan error, 4)
 
 return s
 }
+
+// AddEventToStreamer adds a binlog event to the streamer. You can use it when you want to add an event to the streamer manually.
+// can be used in replication handlers
+func (s *BinlogStreamer) AddEventToStreamer(ev *BinlogEvent) error {
+select {
+case s.ch <- ev:
+return nil
+case err := <-s.ech:
+return err
+}
+}
+
+// AddErrorToStreamer adds an error to the streamer.
+func (s *BinlogStreamer) AddErrorToStreamer(err error) bool {
+select {
+case s.ech <- err:
+return true
+default:
+return false
+}
+}

replication/binlogsyncer.go

@@ -393,7 +393,7 @@ func (b *BinlogSyncer) prepare() error {
 func (b *BinlogSyncer) startDumpStream() *BinlogStreamer {
 b.running = true
 
-s := newBinlogStreamer()
+s := NewBinlogStreamer()
 
 b.wg.Add(1)
 go b.onStream(s)

server/command.go

@@ -5,6 +5,7 @@ import (
 "fmt"
 
 . "github.com/go-mysql-org/go-mysql/mysql"
+"github.com/go-mysql-org/go-mysql/replication"
 "github.com/siddontang/go/hack"
 )
 
@@ -30,6 +31,13 @@ type Handler interface {
 HandleOtherCommand(cmd byte, data []byte) error
 }
 
+type ReplicationHandler interface {
+// handle Replication command
+HandleRegisterSlave(data []byte) error
+HandleBinlogDump(pos *Position) (*replication.BinlogStreamer, error)
+HandleBinlogDumpGTID(gtidSet *MysqlGTIDSet) (*replication.BinlogStreamer, error)
+}
+
 func (c *Conn) HandleCommand() error {
 if c.Conn == nil {
 return fmt.Errorf("connection closed")
@@ -131,6 +139,37 @@ func (c *Conn) dispatch(data []byte) interface{} {
 }
 
 return eofResponse{}
+case COM_REGISTER_SLAVE:
+if h, ok := c.h.(ReplicationHandler); ok {
+return h.HandleRegisterSlave(data)
+} else {
+return c.h.HandleOtherCommand(cmd, data)
+}
+case COM_BINLOG_DUMP:
+if h, ok := c.h.(ReplicationHandler); ok {
+pos, _ := parseBinlogDump(data)
+if s, err := h.HandleBinlogDump(pos); err != nil {
+return err
+} else {
+return s
+}
+} else {
+return c.h.HandleOtherCommand(cmd, data)
+}
+case COM_BINLOG_DUMP_GTID:
+if h, ok := c.h.(ReplicationHandler); ok {
+gtidSet, err := parseBinlogDumpGTID(data)
+if err != nil {
+return err
+}
+if s, err := h.HandleBinlogDumpGTID(gtidSet); err != nil {
+return err
+} else {
+return s
+}
+} else {
+return c.h.HandleOtherCommand(cmd, data)
+}
 default:
 return c.h.HandleOtherCommand(cmd, data)
 }
@@ -139,6 +178,10 @@ func (c *Conn) dispatch(data []byte) interface{} {
 type EmptyHandler struct {
 }
 
+type EmptyReplicationHandler struct {
+EmptyHandler
+}
+
 func (h EmptyHandler) UseDB(dbName string) error {
 return nil
 }
@@ -160,6 +203,18 @@ func (h EmptyHandler) HandleStmtClose(context interface{}) error {
 return nil
 }
 
+func (h EmptyReplicationHandler) HandleRegisterSlave(data []byte) error {
+return fmt.Errorf("not supported now")
+}
+
+func (h EmptyReplicationHandler) HandleBinlogDump(pos *Position) (*replication.BinlogStreamer, error) {
+return nil, fmt.Errorf("not supported now")
+}
+
+func (h EmptyReplicationHandler) HandleBinlogDumpGTID(gtidSet *MysqlGTIDSet) (*replication.BinlogStreamer, error) {
+return nil, fmt.Errorf("not supported now")
+}
+
 func (h EmptyHandler) HandleOtherCommand(cmd byte, data []byte) error {
 return NewError(
 ER_UNKNOWN_ERROR,

server/command_test.go

@@ -2,3 +2,4 @@ package server
 
 // Ensure EmptyHandler implements Handler interface or cause compile time error
 var _ Handler = EmptyHandler{}
+var _ ReplicationHandler = EmptyReplicationHandler{}

server/replication.go

@@ -0,0 +1,21 @@
+package server
+
+import (
+"encoding/binary"
+
+"github.com/go-mysql-org/go-mysql/mysql"
+)
+
+func parseBinlogDump(data []byte) (*mysql.Position, error) {
+var p mysql.Position
+p.Pos = binary.LittleEndian.Uint32(data[0:4])
+p.Name = string(data[10:])
+
+return &p, nil
+}
+
+func parseBinlogDumpGTID(data []byte) (*mysql.MysqlGTIDSet, error) {
+lenPosName := binary.LittleEndian.Uint32(data[11:15])
+
+return mysql.DecodeMysqlGTIDSet(data[22+lenPosName:])
+}

server/resp.go

@@ -1,9 +1,11 @@
 package server
 
 import (
+"context"
 "fmt"
 
 . "github.com/go-mysql-org/go-mysql/mysql"
+"github.com/go-mysql-org/go-mysql/replication"
 )
 
 func (c *Conn) writeOK(r *Result) error {
@@ -197,6 +199,23 @@ func (c *Conn) writeFieldValues(fv []FieldValue) error {
 return c.WritePacket(data)
 }
 
+// see: https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_replication.html
+func (c *Conn) writeBinlogEvents(s *replication.BinlogStreamer) error {
+for {
+ev, err := s.GetEvent(context.Background())
+if err != nil {
+return err
+}
+data := make([]byte, 4, 4+len(ev.RawData))
+data = append(data, OK_HEADER)
+
+data = append(data, ev.RawData...)
+if err := c.WritePacket(data); err != nil {
+return err
+}
+}
+}
+
 type noResponse struct{}
 type eofResponse struct{}
 
@@ -220,6 +239,8 @@ func (c *Conn) WriteValue(value interface{}) error {
 return c.writeFieldList(v, nil)
 case []FieldValue:
 return c.writeFieldValues(v)
+case *replication.BinlogStreamer:
+return c.writeBinlogEvents(v)
 case *Stmt:
 return c.writePrepare(v)
 default: