Merge pull request #1 from siddontang/master

update code in my fork

Author: gaojijun

.travis.yml

@@ -2,6 +2,8 @@ language: go
 
 go:
   - "1.11"
+  - "1.14"
+  - tip
 
 services:
     mysql

README.md

@@ -23,18 +23,18 @@ You can use it as a MySQL slave to sync binlog from master then do something, li
 
 ```go
 import (
-    "github.com/siddontang/go-mysql/replication"
-    "os"
+	"github.com/siddontang/go-mysql/replication"
+	"os"
 )
 // Create a binlog syncer with a unique server id, the server id must be different from other MySQL's. 
 // flavor is mysql or mariadb
 cfg := replication.BinlogSyncerConfig {
-    ServerID: 100,
-    Flavor:   "mysql",
-    Host:     "127.0.0.1",
-    Port:     3306,
-    User:     "root",
-    Password: "",
+	ServerID: 100,
+	Flavor:   "mysql",
+	Host:     "127.0.0.1",
+	Port:     3306,
+	User:     "root",
+	Password: "",
 }
 syncer := replication.NewBinlogSyncer(cfg)
 
@@ -47,23 +47,23 @@ streamer, _ := syncer.StartSync(mysql.Position{binlogFile, binlogPos})
 // the mariadb GTID set likes this "0-1-100"
 
 for {
-    ev, _ := streamer.GetEvent(context.Background())
-    // Dump event
-    ev.Dump(os.Stdout)
+	ev, _ := streamer.GetEvent(context.Background())
+	// Dump event
+	ev.Dump(os.Stdout)
 }
 
 // or we can use a timeout context
 for {
-    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-    ev, err := s.GetEvent(ctx)
-    cancel()
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	ev, err := s.GetEvent(ctx)
+	cancel()
 
-    if err == context.DeadlineExceeded {
-        // meet timeout
-        continue
-    }
+	if err == context.DeadlineExceeded {
+		// meet timeout
+		continue
+	}
 
-    ev.Dump(os.Stdout)
+	ev.Dump(os.Stdout)
 }
 ```
 
@@ -117,16 +117,16 @@ cfg.Dump.Tables = []string{"canal_test"}
 c, err := NewCanal(cfg)
 
 type MyEventHandler struct {
-    DummyEventHandler
+	DummyEventHandler
 }
 
 func (h *MyEventHandler) OnRow(e *RowsEvent) error {
-    log.Infof("%s %v\n", e.Action, e.Rows)
-    return nil
+	log.Infof("%s %v\n", e.Action, e.Rows)
+	return nil
 }
 
 func (h *MyEventHandler) String() string {
-    return "MyEventHandler"
+	return "MyEventHandler"
 }
 
 // Register a handler to handle RowsEvent
@@ -146,7 +146,7 @@ Client package supports a simple MySQL connection driver which you can use it to
 
 ```go
 import (
-    "github.com/siddontang/go-mysql/client"
+	"github.com/siddontang/go-mysql/client"
 )
 
 // Connect MySQL at 127.0.0.1:3306, with user root, an empty password and database test
@@ -155,7 +155,7 @@ conn, _ := client.Connect("127.0.0.1:3306", "root", "", "test")
 // Or to use SSL/TLS connection if MySQL server supports TLS
 //conn, _ := client.Connect("127.0.0.1:3306", "root", "", "test", func(c *Conn) {c.UseSSL(true)})
 
-// or to set your own client-side certificates for identity verification for security
+// Or to set your own client-side certificates for identity verification for security
 //tlsConfig := NewClientTLSConfig(caPem, certPem, keyPem, false, "your-server-name")
 //conn, _ := client.Connect("127.0.0.1:3306", "root", "", "test", func(c *Conn) {c.SetTLSConfig(tlsConfig)})
 
@@ -166,13 +166,29 @@ r, _ := conn.Execute(`insert into table (id, name) values (1, "abc")`)
 
 // Get last insert id
 println(r.InsertId)
+// Or affected rows count
+println(r.AffectedRows)
 
 // Select
-r, _ := conn.Execute(`select id, name from table where id = 1`)
+r, err := conn.Execute(`select id, name from table where id = 1`)
+
+// Close result for reuse memory (it's not necessary but very useful)
+defer r.Close()
 
 // Handle resultset
 v, _ := r.GetInt(0, 0)
-v, _ = r.GetIntByName(0, "id") 
+v, _ = r.GetIntByName(0, "id")
+
+// Direct access to fields
+for _, row := range r.Values {
+	for _, val := range row {
+		_ = val.Value() // interface{}
+		// or
+		if val.Type == mysql.FieldValueTypeFloat {
+			_ = val.AsFloat64() // float64
+		}
+	}   
+}
 ```
 
 Tested MySQL versions for the client include:
@@ -191,8 +207,8 @@ so that most MySQL clients should be able to connect to the Server without modif
 
 ```go
 import (
-    "github.com/siddontang/go-mysql/server"
-    "net"
+	"github.com/siddontang/go-mysql/server"
+	"net"
 )
 
 l, _ := net.Listen("tcp", "127.0.0.1:4000")
@@ -204,7 +220,7 @@ c, _ := l.Accept()
 conn, _ := server.NewConn(c, "root", "", server.EmptyHandler{})
 
 for {
-    conn.HandleCommand()
+	conn.HandleCommand()
 }
 ``` 
 
@@ -243,16 +259,16 @@ Driver is the package that you can use go-mysql with go database/sql like other
 package main
 
 import (
-    "database/sql"
+	"database/sql"
 
-    _ "github.com/siddontang/go-mysql/driver"
+	_ "github.com/siddontang/go-mysql/driver"
 )
 
 func main() {
-    // dsn format: "user:password@addr?dbname"
-    dsn := "[email protected]:3306?test"
-    db, _ := sql.Open(dsn)
-    db.Close()
+	// dsn format: "user:password@addr?dbname"
+	dsn := "[email protected]:3306?test"
+	db, _ := sql.Open(dsn)
+	db.Close()
 }
 ```
 

canal/canal.go

@@ -424,6 +424,7 @@ func (c *Canal) prepareSyncer() error {
 		ParseTime:               c.cfg.ParseTime,
 		SemiSyncEnabled:         c.cfg.SemiSyncEnabled,
 		MaxReconnectAttempts:    c.cfg.MaxReconnectAttempts,
+		DisableRetrySync:        c.cfg.DisableRetrySync,
 		TimestampStringLocation: c.cfg.TimestampStringLocation,
 		TLSConfig:               c.cfg.TLSConfig,
 	}

canal/config.go

@@ -77,10 +77,13 @@ type Config struct {
 	// SemiSyncEnabled enables semi-sync or not.
 	SemiSyncEnabled bool `toml:"semi_sync_enabled"`
 
-	// Set to change the maximum number of attempts to re-establish a broken
-	// connection
+	// maximum number of attempts to re-establish a broken connection, zero or negative number means infinite retry.
+	// this configuration will not work if DisableRetrySync is true
 	MaxReconnectAttempts int `toml:"max_reconnect_attempts"`
 
+	// whether disable re-sync for broken connection
+	DisableRetrySync bool `toml:"disable_retry_sync"`
+
 	// Set TLS config
 	TLSConfig *tls.Config
 }

client/req.go

@@ -1,5 +1,9 @@
 package client
 
+import (
+	"github.com/siddontang/go-mysql/utils"
+)
+
 func (c *Conn) writeCommand(command byte) error {
 	c.ResetSequence()
 
@@ -16,28 +20,20 @@ func (c *Conn) writeCommandBuf(command byte, arg []byte) error {
 	c.ResetSequence()
 
 	length := len(arg) + 1
-
-	data := make([]byte, length+4)
-
+	data := utils.ByteSliceGet(length + 4)
 	data[4] = command
 
 	copy(data[5:], arg)
 
-	return c.WritePacket(data)
-}
-
-func (c *Conn) writeCommandStr(command byte, arg string) error {
-	c.ResetSequence()
-
-	length := len(arg) + 1
-
-	data := make([]byte, length+4)
+	err := c.WritePacket(data)
 
-	data[4] = command
+	utils.ByteSlicePut(data)
 
-	copy(data[5:], arg)
+	return err
+}
 
-	return c.WritePacket(data)
+func (c *Conn) writeCommandStr(command byte, arg string) error {
+	return c.writeCommandBuf(command, utils.StringToByteSlice(arg))
 }
 
 func (c *Conn) writeCommandUint32(command byte, arg uint32) error {

client/resp.go

@@ -1,15 +1,15 @@
 package client
 
 import (
-	"encoding/binary"
-
 	"bytes"
 	"crypto/rsa"
 	"crypto/x509"
+	"encoding/binary"
 	"encoding/pem"
 
 	"github.com/pingcap/errors"
 	. "github.com/siddontang/go-mysql/mysql"
+	"github.com/siddontang/go-mysql/utils"
 	"github.com/siddontang/go/hack"
 )
 
@@ -212,40 +212,35 @@ func (c *Conn) readOK() (*Result, error) {
 }
 
 func (c *Conn) readResult(binary bool) (*Result, error) {
-	data, err := c.ReadPacket()
+	firstPkgBuf, err := c.ReadPacketReuseMem(utils.ByteSliceGet(16)[:0])
+	defer utils.ByteSlicePut(firstPkgBuf)
+
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
-	if data[0] == OK_HEADER {
-		return c.handleOKPacket(data)
-	} else if data[0] == ERR_HEADER {
-		return nil, c.handleErrorPacket(data)
-	} else if data[0] == LocalInFile_HEADER {
+	if firstPkgBuf[0] == OK_HEADER {
+		return c.handleOKPacket(firstPkgBuf)
+	} else if firstPkgBuf[0] == ERR_HEADER {
+		return nil, c.handleErrorPacket(append([]byte{}, firstPkgBuf...))
+	} else if firstPkgBuf[0] == LocalInFile_HEADER {
 		return nil, ErrMalformPacket
 	}
 
-	return c.readResultset(data, binary)
+	return c.readResultset(firstPkgBuf, binary)
 }
 
 func (c *Conn) readResultset(data []byte, binary bool) (*Result, error) {
-	result := &Result{
-		Status:       0,
-		InsertId:     0,
-		AffectedRows: 0,
-
-		Resultset: &Resultset{},
-	}
-
 	// column count
 	count, _, n := LengthEncodedInt(data)
 
 	if n-len(data) != 0 {
 		return nil, ErrMalformPacket
 	}
 
-	result.Fields = make([]*Field, count)
-	result.FieldNames = make(map[string]int, count)
+	result := &Result{
+		Resultset: NewResultset(int(count)),
+	}
 
 	if err := c.readResultColumns(result); err != nil {
 		return nil, errors.Trace(err)
@@ -263,10 +258,12 @@ func (c *Conn) readResultColumns(result *Result) (err error) {
 	var data []byte
 
 	for {
-		data, err = c.ReadPacket()
+		rawPkgLen := len(result.RawPkg)
+		result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
 		if err != nil {
 			return
 		}
+		data = result.RawPkg[rawPkgLen:]
 
 		// EOF Packet
 		if c.isEOFPacket(data) {
@@ -284,7 +281,10 @@ func (c *Conn) readResultColumns(result *Result) (err error) {
 			return
 		}
 
-		result.Fields[i], err = FieldData(data).Parse()
+		if result.Fields[i] == nil {
+			result.Fields[i] = &Field{}
+		}
+		err = result.Fields[i].Parse(data)
 		if err != nil {
 			return
 		}
@@ -299,11 +299,12 @@ func (c *Conn) readResultRows(result *Result, isBinary bool) (err error) {
 	var data []byte
 
 	for {
-		data, err = c.ReadPacket()
-
+		rawPkgLen := len(result.RawPkg)
+		result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
 		if err != nil {
 			return
 		}
+		data = result.RawPkg[rawPkgLen:]
 
 		// EOF Packet
 		if c.isEOFPacket(data) {
@@ -324,10 +325,14 @@ func (c *Conn) readResultRows(result *Result, isBinary bool) (err error) {
 		result.RowDatas = append(result.RowDatas, data)
 	}
 
-	result.Values = make([][]interface{}, len(result.RowDatas))
+	if cap(result.Values) < len(result.RowDatas) {
+		result.Values = make([][]FieldValue, len(result.RowDatas))
+	} else {
+		result.Values = result.Values[:len(result.RowDatas)]
+	}
 
 	for i := range result.Values {
-		result.Values[i], err = result.RowDatas[i].Parse(result.Fields, isBinary)
+		result.Values[i], err = result.RowDatas[i].Parse(result.Fields, isBinary, result.Values[i])
 
 		if err != nil {
 			return errors.Trace(err)