xg_driver/xugu/xugusql_sock_send.go

package xugu

import "encoding/binary"

var BUFF_SIZE = 8 * 1024

// EncryptBuff 假设的加密函数
func EncryptBuff(pconn *xugusqlConn, data []byte) {
	// 加密逻辑，根据实际情况实现
}

func sockSendCommand0(conn *xugusqlConn, sql string) (int, error) {
	cmdLen := len(sql)
	//发送问号字符 ?
	ret, err := RhSend(conn, []byte("?"), cmdLen)
	if err != nil {
		return 0, err
	}
	// Comand_Len
	lengthInt32 := uint32(conn.sendBuff.Len())
	ret, err = rhSendInt32(conn, lengthInt32)
	if err != nil {
		return 0, err
	}
	//  Comand_str
	ret, err = RhSend(conn, []byte(sql), cmdLen+1)
	if err != nil {
		return 0, err
	}

	// Param_num
	ret, err = rhSendInt32(conn, 0)
	if err != nil {
		return 0, err
	}
	return ret, nil
}

// RhSend Go 版本的 rh_send 函数
// TODO :加密连接还没有实现
func RhSend(pconn *xugusqlConn, buff []byte, length int) (int, error) {
	pconn.mu.Lock()
	defer pconn.mu.Unlock()
	//没有内容则不发送
	if length == 0 {
		return 0, nil
	}
	//连接里是否有加密
	if pconn.turingDataR != nil && pconn.useSSL {
		if length > 2*1024 {

			EncryptBuff(pconn, buff)
			if _, err := pconn.conn.Write(pconn.sendBuff.Bytes()); err != nil {
				pconn.sendBuff.Reset()
				return 0, err
			}
			pconn.sendBuff.Reset()
			_, err := pconn.conn.Write(buff)
			return len(buff), err
		} else if pconn.sendBuff.Len()+length >= BUFF_SIZE {
			if _, err := pconn.conn.Write(pconn.sendBuff.Bytes()); err != nil {
				pconn.sendBuff.Reset()
				return 0, err
			}
			pconn.sendBuff.Reset()
			_, err := pconn.sendBuff.Write(buff)
			if err != nil {
				return 0, err
			}
			EncryptBuff(pconn, pconn.sendBuff.Bytes())
			return len(buff), nil
		} else {
			pconn.sendBuff.Write(buff)
			EncryptBuff(pconn, pconn.sendBuff.Bytes()[pconn.sendBuff.Len()-length:])
			return length, nil
		}
	} else {
		/*如果数据长度大于 2048，先发送缓冲区中的数据，然后直接发送新的大数据块。*/
		if length > 2048 {
			if _, err := pconn.conn.Write(pconn.sendBuff.Bytes()); err != nil {
				//将缓冲区重置为空
				pconn.sendBuff.Reset()
				return 0, err
			}
			//将缓冲区重置为空
			pconn.sendBuff.Reset()
			//发送新的大数据块
			_, err := pconn.conn.Write(buff)
			return length, err

		} else if pconn.sendBuff.Len()+length >= BUFF_SIZE {
			//缓冲区空间不足：
			/*- 如果当前缓冲区中的数据加上新数据的长度超过 `8 * 1024` 字节，
			则先发送缓冲区中的数据，然后重置缓冲区，将新数据拷贝到缓冲区中。*/
			if _, err := pconn.conn.Write(pconn.sendBuff.Bytes()); err != nil {
				pconn.sendBuff.Reset()
				return 0, err
			}
			pconn.sendBuff.Reset()
			pconn.sendBuff.Write(buff)
			return length, nil
		} else {
			//TODO: 这里c代码里是添加到缓存buff里面 ，不直接发送
			pconn.sendBuff.Write(buff)
			return length, nil
		}
	}
}

func rhSendInt32(pconn *xugusqlConn, i uint32) (int, error) {

	// 创建一个 4 字节的缓冲区，用于存储网络字节序的 int32 值
	var networkBytes [4]byte
	// 将 int32 值转换为网络字节序（大端字节序）并写入缓冲区
	binary.BigEndian.PutUint32(networkBytes[:], i)

	return RhSend(pconn, networkBytes[:], 4)
}

func RhFlush(pconn *xugusqlConn) (int, error) {
	if pconn.sendBuff.Len() > 0 {
		if _, err := pconn.conn.Write(pconn.sendBuff.Bytes()); err != nil {
			pconn.sendBuff.Reset()
			return 0, err
		}
		pconn.sendBuff.Reset()
	}
	return 0, nil
}