From 7ae03729f3f294db53138c80256c98dd3622c482 Mon Sep 17 00:00:00 2001 From: John Date: Sun, 28 Apr 2019 23:55:23 +0800 Subject: [PATCH] add package support for gtcp --- g/net/gtcp/gtcp_conn.go | 5 +- g/net/gtcp/gtcp_conn_pkg.go | 113 ++++++++++++++++++++++++++++++++++++ g/net/gtcp/gtcp_func.go | 12 ++-- g/net/gtcp/gtcp_func_pkg.go | 49 ++++++++++++++++ g/net/gtcp/gtcp_pool_pkg.go | 72 +++++++++++++++++++++++ geg/other/test.go | 5 +- 6 files changed, 245 insertions(+), 11 deletions(-) create mode 100644 g/net/gtcp/gtcp_conn_pkg.go create mode 100644 g/net/gtcp/gtcp_func_pkg.go create mode 100644 g/net/gtcp/gtcp_pool_pkg.go diff --git a/g/net/gtcp/gtcp_conn.go b/g/net/gtcp/gtcp_conn.go index be618496b..b1fab08ca 100644 --- a/g/net/gtcp/gtcp_conn.go +++ b/g/net/gtcp/gtcp_conn.go @@ -18,6 +18,7 @@ import ( type Conn struct { conn net.Conn // 底层tcp对象 reader *bufio.Reader // 当前链接的缓冲读取对象 + buffer []byte // 读取缓冲区(用于数据读取时的缓冲区处理) recvDeadline time.Time // 读取超时时间 sendDeadline time.Time // 写入超时时间 recvBufferWait time.Duration // 读取全部缓冲区数据时，读取完毕后的写入等待间隔 @@ -234,8 +235,8 @@ func (c *Conn) SetSendDeadline(t time.Time) error { // 读取全部缓冲区数据时，读取完毕后的写入等待间隔，如果超过该等待时间后仍无可读数据，那么读取操作返回。 // 该时间间隔不能设置得太大，会影响Recv读取时长(默认为1毫秒)。 -func (c *Conn) SetRecvBufferWait(d time.Duration) { - c.recvBufferWait = d +func (c *Conn) SetRecvBufferWait(bufferWaitDuration time.Duration) { + c.recvBufferWait = bufferWaitDuration } func (c *Conn) LocalAddr() net.Addr { diff --git a/g/net/gtcp/gtcp_conn_pkg.go b/g/net/gtcp/gtcp_conn_pkg.go new file mode 100644 index 000000000..b23afd7e0 --- /dev/null +++ b/g/net/gtcp/gtcp_conn_pkg.go @@ -0,0 +1,113 @@ +// Copyright 2019 gf Author(https://github.com/gogf/gf). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package gtcp + +import ( + "encoding/binary" + "errors" + "fmt" + "time" +) + +const ( + // 允许最大的简单协议包大小(byte), 15MB + PKG_MAX_SIZE = 0xFFFFFF + // 消息包头大小: "总长度"3字节+"校验码"4字节 + PKG_HEADER_SIZE = 7 +) + +// 根据简单协议发送数据包。 +// 简单协议数据格式：总长度(24bit)|校验码(32bit)|数据(变长)。 +// 注意： +// 1. "总长度"包含自身3字节及"校验码"4字节。 +// 2. 由于"总长度"为3字节，并且使用的BigEndian字节序，因此最后返回的buffer使用了buffer[1:]。 +func (c *Conn) SendPkg(data []byte, retry...Retry) error { + length := uint32(len(data)) + if length - PKG_HEADER_SIZE > PKG_MAX_SIZE { + return errors.New(fmt.Sprintf(`data size %d exceeds max pkg size %d`, length, PKG_MAX_SIZE)) + } + buffer := make([]byte, PKG_HEADER_SIZE + 1 + len(data)) + copy(buffer[PKG_HEADER_SIZE + 1 : ], data) + binary.BigEndian.PutUint32(buffer[0 : ], PKG_HEADER_SIZE + length) + binary.BigEndian.PutUint32(buffer[4 : ], Checksum(data)) + return c.Send(buffer, retry...) +} + +// 简单协议: 带超时时间的数据发送 +func (c *Conn) SendPkgWithTimeout(data []byte, timeout time.Duration, retry...Retry) error { + c.SetSendDeadline(time.Now().Add(timeout)) + defer c.SetSendDeadline(time.Time{}) + return c.SendPkg(data, retry...) +} + +// 简单协议: 发送数据并等待接收返回数据 +func (c *Conn) SendRecvPkg(data []byte, retry...Retry) ([]byte, error) { + if err := c.SendPkg(data, retry...); err == nil { + return c.RecvPkg(retry...) + } else { + return nil, err + } +} + +// 简单协议: 发送数据并等待接收返回数据(带返回超时等待时间) +func (c *Conn) SendRecvPkgWithTimeout(data []byte, timeout time.Duration, retry...Retry) ([]byte, error) { + if err := c.SendPkg(data, retry...); err == nil { + return c.RecvPkgWithTimeout(timeout, retry...) + } else { + return nil, err + } +} + +// 简单协议: 获取一个数据包。 +func (c *Conn) RecvPkg(retry...Retry) (result []byte, err error) { + var temp []byte + var length uint32 + for { + // 先根据对象的缓冲区数据进行计算 + for { + if len(c.buffer) >= PKG_HEADER_SIZE { + // 注意"总长度"为3个字节，不满足4个字节的uint32类型，因此这里"低位"补0 + length = binary.BigEndian.Uint32([]byte{0, c.buffer[0], c.buffer[1], c.buffer[2]}) + // 解析的大小是否符合规范 + if length == 0 || length - PKG_HEADER_SIZE > PKG_MAX_SIZE { + c.buffer = c.buffer[1:] + continue + } + // 不满足包大小，需要继续读取 + if uint32(len(c.buffer)) < length { + break + } + // 数据校验 + if binary.BigEndian.Uint32(c.buffer[3 : PKG_HEADER_SIZE]) != Checksum(c.buffer[PKG_HEADER_SIZE : length]) { + c.buffer = c.buffer[1:] + continue + } + result = c.buffer[PKG_HEADER_SIZE : length] + c.buffer = c.buffer[length: ] + return + } else { + break + } + } + // 读取系统socket缓冲区的完整数据 + temp, err = c.Recv(-1, retry...) + if err != nil { + break + } + if len(temp) > 0 { + c.buffer = append(c.buffer, temp...) + } + } + return +} + +// 简单协议: 带超时时间的消息包获取 +func (c *Conn) RecvPkgWithTimeout(timeout time.Duration, retry...Retry) ([]byte, error) { + c.SetRecvDeadline(time.Now().Add(timeout)) + defer c.SetRecvDeadline(time.Time{}) + return c.RecvPkg(retry...) +} \ No newline at end of file diff --git a/g/net/gtcp/gtcp_func.go b/g/net/gtcp/gtcp_func.go index dbd3bdc14..e4aacd6ef 100644 --- a/g/net/gtcp/gtcp_func.go +++ b/g/net/gtcp/gtcp_func.go @@ -51,12 +51,12 @@ func Send(addr string, data []byte, retry...Retry) error { // (面向短链接)发送数据并等待接收返回数据 func SendRecv(addr string, data []byte, receive int, retry...Retry) ([]byte, error) { - conn, err := NewConn(addr) - if err != nil { - return nil, err - } - defer conn.Close() - return conn.SendRecv(data, receive, retry...) + conn, err := NewConn(addr) + if err != nil { + return nil, err + } + defer conn.Close() + return conn.SendRecv(data, receive, retry...) } // (面向短链接)带超时时间的数据发送 diff --git a/g/net/gtcp/gtcp_func_pkg.go b/g/net/gtcp/gtcp_func_pkg.go new file mode 100644 index 000000000..164d513cb --- /dev/null +++ b/g/net/gtcp/gtcp_func_pkg.go @@ -0,0 +1,49 @@ +// Copyright 2017 gf Author(https://github.com/gogf/gf). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package gtcp + +import "time" + +// 简单协议: (面向短链接)发送消息包 +func SendPkg(addr string, data []byte, retry...Retry) error { + conn, err := NewConn(addr) + if err != nil { + return err + } + defer conn.Close() + return conn.SendPkg(data, retry...) +} + +// 简单协议: (面向短链接)发送数据并等待接收返回数据 +func SendRecvPkg(addr string, data []byte, retry...Retry) ([]byte, error) { + conn, err := NewConn(addr) + if err != nil { + return nil, err + } + defer conn.Close() + return conn.SendRecvPkg(data, retry...) +} + +// 简单协议: (面向短链接)带超时时间的数据发送 +func SendPkgWithTimeout(addr string, data []byte, timeout time.Duration, retry...Retry) error { + conn, err := NewConn(addr) + if err != nil { + return err + } + defer conn.Close() + return conn.SendPkgWithTimeout(data, timeout, retry...) +} + +// 简单协议: (面向短链接)发送数据并等待接收返回数据(带返回超时等待时间) +func SendRecvPkgWithTimeout(addr string, data []byte, timeout time.Duration, retry...Retry) ([]byte, error) { + conn, err := NewConn(addr) + if err != nil { + return nil, err + } + defer conn.Close() + return conn.SendRecvPkgWithTimeout(data, timeout, retry...) +} \ No newline at end of file diff --git a/g/net/gtcp/gtcp_pool_pkg.go b/g/net/gtcp/gtcp_pool_pkg.go new file mode 100644 index 000000000..f22b60da6 --- /dev/null +++ b/g/net/gtcp/gtcp_pool_pkg.go @@ -0,0 +1,72 @@ +// Copyright 2018 gf Author(https://github.com/gogf/gf). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package gtcp + +import ( + "time" +) + +// 简单协议: (方法覆盖)发送数据 +func (c *PoolConn) SendPkg(data []byte, retry...Retry) (err error) { + if err = c.Conn.SendPkg(data, retry...); err != nil && c.status == gCONN_STATUS_UNKNOWN { + if v, e := c.pool.NewFunc(); e == nil { + c.Conn = v.(*PoolConn).Conn + err = c.Conn.SendPkg(data, retry...) + } else { + err = e + } + } + if err != nil { + c.status = gCONN_STATUS_ERROR + } else { + c.status = gCONN_STATUS_ACTIVE + } + return err +} + +// 简单协议: (方法覆盖)接收数据 +func (c *PoolConn) RecvPkg(retry...Retry) ([]byte, error) { + data, err := c.Conn.RecvPkg(retry...) + if err != nil { + c.status = gCONN_STATUS_ERROR + } else { + c.status = gCONN_STATUS_ACTIVE + } + return data, err +} + +// 简单协议: (方法覆盖)带超时时间的数据获取 +func (c *PoolConn) RecvPkgWithTimeout(timeout time.Duration, retry...Retry) ([]byte, error) { + c.SetRecvDeadline(time.Now().Add(timeout)) + defer c.SetRecvDeadline(time.Time{}) + return c.RecvPkg(retry...) +} + +// 简单协议: (方法覆盖)带超时时间的数据发送 +func (c *PoolConn) SendPkgWithTimeout(data []byte, timeout time.Duration, retry...Retry) error { + c.SetSendDeadline(time.Now().Add(timeout)) + defer c.SetSendDeadline(time.Time{}) + return c.SendPkg(data, retry...) +} + +// 简单协议: (方法覆盖)发送数据并等待接收返回数据 +func (c *PoolConn) SendRecvPkg(data []byte, retry...Retry) ([]byte, error) { + if err := c.SendPkg(data, retry...); err == nil { + return c.RecvPkg(retry...) + } else { + return nil, err + } +} + +// 简单协议: (方法覆盖)发送数据并等待接收返回数据(带返回超时等待时间) +func (c *PoolConn) SendRecvPkgWithTimeout(data []byte, timeout time.Duration, retry...Retry) ([]byte, error) { + if err := c.SendPkg(data, retry...); err == nil { + return c.RecvPkgWithTimeout(timeout, retry...) + } else { + return nil, err + } +} \ No newline at end of file diff --git a/geg/other/test.go b/geg/other/test.go index 18343ac6c..ac6545942 100644 --- a/geg/other/test.go +++ b/geg/other/test.go @@ -1,11 +1,10 @@ package main import ( + "encoding/binary" "fmt" ) func main() { - array := make([]interface{}, 0, 10) - array[8] = 1 - fmt.Println(array) + fmt.Println(binary.BigEndian.Uint32([]byte{byte(1), byte(1), byte(1), byte(1), byte(1)})) } \ No newline at end of file