Adding replication protocol support to mysql server implementation

replication/binlogstreamer.go

@@ -84,11 +84,24 @@ 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
 }
+
+func (s *BinlogStreamer) AddEventToStreamer(ev *BinlogEvent) error {
+	select {
+	case s.ch <- ev:
+		return nil
+	case err := <-s.ech:
+		return err
+	}
+}
+
+func (s *BinlogStreamer) AddErrorToStreamer(err error) {
+	s.ech <- err
+}

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, s *replication.BinlogStreamer)
+	HandleBinlogDumpGTID(gtidSet *MysqlGTIDSet, s *replication.BinlogStreamer)
+}
+
 func (c *Conn) HandleCommand() error {
 	if c.Conn == nil {
 		return fmt.Errorf("connection closed")
@@ -131,6 +139,39 @@ func (c *Conn) dispatch(data []byte) interface{} {
 		}
 
 		return eofResponse{}
+	case COM_REGISTER_SLAVE:
+		if h, ok := c.h.(ReplicationHandler); ok {
+			if err := h.HandleRegisterSlave(data); err != nil {
+				return err
+			}
+			return nil
+		} else {
+			return fmt.Errorf("the handler does not support replication protocol, use ReplicationHandler instead")
+		}
+	case COM_BINLOG_DUMP:
+		if h, ok := c.h.(ReplicationHandler); ok {
+			pos, _ := parseBinlogDump(data)
+			s := replication.NewBinlogStreamer()
+			go h.HandleBinlogDump(pos, s)
+
+			return s
+		} else {
+			return fmt.Errorf("the handler does not support replication protocol, use ReplicationHandler instead")
+		}
+	case COM_BINLOG_DUMP_GTID:
+		if h, ok := c.h.(ReplicationHandler); ok {
+			gtidSet, err := parseBinlogDumpGTID(data)
+			if err != nil {
+				return err
+			}
+
+			s := replication.NewBinlogStreamer()
+			go h.HandleBinlogDumpGTID(gtidSet, s)
+
+			return s
+		} else {
+			return fmt.Errorf("the handler does not support replication protocol, use ReplicationHandler instead")
+		}
 	default:
 		return c.h.HandleOtherCommand(cmd, data)
 	}
@@ -139,6 +180,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 +205,16 @@ 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, r *replication.BinlogStreamer) {
+}
+
+func (h EmptyReplicationHandler) HandleBinlogDumpGTID(gtidSet *MysqlGTIDSet, r *replication.BinlogStreamer) {
+}
+
 func (h EmptyHandler) HandleOtherCommand(cmd byte, data []byte) error {
 	return NewError(
 		ER_UNKNOWN_ERROR,

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,12 @@
 package server
 
 import (
+	"context"
 	"fmt"
+	"time"
 
 	. "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 +200,25 @@ func (c *Conn) writeFieldValues(fv []FieldValue) error {
 	return c.WritePacket(data)
 }
 
+func (c *Conn) writeBinlogEvents(s *replication.BinlogStreamer) error {
+	for {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		ev, err := s.GetEvent(ctx)
+		cancel()
+
+		if err == context.DeadlineExceeded {
+			continue
+		}
+		data := make([]byte, 4, 32)
+		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 +242,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: