xugu_driver_go/xugu/xugu_sock_send.go

package xugu

import (
	"encoding/binary"
	"fmt"
)

func SockSendCommandN(pConn *xuguConn, sql string, num int) (int, error) {
	ret := 0
	var err error
	cmd_len := len(sql)
	ret, err = RhSend(pConn, []byte("?"), 1)
	if err != nil {
		return 0, err
	}
	ret, err = rhSendInt32(pConn, uint32(cmd_len))
	if err != nil {
		return 0, err
	}
	ret, err = RhSend(pConn, []byte(sql), cmd_len+1)
	if err != nil {
		return 0, err
	}
	if num != 0 {
		ret, err = rhSendInt32(pConn, uint32(num)) //param num
		if err != nil {
			return 0, err
		}
	} else {
		ret, err = rhSendInt32(pConn, uint32(0))
		if err != nil {
			return 0, err
		}
	}

	return ret, nil
}

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

	data := make([]byte, 1)
	ret, err = RhSend(pConn, data, 1)
	if err != nil {
		return 0, err
	}

	// Param_num
	ret, err = rhSendInt32(pConn, 0)
	if err != nil {
		return 0, err
	}
	fmt.Println("SockSendCommand0 msg", pConn.sendBuff.String())

	RhFlush(pConn)

	return ret, nil
}

// RhSend Go 版本的 rh_send 函数
// TODO :加密连接还没有实现
func RhSend(pConn *xuguConn, buff []byte, length int) (int, error) {
	pConn.mu.Lock()
	defer pConn.mu.Unlock()
	//没有内容则不发送
	if length == 0 {
		return 0, nil
	}
	//连接里是否有加密
	if pConn.useSSL {
		fmt.Println("加密未实现")
		return length, nil
		// 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(bufPrintln()
		// 	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 >= 8*1024 {
			//缓冲区空间不足：
			/*- 如果当前缓冲区中的数据加上新数据的长度超过 `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 *xuguConn, i uint32) (int, error) {

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

	//fmt.Println("networkBytes[] ", networkBytes[:])

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

func RhFlush(pConn *xuguConn) (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
}