Merge branch 'master' of github.com:go-mysql-org/go-mysql into improveExample

Author: skoef

.github/workflows/ci.yml

@@ -5,7 +5,7 @@ jobs:
  test:
  strategy:
  matrix:
- go: [ 1.16, 1.15 ]
+ go: [ 1.18, 1.17, 1.16, 1.15 ]
  name: Tests Go ${{ matrix.go }}
  runs-on: ubuntu-18.04
 

CHANGELOG.md

@@ -1,3 +1,26 @@
+### Tag v1.5.0 (2022-04-30)
+* Replacing github.com/satori/go.uuid with github.com/google/uuid. [#690](https://github.com/go-mysql-org/go-mysql/pull/690) ([martinhrvn](https://github.com/martinhrvn))
+* Fix dump gtid regexp. [#688](https://github.com/go-mysql-org/go-mysql/pull/688) ([yrbb](https://github.com/yrbb))
+* Utilise byteSlice as a part of struct to avoid allocations. [#686](https://github.com/go-mysql-org/go-mysql/pull/686) ([moredure](https://github.com/moredure))
+* Update bytes_buffer_pool.go. [#685](https://github.com/go-mysql-org/go-mysql/pull/685) ([moredure](https://github.com/moredure))
+* Add go1.18 to the test matrix. [#684](https://github.com/go-mysql-org/go-mysql/pull/684) ([moredure](https://github.com/moredure))
+* Add 1.17 version into the test matrix. [#683](https://github.com/go-mysql-org/go-mysql/pull/683) ([moredure](https://github.com/moredure))
+* Stop using channel for memory pooling. [#682](https://github.com/go-mysql-org/go-mysql/pull/682) ([moredure](https://github.com/moredure))
+* Fix TIME value near zero will become zero. [#678](https://github.com/go-mysql-org/go-mysql/pull/678) ([lance6716](https://github.com/lance6716))
+* Fixed reading connection attributes on server side. [#676](https://github.com/go-mysql-org/go-mysql/pull/676) ([skoef](https://github.com/skoef))
+* Support client connection attributes on the client side. [#675](https://github.com/go-mysql-org/go-mysql/pull/675) ([skoef](https://github.com/skoef))
+* Support client connection attributes on the server side. [#672](https://github.com/go-mysql-org/go-mysql/pull/672) ([skoef](https://github.com/skoef))
+* Added constants for COM_SET_OPTION. [#670](https://github.com/go-mysql-org/go-mysql/pull/670) ([skoef](https://github.com/skoef))
+* Implemented ExecuteMultiple. [#668](https://github.com/go-mysql-org/go-mysql/pull/668) ([skoef](https://github.com/skoef))
+* Add MysqlGTIDSet.Add() and Minus() methods. [#667](https://github.com/go-mysql-org/go-mysql/pull/667) ([ostinru](https://github.com/ostinru))
+* Enabled ineffassign in golangci. [#666](https://github.com/go-mysql-org/go-mysql/pull/666) ([skoef](https://github.com/skoef))
+* Enabled gosimple in golangci. [#665](https://github.com/go-mysql-org/go-mysql/pull/665) ([skoef](https://github.com/skoef))
+* Add MysqlGTIDSet.Add() and Minus() methods. [#662](https://github.com/go-mysql-org/go-mysql/pull/662) ([ostinru](https://github.com/ostinru))
+* Store json as string in rows events. [#658](https://github.com/go-mysql-org/go-mysql/pull/658) ([D3Hunter](https://github.com/D3Hunter))
+* Set slave_uuid and replica_uuid. [#656](https://github.com/go-mysql-org/go-mysql/pull/656) ([lance6716](https://github.com/lance6716))
+* Fix malformed packet and a panic. [#655](https://github.com/go-mysql-org/go-mysql/pull/655) ([lance6716](https://github.com/lance6716))
+* Fix fetchIndexesViaSqlDB/NewTableFromSqlDB for MySQL 8.0. [#527](https://github.com/go-mysql-org/go-mysql/pull/527) ([RobinGeuze](https://github.com/RobinGeuze))
+
 ### Tag v1.4.0 (2021-12-15)
 * Fix that forget to readOK after writing. [#652](https://github.com/go-mysql-org/go-mysql/pull/652) ([lance6716](https://github.com/lance6716))
 * Packet: memory optimization for reading network packet. [#651](https://github.com/go-mysql-org/go-mysql/pull/651) ([zr-hebo](https://github.com/zr-hebo))

README.md

@@ -9,10 +9,10 @@ A pure go library to handle MySQL network protocol and replication.
 ## How to migrate to this repo
 To change the used package in your repo it's enough to add this `replace` directive to your `go.mod`:
 ```
-replace github.com/siddontang/go-mysql => github.com/go-mysql-org/go-mysql v1.4.0
+replace github.com/siddontang/go-mysql => github.com/go-mysql-org/go-mysql v1.5.0
 ```
 
-v1.4.0 - is the last tag in repo, feel free to choose what you want.
+v1.5.0 - is the last tag in repo, feel free to choose what you want.
 
 ## Changelog
 This repo uses [Changelog](CHANGELOG.md).

canal/sync.go

@@ -8,9 +8,9 @@ import (
 "github.com/go-mysql-org/go-mysql/mysql"
 "github.com/go-mysql-org/go-mysql/replication"
 "github.com/go-mysql-org/go-mysql/schema"
+"github.com/google/uuid"
 "github.com/pingcap/errors"
 "github.com/pingcap/parser/ast"
-uuid "github.com/satori/go.uuid"
 "github.com/siddontang/go-log/log"
 )
 

client/auth.go

@@ -139,6 +139,20 @@ func (c *Conn) genAuthResponse(authData []byte) ([]byte, bool, error) {
 }
 }
 
+// generate connection attributes data
+func (c *Conn) genAttributes() []byte {
+if len(c.attributes) == 0 {
+return nil
+}
+
+attrData := make([]byte, 0)
+for k, v := range c.attributes {
+attrData = append(attrData, PutLengthEncodedString([]byte(k))...)
+attrData = append(attrData, PutLengthEncodedString([]byte(v))...)
+}
+return append(PutLengthEncodedInt(uint64(len(attrData))), attrData...)
+}
+
 // See: http://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::HandshakeResponse
 func (c *Conn) writeAuthHandshake() error {
 if !authPluginAllowed(c.authPluginName) {
@@ -195,6 +209,12 @@ func (c *Conn) writeAuthHandshake() error {
 capability |= CLIENT_CONNECT_WITH_DB
 length += len(c.db) + 1
 }
+// connection attributes
+attrData := c.genAttributes()
+if len(attrData) > 0 {
+capability |= CLIENT_CONNECT_ATTRS
+length += len(attrData)
+}
 
 data := make([]byte, length+4)
 
@@ -264,6 +284,12 @@ func (c *Conn) writeAuthHandshake() error {
 // Assume native client during response
 pos += copy(data[pos:], c.authPluginName)
 data[pos] = 0x00
+pos++
+
+// connection attributes
+if len(attrData) > 0 {
+copy(data[pos:], attrData)
+}
 
 return c.WritePacket(data)
 }

client/auth_test.go

@@ -0,0 +1,42 @@
+package client
+
+import (
+"bytes"
+"testing"
+
+"github.com/go-mysql-org/go-mysql/mysql"
+)
+
+func TestConnGenAttributes(t *testing.T) {
+c := &Conn{
+// example data from
+// https://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::HandshakeResponse41
+attributes: map[string]string{
+"_os": "debian6.0",
+"_client_name": "libmysql",
+"_pid": "22344",
+"_client_version": "5.6.6-m9",
+"_platform": "x86_64",
+"foo": "bar",
+},
+}
+
+data := c.genAttributes()
+
+// the order of the attributes map cannot be guaranteed so to test the content
+// of the attribute data we need to check its partial contents
+
+if len(data) != 98 {
+t.Fatalf("unexpected data length, got %d", len(data))
+}
+if data[0] != 0x61 {
+t.Fatalf("unexpected length-encoded int, got %#x", data[0])
+}
+
+for k, v := range c.attributes {
+fixt := append(mysql.PutLengthEncodedString([]byte(k)), mysql.PutLengthEncodedString([]byte(v))...)
+if !bytes.Contains(data, fixt) {
+t.Fatalf("%s attribute not found", k)
+}
+}
+}

client/conn.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+"bytes"
 "context"
 "crypto/tls"
 "fmt"
@@ -28,6 +29,8 @@ type Conn struct {
 // client-set capabilities only
 ccaps uint32
 
+attributes map[string]string
+
 status uint16
 
 charset string
@@ -221,30 +224,30 @@ func (c *Conn) ExecuteMultiple(query string, perResultCallback ExecPerResultCall
 return nil, errors.Trace(err)
 }
 
-var buf []byte
 var err error
 var result *Result
-defer utils.ByteSlicePut(buf)
+
+bs := utils.ByteSliceGet(16)
+defer utils.ByteSlicePut(bs)
 
 for {
-buf, err = c.ReadPacketReuseMem(utils.ByteSliceGet(16)[:0])
+bs.B, err = c.ReadPacketReuseMem(bs.B[:0])
 if err != nil {
 return nil, errors.Trace(err)
 }
 
-switch buf[0] {
+switch bs.B[0] {
 case OK_HEADER:
-result, err = c.handleOKPacket(buf)
+result, err = c.handleOKPacket(bs.B)
 case ERR_HEADER:
-err = c.handleErrorPacket(append([]byte{}, buf...))
+err = c.handleErrorPacket(bytes.Repeat(bs.B, 1))
 result = nil
 case LocalInFile_HEADER:
 err = ErrMalformPacket
 result = nil
 default:
-result, err = c.readResultset(buf, false)
+result, err = c.readResultset(bs.B, false)
 }
-
 // call user-defined callback
 perResultCallback(result, err)
 
@@ -302,6 +305,10 @@ func (c *Conn) Rollback() error {
 return errors.Trace(err)
 }
 
+func (c *Conn) SetAttributes(attributes map[string]string) {
+c.attributes = attributes
+}
+
 func (c *Conn) SetCharset(charset string) error {
 if c.charset == charset {
 return nil

client/req.go

@@ -21,11 +21,11 @@ func (c *Conn) writeCommandBuf(command byte, arg []byte) error {
 
 length := len(arg) + 1
 data := utils.ByteSliceGet(length + 4)
-data[4] = command
+data.B[4] = command
 
-copy(data[5:], arg)
+copy(data.B[5:], arg)
 
-err := c.WritePacket(data)
+err := c.WritePacket(data.B)
 
 utils.ByteSlicePut(data)
 

client/resp.go

@@ -216,38 +216,42 @@ func (c *Conn) readOK() (*Result, error) {
 }
 
 func (c *Conn) readResult(binary bool) (*Result, error) {
-firstPkgBuf, err := c.ReadPacketReuseMem(utils.ByteSliceGet(16)[:0])
-defer utils.ByteSlicePut(firstPkgBuf)
-
+bs := utils.ByteSliceGet(16)
+defer utils.ByteSlicePut(bs)
+var err error
+bs.B, err = c.ReadPacketReuseMem(bs.B[:0])
 if err != nil {
 return nil, errors.Trace(err)
 }
 
-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 {
+switch bs.B[0] {
+case OK_HEADER:
+return c.handleOKPacket(bs.B)
+case ERR_HEADER:
+return nil, c.handleErrorPacket(bytes.Repeat(bs.B, 1))
+case LocalInFile_HEADER:
 return nil, ErrMalformPacket
+default:
+return c.readResultset(bs.B, binary)
 }
-
-return c.readResultset(firstPkgBuf, binary)
 }
 
 func (c *Conn) readResultStreaming(binary bool, result *Result, perRowCb SelectPerRowCallback, perResCb SelectPerResultCallback) error {
-firstPkgBuf, err := c.ReadPacketReuseMem(utils.ByteSliceGet(16)[:0])
-defer utils.ByteSlicePut(firstPkgBuf)
-
+bs := utils.ByteSliceGet(16)
+defer utils.ByteSlicePut(bs)
+var err error
+bs.B, err = c.ReadPacketReuseMem(bs.B[:0])
 if err != nil {
 return errors.Trace(err)
 }
 
-if firstPkgBuf[0] == OK_HEADER {
+switch bs.B[0] {
+case OK_HEADER:
 // https://dev.mysql.com/doc/internals/en/com-query-response.html
 // 14.6.4.1 COM_QUERY Response
 // If the number of columns in the resultset is 0, this is a OK_Packet.
 
-okResult, err := c.handleOKPacket(firstPkgBuf)
+okResult, err := c.handleOKPacket(bs.B)
 if err != nil {
 return errors.Trace(err)
 }
@@ -262,13 +266,13 @@ func (c *Conn) readResultStreaming(binary bool, result *Result, perRowCb SelectP
 result.Reset(0)
 }
 return nil
-} else if firstPkgBuf[0] == ERR_HEADER {
-return c.handleErrorPacket(append([]byte{}, firstPkgBuf...))
-} else if firstPkgBuf[0] == LocalInFile_HEADER {
+case ERR_HEADER:
+return c.handleErrorPacket(bytes.Repeat(bs.B, 1))
+case LocalInFile_HEADER:
 return ErrMalformPacket
+default:
+return c.readResultsetStreaming(bs.B, binary, result, perRowCb, perResCb)
 }
-
-return c.readResultsetStreaming(firstPkgBuf, binary, result, perRowCb, perResCb)
 }
 
 func (c *Conn) readResultset(data []byte, binary bool) (*Result, error) {

client/tls.go

@@ -13,16 +13,28 @@ func NewClientTLSConfig(caPem, certPem, keyPem []byte, insecureSkipVerify bool,
 panic("failed to add ca PEM")
 }
 
-cert, err := tls.X509KeyPair(certPem, keyPem)
-if err != nil {
-panic(err)
-}
+var config *tls.Config
 
-config := &tls.Config{
-Certificates: []tls.Certificate{cert},
-RootCAs: pool,
-InsecureSkipVerify: insecureSkipVerify,
-ServerName: serverName,
+// Allow cert and key to be optional
+// Send through `make([]byte, 0)` for "nil"
+if string(certPem) != "" && string(keyPem) != "" {
+cert, err := tls.X509KeyPair(certPem, keyPem)
+if err != nil {
+panic(err)
+}
+config = &tls.Config{
+RootCAs: pool,
+Certificates: []tls.Certificate{cert},
+InsecureSkipVerify: insecureSkipVerify,
+ServerName: serverName,
+}
+} else {
+config = &tls.Config{
+RootCAs: pool,
+InsecureSkipVerify: insecureSkipVerify,
+ServerName: serverName,
+}
 }
+
 return config
 }