make it work

Author: serprex

client/auth.go

@@ -92,7 +92,7 @@ func (c *Conn) readInitialHandshake() error {
 		pos += 2
 
 		// The upper 2 bytes of the Capabilities Flags
-		c.capability = uint32(binary.LittleEndian.Uint16(data[pos:pos+2]))<<16 | c.capability
+		c.capability |= uint32(binary.LittleEndian.Uint16(data[pos:pos+2])) << 16
 		pos += 2
 
 		// length of the combined auth_plugin_data (scramble), if auth_plugin_data_len is > 0
@@ -209,10 +209,8 @@ func (c *Conn) writeAuthHandshake() error {
 
 	// Set default client capabilities that reflect the abilities of this library
 	capability := mysql.CLIENT_PROTOCOL_41 | mysql.CLIENT_SECURE_CONNECTION |
-		mysql.CLIENT_LONG_PASSWORD | mysql.CLIENT_TRANSACTIONS | mysql.CLIENT_PLUGIN_AUTH
-	// Adjust client capability flags based on server support
-	capability |= c.capability & mysql.CLIENT_LONG_FLAG
-	capability |= c.capability & mysql.CLIENT_QUERY_ATTRIBUTES
+		mysql.CLIENT_LONG_PASSWORD | mysql.CLIENT_TRANSACTIONS | mysql.CLIENT_PLUGIN_AUTH |
+		mysql.CLIENT_LONG_FLAG | mysql.CLIENT_QUERY_ATTRIBUTES
 	// Adjust client capability flags on specific client requests
 	// Only flags that would make any sense setting and aren't handled elsewhere
 	// in the library are supported here
@@ -275,6 +273,7 @@ func (c *Conn) writeAuthHandshake() error {
 	data := make([]byte, length+4)
 
 	// capability [32 bit]
+	c.capability &= capability
 	data[4] = byte(capability)
 	data[5] = byte(capability >> 8)
 	data[6] = byte(capability >> 16)

client/conn.go

@@ -252,7 +252,7 @@ func (c *Conn) UnsetCapability(cap uint32) {
 
 // HasCapability returns true if the connection has the specific capability
 func (c *Conn) HasCapability(cap uint32) bool {
-	return c.ccaps&cap > 0
+	return c.ccaps&cap != 0
 }
 
 // UseSSL: use default SSL

client/resp.go

@@ -14,22 +14,6 @@ import (
 	"github.com/go-mysql-org/go-mysql/utils"
 )
 
-func (c *Conn) readUntilEOF() (err error) {
-	var data []byte
-
-	for {
-		data, err = c.ReadPacket()
-		if err != nil {
-			return err
-		}
-
-		// EOF Packet
-		if c.isEOFPacket(data) {
-			return err
-		}
-	}
-}
-
 func (c *Conn) isEOFPacket(data []byte) bool {
 	return data[0] == mysql.EOF_HEADER && len(data) <= 5
 }
@@ -357,7 +341,7 @@ func (c *Conn) readResultColumns(result *mysql.Result) (err error) {
 		result.FieldNames[utils.ByteSliceToString(result.Fields[i].Name)] = i
 	}
 
-	if !c.HasCapability(mysql.CLIENT_DEPRECATE_EOF) {
+	if c.capability&mysql.CLIENT_DEPRECATE_EOF == 0 {
 		// EOF Packet
 		rawPkgLen := len(result.RawPkg)
 		result.RawPkg, err = c.ReadPacketReuseMem(result.RawPkg)
@@ -394,7 +378,7 @@ func (c *Conn) readResultRows(result *mysql.Result, isBinary bool) (err error) {
 		data = result.RawPkg[rawPkgLen:]
 
 		if data[0] == mysql.EOF_HEADER && len(data) <= 0xffffff {
-			if c.HasCapability(mysql.CLIENT_DEPRECATE_EOF) {
+			if c.capability&mysql.CLIENT_DEPRECATE_EOF != 0 {
 				// Treat like OK
 				affectedRows, _, n := mysql.LengthEncodedInt(data[1:])
 				insertId, _, m := mysql.LengthEncodedInt(data[1+n:])
@@ -446,9 +430,16 @@ func (c *Conn) readResultRowsStreaming(result *mysql.Result, isBinary bool, perR
 			return err
 		}
 
-		// EOF Packet
-		if c.isEOFPacket(data) {
-			if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
+		if data[0] == mysql.EOF_HEADER && len(data) <= 0xffffff {
+			if c.capability&mysql.CLIENT_DEPRECATE_EOF != 0 {
+				// Treat like OK
+				affectedRows, _, n := mysql.LengthEncodedInt(data[1:])
+				insertId, _, m := mysql.LengthEncodedInt(data[1+n:])
+				result.Status = binary.LittleEndian.Uint16(data[1+n+m:])
+				result.AffectedRows = affectedRows
+				result.InsertId = insertId
+				c.status = result.Status
+			} else if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
 				result.Warnings = binary.LittleEndian.Uint16(data[1:])
 				// todo add strict_mode, warning will be treat as error
 				result.Status = binary.LittleEndian.Uint16(data[3:])

client/stmt.go

@@ -275,14 +275,33 @@ func (c *Conn) Prepare(query string) (*Stmt, error) {
 	}
 
 	if s.params > 0 {
-		if err := s.conn.readUntilEOF(); err != nil {
-			return nil, errors.Trace(err)
+		for range s.params {
+			if _, err := s.conn.ReadPacket(); err != nil {
+				return nil, errors.Trace(err)
+			}
+		}
+		if s.conn.capability&mysql.CLIENT_DEPRECATE_EOF == 0 {
+			if packet, err := s.conn.ReadPacket(); err != nil {
+				return nil, errors.Trace(err)
+			} else if c.isEOFPacket(packet) {
+				return nil, mysql.ErrMalformPacket
+			}
 		}
 	}
 
 	if s.columns > 0 {
-		if err := s.conn.readUntilEOF(); err != nil {
-			return nil, errors.Trace(err)
+		// TODO process when CLIENT_CACHE_METADATA enabled
+		for range s.columns {
+			if _, err := s.conn.ReadPacket(); err != nil {
+				return nil, errors.Trace(err)
+			}
+		}
+		if s.conn.capability&mysql.CLIENT_DEPRECATE_EOF == 0 {
+			if packet, err := s.conn.ReadPacket(); err != nil {
+				return nil, errors.Trace(err)
+			} else if c.isEOFPacket(packet) {
+				return nil, mysql.ErrMalformPacket
+			}
 		}
 	}
 

replication/row_event.go

@@ -1082,7 +1082,6 @@ func (e *RowsEvent) DecodeData(pos int, data []byte) (err2 error) {
 	if e.compressed {
 		data, err2 = mysql.DecompressMariadbData(data[pos:])
 		if err2 != nil {
-			//nolint:nakedret
 			return err2
 		}
 		pos = 0