func (dc *DirectConnection) readResultRows(result *mysql.Result, isBinary bool) (err error) {
	var data []byte
	for {
		data, err = dc.readPacket()
		if err != nil {
			return
		}
		if dc.isEOFPacket(data) {
			if dc.capability&mysql.ClientProtocol41 > 0 {
				result.Status = binary.LittleEndian.Uint16(data[3:])
				dc.status = result.Status
			}
			break
		}
		result.RowDatas = append(result.RowDatas, data)
	}
	result.Values = make([][]interface{}, len(result.RowDatas))
	for i := range result.Values {
		result.Values[i], err = result.RowDatas[i].Parse(result.Fields, isBinary)
		if err != nil {
			return err
		}
	}
	return nil
}
readResultRows read result rows
package backend
import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"net"
	"strings"
	"github.com/nooncall/shazam/log"
	"github.com/nooncall/shazam/mysql"
	"github.com/nooncall/shazam/util/sync2"
)
type DirectConnection struct {
	conn *mysql.Conn
	addr     string
	user     string
	password string
	db       string
	capability uint32
	sessionVariables *mysql.SessionVariables
	status uint16
	collation mysql.CollationID
	charset   string
	salt      []byte
	defaultCollation mysql.CollationID
	defaultCharset   string
	pkgErr error
	closed sync2.AtomicBool
}
func NewDirectConnection(addr string, user string, password string, db string, charset string, collationID mysql.CollationID) (*DirectConnection, error) {
	dc := &DirectConnection{
		addr:             addr,
		user:             user,
		password:         password,
		db:               db,
		charset:          charset,
		collation:        collationID,
		defaultCharset:   charset,
		defaultCollation: collationID,
		closed:           sync2.NewAtomicBool(false),
		sessionVariables: mysql.NewSessionVariables(),
	}
	err := dc.connect()
	return dc, err
}
func (dc *DirectConnection) connect() error {
	if dc.conn != nil {
		dc.conn.Close()
	}
	typ := "tcp"
	if strings.Contains(dc.addr, "/") {
		typ = "unix"
	}
	netConn, err := net.Dial(typ, dc.addr)
	if err != nil {
		return err
	}
	tcpConn := netConn.(*net.TCPConn)
	tcpConn.SetNoDelay(true)
	tcpConn.SetKeepAlive(true)
	dc.conn = mysql.NewConn(tcpConn)
	if err := dc.readInitialHandshake(); err != nil {
		dc.conn.Close()
		return err
	}
	if err := dc.writeHandshakeResponse41(); err != nil {
		dc.conn.Close()
		return err
	}
	response, err := dc.readPacket()
	if err != nil {
		dc.conn.Close()
		return err
	}
	switch response[0] {
	case mysql.OKHeader:
	default:
		return errors.New("dc connection handshake failed with mysql")
	}
	if !dc.IsAutoCommit() {
		if _, err := dc.exec("set autocommit = 1"); err != nil {
			dc.conn.Close()
			return err
		}
	}
	return nil
}
func (dc *DirectConnection) Close() {
	if dc.conn != nil {
		dc.conn.Close()
	}
	dc.conn = nil
	dc.salt = nil
	dc.pkgErr = nil
	dc.closed.Set(true)
	return
}
func (dc *DirectConnection) IsClosed() bool {
	return dc.closed.Get()
}
func (dc *DirectConnection) readPacket() ([]byte, error) {
	data, err := dc.conn.ReadPacket()
	dc.pkgErr = err
	return data, err
}
func (dc *DirectConnection) writePacket(data []byte) error {
	err := dc.conn.WritePacket(data)
	if err != nil && strings.Contains(err.Error(), "broken pipe") {
		for i := 0; i < 3; i++ {
			dc.Close()
			e := dc.connect()
			if e == nil {  
				break
			}
		}
	}
	return err
}
func (dc *DirectConnection) writeEphemeralPacket() error {
	err := dc.conn.WriteEphemeralPacket()
	if err != nil && strings.Contains(err.Error(), "broken pipe") {
		for i := 0; i < 3; i++ {
			dc.Close()
			e := dc.connect()
			if e == nil {  
				break
			}
		}
	}
	return err
}
func (dc *DirectConnection) readInitialHandshake() error {
	data, err := dc.readPacket()
	if err != nil {
		return err
	}
	if data[0] == mysql.ErrHeader {
		return errors.New("read initial handshake error")
	}
	if data[0] < mysql.MinProtocolVersion {
		return fmt.Errorf("invalid protocol version %d, must >= 10", data[0])
	}
	pos := 1 + bytes.IndexByte(data[1:], 0x00) + 1 + 4
	dc.salt = append(dc.salt, data[pos:pos+8]...)
	pos += 8 + 1
	dc.capability = uint32(binary.LittleEndian.Uint16(data[pos : pos+2]))
	pos += 2
	if len(data) > pos {
		pos++
		dc.status = binary.LittleEndian.Uint16(data[pos : pos+2])
		pos += 2
		dc.capability = uint32(binary.LittleEndian.Uint16(data[pos:pos+2]))<<16 | dc.capability
		pos += 2
		pos += 10 + 1
		dc.salt = append(dc.salt, data[pos:pos+12]...)
	}
	return nil
}
func (dc *DirectConnection) writeHandshakeResponse41() error {
	capability := mysql.ClientProtocol41 | mysql.ClientSecureConnection |
		mysql.ClientLongPassword | mysql.ClientTransactions | mysql.ClientLongFlag
	capability &= dc.capability
	auth := mysql.CalcPassword(dc.salt, []byte(dc.password))
	length := 4 +  
		4 +  
		1 +  
		23 +  
		mysql.LenNullString(dc.user) +  
		1 +
		len(auth)
	if len(dc.db) > 0 {
		capability |= mysql.ClientConnectWithDB
		length += mysql.LenNullString(dc.db)
	}
	dc.capability = capability
	data := make([]byte, length, length)
	pos := 0
	pos = mysql.WriteUint32(data, pos, capability)
	pos = mysql.WriteZeroes(data, pos, 4)
	pos = mysql.WriteByte(data, pos, byte(dc.collation))
	pos = mysql.WriteZeroes(data, pos, 23)
	pos = mysql.WriteNullString(data, pos, dc.user)
	data[pos] = byte(len(auth))
	pos++
	pos += copy(data[pos:], auth)
	if len(dc.db) > 0 {
		pos = mysql.WriteNullString(data, pos, dc.db)
	}
	if err := dc.writePacket(data); err != nil {
		return err
	}
	return nil
}
func (dc *DirectConnection) writeComInitDB(db string) error {
	dc.conn.SetSequence(0)
	data := make([]byte, len(db)+1, len(db)+1)
	data[0] = mysql.ComInitDB
	copy(data[1:], db)
	if err := dc.writePacket(data); err != nil {
		return err
	}
	return nil
}
func (dc *DirectConnection) writeComQuery(sql string) error {
	dc.conn.SetSequence(0)
	data := dc.conn.StartEphemeralPacket(len(sql) + 1)
	data[0] = mysql.ComQuery
	copy(data[1:], sql)
	if err := dc.writeEphemeralPacket(); err != nil {
		return err
	}
	return nil
}
func (dc *DirectConnection) writeComFieldList(table string, wildcard string) error {
	dc.conn.SetSequence(0)
	length := 1 +
		mysql.LenNullString(table) +
		mysql.LenNullString(wildcard)
	data := make([]byte, length, length)
	pos := 0
	pos = mysql.WriteByte(data, 0, mysql.ComFieldList)
	pos = mysql.WriteNullString(data, pos, table)
	pos = mysql.WriteNullString(data, pos, wildcard)
	if err := dc.writePacket(data); err != nil {
		return err
	}
	return nil
}
func (dc *DirectConnection) Ping() error {
	dc.conn.SetSequence(0)
	if err := dc.writePacket([]byte{mysql.ComPing}); err != nil {
		return err
	}
	data, err := dc.readPacket()
	if err != nil {
		return err
	}
	switch data[0] {
	case mysql.OKHeader:
		return nil
	case mysql.ErrHeader:
		return errors.New("dc connection ping failed")
	}
	return fmt.Errorf("unexpected packet type: %d", data[0])
}
func (dc *DirectConnection) UseDB(dbName string) error {
	dc.conn.SetSequence(0)
	if dc.db == dbName || len(dbName) == 0 {
		return nil
	}
	if err := dc.writeComInitDB(dbName); err != nil {
		return err
	}
	if r, err := dc.readPacket(); err != nil {
		return err
	} else if !mysql.IsOKPacket(r) {
		return errors.New("dc connection use db failed")
	}
	dc.db = dbName
	return nil
}
func (dc *DirectConnection) GetDB() string {
	return dc.db
}
func (dc *DirectConnection) GetAddr() string {
	return dc.addr
}
func (dc *DirectConnection) Execute(sql string) (*mysql.Result, error) {
	return dc.exec(sql)
}
func (dc *DirectConnection) Begin() error {
	_, err := dc.exec("begin")
	return err
}
func (dc *DirectConnection) Commit() error {
	_, err := dc.exec("commit")
	return err
}
func (dc *DirectConnection) Rollback() error {
	_, err := dc.exec("rollback")
	return err
}
func (dc *DirectConnection) SetAutoCommit(v uint8) error {
	if v == 0 {
		if _, err := dc.exec("set autocommit = 0"); err != nil {
			dc.conn.Close()
			return err
		}
	} else {
		if _, err := dc.exec("set autocommit = 1"); err != nil {
			dc.conn.Close()
			return err
		}
	}
	return nil
}
func (dc *DirectConnection) SetCharset(charset string, collation mysql.CollationID) (   bool, error) {
	charset = strings.Trim(charset, "\"'`")
	if collation == 0 {
		collation = mysql.CollationNames[mysql.Charsets[charset]]
	}
	if dc.charset == charset && dc.collation == collation {
		return false, nil
	}
	_, ok := mysql.CharsetIds[charset]
	if !ok {
		return false, fmt.Errorf("invalid charset %s", charset)
	}
	_, ok = mysql.Collations[collation]
	if !ok {
		return false, fmt.Errorf("invalid collation %d", collation)
	}
	dc.collation = collation
	dc.charset = charset
	return true, nil
}
func (dc *DirectConnection) ResetConnection() error {
	if dc.IsInTransaction() {
		log.Debug("get transaction connection from pool, addr: %s, user: %s, db: %s, status: %d", dc.addr, dc.user, dc.db, dc.status)
		if err := dc.Rollback(); err != nil {
			log.Warn("rollback in reset connection error, addr: %s, user: %s, db: %s, status: %d, err: %v", dc.addr, dc.user, dc.db, dc.status, err)
			return err
		}
	}
	if !dc.IsAutoCommit() {
		log.Debug("get autocommit = 0 connection from pool, addr: %s, user: %s, db: %s, status: %d", dc.addr, dc.user, dc.db, dc.status)
		if err := dc.SetAutoCommit(1); err != nil {
			log.Warn("set autocommit = 1 in reset connection error, addr: %s, user: %s, db: %s, status: %d, err: %v", dc.addr, dc.user, dc.db, dc.status, err)
			return err
		}
	}
	return nil
}
func (dc *DirectConnection) SetSessionVariables(frontend *mysql.SessionVariables) (bool, error) {
	return dc.sessionVariables.SetEqualsWith(frontend)
}
func (dc *DirectConnection) WriteSetStatement() error {
	var setVariableSQL bytes.Buffer
	collation, ok := mysql.Collations[dc.collation]
	if !ok {
		return fmt.Errorf("invalid collationId: %v", dc.collation)
	}
	appendSetCharset(&setVariableSQL, dc.charset, collation)
	for _, v := range dc.sessionVariables.GetAll() {
		appendSetVariable(&setVariableSQL, v.Name(), v.Get())
	}
	for _, v := range dc.sessionVariables.GetUnusedAndClear() {
		appendSetVariableToDefault(&setVariableSQL, v.Name())
	}
	setSQL := setVariableSQL.String()
	if setSQL == "" {
		return nil
	}
	if _, err := dc.exec(setSQL); err != nil {
		return err
	}
	return nil
}
func (dc *DirectConnection) FieldList(table string, wildcard string) ([]*mysql.Field, error) {
	if err := dc.writeComFieldList(table, wildcard); err != nil {
		return nil, err
	}
	data, err := dc.readPacket()
	if err != nil {
		return nil, err
	}
	fs := make([]*mysql.Field, 0, 4)
	var f *mysql.Field
	if data[0] == mysql.ErrHeader {
		return nil, dc.handleErrorPacket(data)
	}
	for {
		if data, err = dc.readPacket(); err != nil {
			return nil, err
		}
		if dc.isEOFPacket(data) {
			return fs, nil
		}
		if f, err = mysql.FieldData(data).Parse(); err != nil {
			return nil, err
		}
		fs = append(fs, f)
	}
}
func (dc *DirectConnection) exec(query string) (*mysql.Result, error) {
	if err := dc.writeComQuery(query); err != nil {
		return nil, err
	}
	return dc.readResult(false)
}
func (dc *DirectConnection) readResultset(data []byte, binary bool) (*mysql.Result, error) {
	result := &mysql.Result{
		Status:       0,
		InsertID:     0,
		AffectedRows: 0,
		Resultset: &mysql.Resultset{},
	}
	pos := 0
	count, pos, _, _ := mysql.ReadLenEncInt(data, pos)
	if pos-len(data) != 0 {
		return nil, mysql.ErrMalformPacket
	}
	result.Fields = make([]*mysql.Field, count)
	result.FieldNames = make(map[string]int, count)
	if err := dc.readResultColumns(result); err != nil {
		return nil, err
	}
	if err := dc.readResultRows(result, binary); err != nil {
		return nil, err
	}
	return result, nil
}
func (dc *DirectConnection) readResultColumns(result *mysql.Result) (err error) {
	var i = 0
	var data []byte
	for {
		data, err = dc.readPacket()
		if err != nil {
			return
		}
		if dc.isEOFPacket(data) {
			if dc.capability&mysql.ClientProtocol41 > 0 {
				result.Status = binary.LittleEndian.Uint16(data[3:])
				dc.status = result.Status
			}
			if i != len(result.Fields) {
				err = mysql.ErrMalformPacket
			}
			return
		}
		result.Fields[i], err = mysql.FieldData(data).Parse()
		if err != nil {
			return
		}
		result.FieldNames[string(result.Fields[i].Name)] = i
		i++
	}
}