Fixxed bad packet seq

binlog/binlog.go

@@ -12,7 +12,7 @@ func (c *Conn) registerAsSlave() error {
 		User:     "",
 		Password: "",
 		Port:     0,
-		ReplRank: 1,
+		ReplRank: 0,
 		MasterId: 0,
 	}
 

binlog/commands.go

@@ -21,9 +21,9 @@ func (c *Conn) writeBinlogRegisterSlaveCommand(brsc *BinlogRegisterSlaveCommand)
 	c.putString(TypeLenEncString, brsc.Hostname)
 	c.putString(TypeLenEncString, brsc.User)
 	c.putString(TypeLenEncString, brsc.Password)
-	c.putInt(TypeLenEncInt, brsc.Port, 2)
-	c.putInt(TypeLenEncInt, brsc.ReplRank, 4)
-	c.putInt(TypeLenEncInt, brsc.MasterId, 4)
+	c.putInt(TypeFixedInt, brsc.Port, 2)
+	c.putInt(TypeFixedInt, brsc.ReplRank, 4)
+	c.putInt(TypeFixedInt, brsc.MasterId, 4)
 
 	if c.Flush() != nil {
 		return c.Flush()

binlog/connection.go

@@ -184,6 +184,18 @@ func (d Driver) Open(dsn string) (driver.Conn, error) {
 		return nil, err
 	}
 
+	c.sequenceId = 0
+
+	_, err = c.readPacket()
+	if err != nil {
+		return nil, err
+	}
+
+	err = c.startBinlogStream()
+	if err != nil {
+		return nil, err
+	}
+
 	_, err = c.readPacket()
 	if err != nil {
 		return nil, err
@@ -198,8 +210,6 @@ func (c *Conn) readPacket() (interface{}, error) {
 		return nil, err
 	}
 
-	c.sequenceId++
-
 	var res interface{}
 
 	switch ph.Status {
@@ -617,6 +627,7 @@ func (c *Conn) Flush() error {
 	}
 
 	c.writeBuf = c.addHeader()
+	//fmt.Printf("c.writeBuf = %x\n", c.writeBuf.Bytes())
 	_, _ = c.buffer.Write(c.writeBuf.Bytes())
 	if c.buffer.Flush() != nil {
 		return c.buffer.Flush()

main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"database/sql"
 	"fmt"
-	"os"
 
 	_ "github.com/macinjosh/mysql-binlog-filter/binlog"
 )
@@ -12,12 +11,10 @@ func main() {
 	conn, err := sql.Open("mysql-binlog", "config.json")
 	if err != nil {
 		fmt.Printf("Open Error: %+v\n", err)
-		os.Exit(1)
 	}
 
 	err = conn.Ping()
 	if err != nil {
 		fmt.Printf("%+v\n", err)
-		os.Exit(1)
 	}
 }