完成gtcp/gudp包的改进

2026-06-07 02:12:11 +08:00 · 2018-07-18 11:43:30 +08:00
parent 7601edecb9
commit 61a0a163b7
27 changed files with 790 additions and 383 deletions
--- a/3
+++ b/3
@ -13,6 +13,9 @@ gtime增加对时区转换的封装，并简化失去转换时对类似+80500时
 改进gf-orm的where查询功能，参考thinkphp 里的where查询语法；
 gpage分页增加对自定义后缀的支持，如:2.html, 2.php等等；
 gvalid包增加tag与校验规则绑定的支持特性；
+ghttp获取参数支持直接转struct功能；
+map转struct增加对tag的支持；
+

 DONE:
 1.  gconv完善针对不同类型的判断，例如：尽量减少sprintf("%v", xxx)来执行string类型的转换；
--- a/g/net/gtcp/gtcp.go
+++ b/g/net/gtcp/gtcp.go
--- a/g/net/gtcp/gtcp_conn.go
+++ b/g/net/gtcp/gtcp_conn.go
@ -0,0 +1,159 @@
+// Copyright 2018 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
+
+package gtcp
+
+import (
+    "net"
+    "time"
+    "io"
+    "bufio"
+)
+
+// 封装的链接对象
+type Conn struct {
+    net.Conn
+    reader *bufio.Reader // 当前链接的缓冲读取对象
+}
+
+// 创建TCP链接
+func NewConn(addr string, timeout...int) (*Conn, error) {
+    if conn, err := NewNetConn(addr, timeout...); err == nil {
+        return &Conn {
+            Conn : conn,
+        }, nil
+    } else {
+        return nil, err
+    }
+}
+
+// 将net.Conn接口对象转换为*gtcp.Conn对象(注意递归影响，因为*gtcp.Conn本身也实现了net.Conn接口)
+func NewConnByNetConn(conn net.Conn) *Conn {
+    return &Conn { Conn: conn }
+}
+
+// 发送数据
+func (c *Conn) Send(data []byte, retry...Retry) error {
+    length := 0
+    for {
+        n, err := c.Write(data)
+        if err != nil {
+            // 链接已关闭
+            if err == io.EOF {
+                return err
+            }
+            // 其他错误，重试之后仍不能成功
+            if len(retry) == 0 || retry[0].Count == 0 {
+                return err
+            }
+            if len(retry) > 0 {
+                retry[0].Count--
+                if retry[0].Interval == 0 {
+                    retry[0].Interval = gDEFAULT_RETRY_INTERVAL
+                }
+                time.Sleep(time.Duration(retry[0].Interval) * time.Millisecond)
+            }
+        } else {
+            length += n
+            if length == len(data) {
+                return nil
+            }
+        }
+    }
+}
+
+// 获取数据，指定读取的数据长度(length < 1表示获取所有可读数据)，以及重试策略(retry)
+// 需要注意：
+// 1、往往在socket通信中需要指定固定的数据结构，并在设定对应的长度字段，并在读取数据时便于区分包大小；
+// 2、当length < 1时表示获取缓冲区所有的数据，但是可能会引起包解析问题(可能出现非完整的包情况)，因此需要解析端注意解析策略；
+func (c *Conn) Receive(length int, retry...Retry) ([]byte, error) {
+    var err    error  // 读取错误
+    var size   int    // 读取长度
+    var index  int    // 已读取长度
+    var buffer []byte // 读取缓冲区
+
+    if c.reader == nil {
+        c.reader = bufio.NewReader(c)
+    }
+    if length > 0 {
+        buffer = make([]byte, length)
+    } else {
+        buffer = make([]byte, gDEFAULT_READ_BUFFER_SIZE)
+    }
+    for {
+        size, err = c.reader.Read(buffer[index:])
+        if size > 0 {
+            index += size
+            if length > 0 {
+                // 如果指定了读取大小，那么必须读取到指定长度才返回
+                if index == length {
+                    break
+                }
+            } else {
+                // 否则读取所有缓冲区数据，直到没有可读数据为止
+                if c.reader.Buffered() < 1 {
+                    break
+                }
+                // 如果长度超过了自定义的读取缓冲区，那么自动增长
+                if index >= gDEFAULT_READ_BUFFER_SIZE {
+                    buffer = append(buffer, make([]byte, gDEFAULT_READ_BUFFER_SIZE)...)
+                }
+            }
+        }
+        if err != nil {
+            // 链接已关闭
+            if err == io.EOF {
+                break
+            }
+            if len(retry) > 0 {
+                // 其他错误，重试之后仍不能成功
+                if retry[0].Count == 0 {
+                    break
+                }
+                retry[0].Count--
+                if retry[0].Interval == 0 {
+                    retry[0].Interval = gDEFAULT_RETRY_INTERVAL
+                }
+                time.Sleep(time.Duration(retry[0].Interval) * time.Millisecond)
+                continue
+            }
+            break
+        }
+    }
+    return buffer[:index], err
+}
+
+// 带超时时间的数据获取
+func (c *Conn) ReceiveWithTimeout(length int, timeout time.Duration, retry...Retry) ([]byte, error) {
+    c.SetReadDeadline(time.Now().Add(timeout))
+    defer c.SetReadDeadline(time.Time{})
+    return c.Receive(length, retry...)
+}
+
+// 带超时时间的数据发送
+func (c *Conn) SendWithTimeout(data []byte, timeout time.Duration, retry...Retry) error {
+    c.SetWriteDeadline(time.Now().Add(timeout))
+    defer c.SetWriteDeadline(time.Time{})
+    return c.Send(data, retry...)
+}
+
+// 发送数据并等待接收返回数据
+func (c *Conn) SendReceive(data []byte, receive int, retry...Retry) ([]byte, error) {
+    if err := c.Send(data, retry...); err == nil {
+        return c.Receive(receive, retry...)
+    } else {
+        return nil, err
+    }
+}
+
+// 发送数据并等待接收返回数据(带返回超时等待时间)
+func (c *Conn) SendReceiveWithTimeout(data []byte, receive int, timeout time.Duration, retry...Retry) ([]byte, error) {
+    if err := c.Send(data, retry...); err == nil {
+        return c.ReceiveWithTimeout(receive, timeout, retry...)
+    } else {
+        return nil, err
+    }
+}
--- a/g/net/gtcp/gtcp_func.go
+++ b/g/net/gtcp/gtcp_func.go
@ -0,0 +1,80 @@
+// Copyright 2017 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
+
+package gtcp
+
+import (
+    "net"
+    "time"
+)
+
+const (
+    gDEFAULT_RETRY_INTERVAL   = 100   // (毫秒)默认重试时间间隔
+    gDEFAULT_READ_BUFFER_SIZE = 1024  // 默认数据读取缓冲区大小
+)
+
+type Retry struct {
+    Count    int  // 重试次数
+    Interval int  // 重试间隔(毫秒)
+}
+
+// 常见的二进制数据校验方式，生成校验结果
+func Checksum(buffer []byte) uint32 {
+    var checksum uint32
+    for _, b := range buffer {
+        checksum += uint32(b)
+    }
+    return checksum
+}
+
+// 创建原生TCP链接, addr地址格式形如：127.0.0.1:80
+func NewNetConn(addr string, timeout...int) (net.Conn, error) {
+    if len(timeout) > 0 {
+        return net.DialTimeout("tcp", addr, time.Duration(timeout[0]) * time.Millisecond)
+    } else {
+        return net.Dial("tcp", addr)
+    }
+}
+
+// (面向短链接)发送数据
+func Send(addr string, data []byte, retry...Retry) error {
+    conn, err := NewConn(addr)
+    if err != nil {
+        return err
+    }
+    defer conn.Close()
+    return conn.Send(data, retry...)
+}
+
+// (面向短链接)发送数据并等待接收返回数据
+func SendReceive(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.SendReceive(data, receive, retry...)
+}
+
+// (面向短链接)带超时时间的数据发送
+func SendWithTimeout(addr string, data []byte, timeout time.Duration, retry...Retry) error {
+    conn, err := NewConn(addr)
+    if err != nil {
+        return err
+    }
+    defer conn.Close()
+    return conn.SendWithTimeout(data, timeout, retry...)
+}
+
+// (面向短链接)发送数据并等待接收返回数据(带返回超时等待时间)
+func SendReceiveWithTimeout(addr string, data []byte, receive int, timeout time.Duration, retry...Retry) ([]byte, error) {
+    conn, err := NewConn(addr)
+    if err != nil {
+        return nil, err
+    }
+    defer conn.Close()
+    return conn.SendReceiveWithTimeout(data, receive, timeout, retry...)
+}
--- a/g/net/gtcp/gtcp_pool.go
+++ b/g/net/gtcp/gtcp_pool.go
@ -7,15 +7,14 @@
 package gtcp

 import (
-    "net"
    "time"
    "gitee.com/johng/gf/g/container/gmap"
    "gitee.com/johng/gf/g/container/gpool"
 )

-// 封装的链接对象
-type Conn struct {
-    net.Conn           // 继承底层链接接口对象
+// 链接池链接对象
+type PoolConn struct {
+    *Conn              // 继承底层链接接口对象
    pool   *gpool.Pool // 对应的链接池对象
    status int         // 当前对象的状态，主要用于失败重连判断
 }
@ -33,8 +32,8 @@ var (
    pools = gmap.NewStringInterfaceMap()
 )

-// 创建TCP链接
-func NewConn(addr string, timeout...int) (*Conn, error) {
+// 创建TCP链接池对象
+func NewPoolConn(addr string, timeout...int) (*PoolConn, error) {
    var pool *gpool.Pool
    if v := pools.Get(addr); v == nil {
        pools.LockFunc(func(m map[string]interface{}) {
@ -42,11 +41,8 @@ func NewConn(addr string, timeout...int) (*Conn, error) {
                pool = v.(*gpool.Pool)
            } else {
                pool = gpool.New(gDEFAULT_POOL_EXPIRE, func() (interface{}, error) {
-                    if conn, err := NewNetConn(addr, timeout...); err == nil {
-                        return &Conn {
-                            Conn   : conn,
-                            pool   : pool,
-                        }, nil
+                    if conn, err := NewConn(addr, timeout...); err == nil {
+                        return &PoolConn { conn, pool, gCONN_STATUS_ACTIVE }, nil
                    } else {
                        return nil, err
                    }
@ -59,21 +55,14 @@ func NewConn(addr string, timeout...int) (*Conn, error) {
    }

    if v, err := pool.Get(); err == nil {
-        return v.(*Conn), nil
+        return v.(*PoolConn), nil
    } else {
        return nil, err
    }
 }

-// 将net.Conn接口对象转换为*gtcp.Conn对象(注意递归影响，因为*gtcp.Conn本身也实现了net.Conn接口)
-func NewConnByNetConn(conn net.Conn) *Conn {
-    return &Conn {
-        Conn : conn,
-    }
-}
-
 // 覆盖底层接口对象的Close方法
-func (c *Conn) Close() error {
+func (c *PoolConn) Close() error {
    if c.pool != nil && c.status == gCONN_STATUS_ACTIVE {
        c.status = gCONN_STATUS_UNKNOWN
        c.pool.Put(c)
@ -84,12 +73,12 @@ func (c *Conn) Close() error {
 }

 // 发送数据
-func (c *Conn) Send(data []byte, retry...Retry) error {
+func (c *PoolConn) Send(data []byte, retry...Retry) error {
    var err error
-    if err = Send(c, data, retry...); err != nil && c.status == gCONN_STATUS_UNKNOWN {
+    if err = c.Send(data, retry...); err != nil && c.status == gCONN_STATUS_UNKNOWN {
        if v, e := c.pool.NewFunc(); e == nil {
-            c.Conn = v.(net.Conn)
-            err    = Send(c, data, retry...)
+            c.Conn = v.(*PoolConn).Conn
+            err    = c.Send(data, retry...)
        } else {
            err    = e
        }
@ -103,16 +92,8 @@ func (c *Conn) Send(data []byte, retry...Retry) error {
 }

 // 接收数据
-func (c *Conn) Receive(retry...Retry) ([]byte, error) {
-    data, err := Receive(c, retry...)
-    if err != nil && len(data) == 0 && c.status == gCONN_STATUS_UNKNOWN {
-        if v, e := c.pool.NewFunc(); e == nil {
-            c.Conn    = v.(net.Conn)
-            data, err = Receive(c, retry...)
-        } else {
-            err       = e
-        }
-    }
+func (c *PoolConn) Receive(length int, retry...Retry) ([]byte, error) {
+    data, err := c.Receive(length, retry...)
    if err != nil {
        c.status = gCONN_STATUS_ERROR
    } else {
@ -122,30 +103,32 @@ func (c *Conn) Receive(retry...Retry) ([]byte, error) {
 }

 // 带超时时间的数据获取
-func (c *Conn) ReceiveWithTimeout(timeout time.Duration, retry...Retry) ([]byte, error) {
+func (c *PoolConn) ReceiveWithTimeout(length int, timeout time.Duration, retry...Retry) ([]byte, error) {
    c.SetReadDeadline(time.Now().Add(timeout))
-    return c.Receive(retry...)
+    defer c.SetReadDeadline(time.Time{})
+    return c.Receive(length, retry...)
 }

 // 带超时时间的数据发送
-func (c *Conn) SendWithTimeout(data []byte, timeout time.Duration, retry...Retry) error {
+func (c *PoolConn) SendWithTimeout(data []byte, timeout time.Duration, retry...Retry) error {
    c.SetWriteDeadline(time.Now().Add(timeout))
+    defer c.SetWriteDeadline(time.Time{})
    return c.Send(data, retry...)
 }

 // 发送数据并等待接收返回数据
-func (c *Conn) SendReceive(data []byte, retry...Retry) ([]byte, error) {
+func (c *PoolConn) SendReceive(data []byte, receive int, retry...Retry) ([]byte, error) {
    if err := c.Send(data, retry...); err == nil {
-        return c.Receive(retry...)
+        return c.Receive(receive, retry...)
    } else {
        return nil, err
    }
 }

 // 发送数据并等待接收返回数据(带返回超时等待时间)
-func (c *Conn) SendReceiveWithTimeout(data []byte, timeout time.Duration, retry...Retry) ([]byte, error) {
+func (c *PoolConn) SendReceiveWithTimeout(data []byte, receive int, timeout time.Duration, retry...Retry) ([]byte, error) {
    if err := c.Send(data, retry...); err == nil {
-        return c.ReceiveWithTimeout(timeout, retry...)
+        return c.ReceiveWithTimeout(receive, timeout, retry...)
    } else {
        return nil, err
    }
--- a/g/net/gtcp/gtcp_server.go
+++ b/g/net/gtcp/gtcp_server.go
@ -28,7 +28,7 @@ func (s *Server) Run() error {
    if s.handler == nil {
        return errors.New("start running failed: socket handler not defined")
    }
-    tcpaddr, err := net.ResolveTCPAddr("tcp4", s.address)
+    tcpaddr, err := net.ResolveTCPAddr("tcp", s.address)
    if err != nil {
        return err
    }
--- a/g/net/gtcp/tcp_func.go
+++ b/g/net/gtcp/tcp_func.go
@ -1,156 +0,0 @@
-// Copyright 2017 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
-
-package gtcp
-
-import (
-    "io"
-    "net"
-    "time"
-    "bytes"
-)
-
-const (
-    gDEFAULT_RETRY_INTERVAL   = 100   // (毫秒)默认重试时间间隔
-    gDEFAULT_READ_BUFFER_SIZE = 1024  // 默认数据读取缓冲区大小
-
-)
-
-type Retry struct {
-    Count    int  // 重试次数
-    Interval int  // 重试间隔(毫秒)
-}
-
-// 自定义的包分割符号，用于标识包是否读取结束
-// 注意：
-// 1. 必须使用gtcp包来发送和接收tcp数据才有效；
-// 2. 只有在发送的字节数为buffer size倍数时才有效；
-var pkgSplitMark = []byte{0, 'E', 'O', 'P', 0}
-
-// 常见的二进制数据校验方式，生成校验结果
-func Checksum(buffer []byte) uint32 {
-    var checksum uint32
-    for _, b := range buffer {
-        checksum += uint32(b)
-    }
-    return checksum
-}
-
-// 创建原生TCP链接, addr地址格式形如：127.0.0.1:80
-func NewNetConn(addr string, timeout...int) (net.Conn, error) {
-    if len(timeout) > 0 {
-        return net.DialTimeout("tcp", addr, time.Duration(timeout[0]) * time.Millisecond)
-    } else {
-        return net.Dial("tcp", addr)
-    }
-}
-
-// 获取数据
-func Receive(conn net.Conn, retry...Retry) ([]byte, error) {
-    var err    error
-    var length int
-    var buffer []byte
-    size := gDEFAULT_READ_BUFFER_SIZE
-    data := make([]byte, 0)
-    for {
-        buffer      = make([]byte, size)
-        length, err = conn.Read(buffer)
-        // 这里使用 "&&" 只要有数据不管有无错误都将先进行解析
-        if length < 1 && err != nil {
-            // 链接已关闭
-            if err == io.EOF {
-                break
-            }
-            if len(retry) > 0 {
-                // 其他错误，重试之后仍不能成功
-                if retry[0].Count == 0 {
-                    break
-                }
-                retry[0].Count--
-                if retry[0].Interval == 0 {
-                    retry[0].Interval = gDEFAULT_RETRY_INTERVAL
-                }
-                time.Sleep(time.Duration(retry[0].Interval) * time.Millisecond)
-                continue
-            }
-            break
-        } else {
-            // 自定义结束标识符判断
-            if length == len(pkgSplitMark) && bytes.Compare(pkgSplitMark, buffer[0 : length]) == 0 {
-                break
-            }
-            data = append(data, buffer[0 : length]...)
-            if length < gDEFAULT_READ_BUFFER_SIZE || err == io.EOF {
-                break
-            }
-        }
-    }
-    return data, err
-}
-
-// 带超时时间的数据获取
-func ReceiveWithTimeout(conn net.Conn, timeout time.Duration, retry...Retry) ([]byte, error) {
-    conn.SetReadDeadline(time.Now().Add(timeout))
-    return Receive(conn, retry...)
-}
-
-// 发送数据
-func Send(conn net.Conn, data []byte, retry...Retry) error {
-    if len(data) % gDEFAULT_READ_BUFFER_SIZE == 0 {
-        data = append(data, pkgSplitMark...)
-    }
-    length := 0
-    for {
-        n, err := conn.Write(data)
-        if err != nil {
-            // 链接已关闭
-            if err == io.EOF {
-                return err
-            }
-            // 其他错误，重试之后仍不能成功
-            if len(retry) == 0 || retry[0].Count == 0 {
-                return err
-            }
-            if len(retry) > 0 {
-                retry[0].Count--
-                if retry[0].Interval == 0 {
-                    retry[0].Interval = gDEFAULT_RETRY_INTERVAL
-                }
-                time.Sleep(time.Duration(retry[0].Interval) * time.Millisecond)
-            }
-        } else {
-            length += n
-            if length == len(data) {
-                return nil
-            }
-        }
-    }
-    return nil
-}
-
-// 带超时时间的数据发送
-func SendWithTimeout(conn net.Conn, data []byte, timeout time.Duration, retry...Retry) error {
-    conn.SetWriteDeadline(time.Now().Add(timeout))
-    return Send(conn, data, retry...)
-}
-
-// 发送数据并等待接收返回数据
-func SendReceive(conn net.Conn, data []byte, retry...Retry) ([]byte, error) {
-    if err := Send(conn, data, retry...); err == nil {
-        return Receive(conn, retry...)
-    } else {
-        return nil, err
-    }
-}
-
-// 发送数据并等待接收返回数据(带返回超时等待时间)
-func SendReceiveWithTimeout(conn net.Conn, data []byte, timeout time.Duration, retry...Retry) ([]byte, error) {
-    if err := Send(conn, data, retry...); err == nil {
-        return ReceiveWithTimeout(conn, timeout, retry...)
-    } else {
-        return nil, err
-    }
-}
--- a/g/net/gudp/gudp.go
+++ b/g/net/gudp/gudp.go
@ -0,0 +1,11 @@
+// Copyright 2017 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
+
+// UDP
+package gudp
+
+
+
--- a/g/net/gudp/gudp_conn.go
+++ b/g/net/gudp/gudp_conn.go
@ -0,0 +1,145 @@
+// Copyright 2018 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
+
+package gudp
+
+import (
+    "net"
+    "time"
+    "io"
+)
+
+// 封装的链接对象
+type Conn struct {
+    *net.UDPConn
+    raddr *net.UDPAddr
+}
+
+const (
+    gDEFAULT_RETRY_INTERVAL   = 100   // (毫秒)默认重试时间间隔
+    gDEFAULT_READ_BUFFER_SIZE = 1024  // 默认数据读取缓冲区大小
+)
+
+type Retry struct {
+    Count    int  // 重试次数
+    Interval int  // 重试间隔(毫秒)
+}
+
+// 创建TCP链接
+func NewConn(raddr string, laddr...string) (*Conn, error) {
+    if conn, err := NewNetConn(raddr, laddr...); err == nil {
+        return NewConnByNetConn(conn), nil
+    } else {
+        return nil, err
+    }
+}
+
+// 将*net.UDPConn对象转换为*Conn对象
+func NewConnByNetConn(udp *net.UDPConn) *Conn {
+    return &Conn {
+        UDPConn : udp,
+    }
+}
+
+// 发送数据
+func (c *Conn) Send(data []byte, retry...Retry) error {
+    var err     error
+    var size    int
+    var length  int
+    for {
+        if c.raddr != nil {
+            size, err = c.WriteToUDP(data, c.raddr)
+        } else {
+            size, err = c.Write(data)
+        }
+        if err != nil {
+            // 链接已关闭
+            if err == io.EOF {
+                return err
+            }
+            // 其他错误，重试之后仍不能成功
+            if len(retry) == 0 || retry[0].Count == 0 {
+                return err
+            }
+            if len(retry) > 0 {
+                retry[0].Count--
+                if retry[0].Interval == 0 {
+                    retry[0].Interval = gDEFAULT_RETRY_INTERVAL
+                }
+                time.Sleep(time.Duration(retry[0].Interval) * time.Millisecond)
+            }
+        } else {
+            length += size
+            if length == len(data) {
+                return nil
+            }
+        }
+    }
+    return nil
+}
+
+// 接收数据
+func (c *Conn) Receive(length int, retry...Retry) ([]byte, error) {
+    var err     error         // 读取错误
+    var size    int           // 读取长度
+    var index   int           // 已读取长度
+    var buffer  []byte        // 读取缓冲区
+
+    if length > 0 {
+        buffer = make([]byte, length)
+    } else {
+        buffer = make([]byte, gDEFAULT_READ_BUFFER_SIZE)
+    }
+    for {
+        size, c.raddr, err = c.ReadFromUDP(buffer[index:])
+        if size > 0 {
+            index += size
+            if length > 0 {
+                // 如果指定了读取大小，那么必须读取到指定长度才返回
+                if index == length {
+                    break
+                }
+            } else {
+                // 如果长度超过了自定义的读取缓冲区，那么自动增长
+                if index >= gDEFAULT_READ_BUFFER_SIZE {
+                    buffer = append(buffer, make([]byte, gDEFAULT_READ_BUFFER_SIZE)...)
+                }
+                if size < gDEFAULT_READ_BUFFER_SIZE {
+                    break
+                }
+            }
+        }
+        if err != nil {
+            // 链接已关闭
+            if err == io.EOF {
+                break
+            }
+            if len(retry) > 0 {
+                // 其他错误，重试之后仍不能成功
+                if retry[0].Count == 0 {
+                    break
+                }
+                retry[0].Count--
+                if retry[0].Interval == 0 {
+                    retry[0].Interval = gDEFAULT_RETRY_INTERVAL
+                }
+                time.Sleep(time.Duration(retry[0].Interval) * time.Millisecond)
+                continue
+            }
+            break
+        }
+    }
+    return buffer[:index], err
+}
+
+// 发送数据并等待接收返回数据
+func (c *Conn) SendReceive(data []byte, receive int, retry...Retry) ([]byte, error) {
+    if err := c.Send(data, retry...); err == nil {
+        return c.Receive(receive, retry...)
+    } else {
+        return nil, err
+    }
+}
--- a/g/net/gudp/gudp_func.go
+++ b/g/net/gudp/gudp_func.go
@ -0,0 +1,50 @@
+// Copyright 2017-2018 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
+
+package gudp
+
+import "net"
+
+// 创建标准库UDP链接操作对象
+func NewNetConn(raddr string, laddr...string) (*net.UDPConn, error) {
+    var err error
+    var rudpaddr, ludpaddr *net.UDPAddr
+    rudpaddr, err = net.ResolveUDPAddr("udp", raddr)
+    if err != nil {
+        return nil, err
+    }
+    if len(laddr) > 0 {
+        ludpaddr, err = net.ResolveUDPAddr("udp", laddr[0])
+        if err != nil {
+            return nil, err
+        }
+    }
+    conn, err := net.DialUDP("udp", ludpaddr, rudpaddr)
+    if err != nil {
+        return nil, err
+    }
+    return conn, nil
+}
+
+// (面向短链接)发送数据
+func Send(addr string, data []byte, retry...Retry) error {
+    conn, err := NewConn(addr)
+    if err != nil {
+        return err
+    }
+    defer conn.Close()
+    return conn.Send(data, retry...)
+}
+
+// (面向短链接)发送数据并等待接收返回数据
+func SendReceive(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.SendReceive(data, receive, retry...)
+}
--- a/g/net/gudp/gudp_server.go
+++ b/g/net/gudp/gudp_server.go
@ -4,11 +4,12 @@
 // If a copy of the MIT was not distributed with this file,
 // You can obtain one at https://gitee.com/johng/gf.

-// UDP服务端
+
 package gudp

 import (
    "net"
+    "errors"
    "gitee.com/johng/gf/g/container/gmap"
    "gitee.com/johng/gf/g/util/gconv"
 )
@ -20,7 +21,7 @@ const (
 // tcp server结构体
 type Server struct {
    address   string
-    handler   func (*net.UDPConn)
+    handler   func (*Conn)
 }

 // Server表，用以存储和检索名称与Server对象之间的关联关系
@ -42,7 +43,7 @@ func GetServer(name...interface{}) (*Server) {
 }

 // 创建一个tcp server对象，并且可以选择指定一个单例名字
-func NewServer (address string, handler func (*net.UDPConn), names...string) *Server {
+func NewServer (address string, handler func (*Conn), names...string) *Server {
    s := &Server{address, handler}
    if len(names) > 0 {
        serverMapping.Set(names[0], s)
@ -50,3 +51,31 @@ func NewServer (address string, handler func (*net.UDPConn), names...string) *Se
    return s
 }

+// 设置参数 - address
+func (s *Server) SetAddress (address string) {
+    s.address = address
+}
+
+// 设置参数 - handler
+func (s *Server) SetHandler (handler func (*Conn)) {
+    s.handler = handler
+}
+
+// 执行监听
+func (s *Server) Run() error {
+    if s.handler == nil {
+        return errors.New("start running failed: socket handler not defined")
+    }
+    addr, err := net.ResolveUDPAddr("udp", s.address)
+    if err != nil {
+        return err
+    }
+    conn, err := net.ListenUDP("udp", addr)
+    if err != nil {
+        return err
+    }
+    for {
+        s.handler(NewConnByNetConn(conn))
+    }
+    return nil
+}
--- a/g/net/gudp/udp_server.go
+++ b/g/net/gudp/udp_server.go
@ -1,42 +0,0 @@
-// Copyright 2017 gf Author(https://gitee.com/johng/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://gitee.com/johng/gf.
-
-
-package gudp
-
-import (
-    "net"
-    "errors"
-)
-
-// 设置参数 - address
-func (s *Server) SetAddress (address string) {
-    s.address = address
-}
-
-// 设置参数 - handler
-func (s *Server) SetHandler (handler func (*net.UDPConn)) {
-    s.handler = handler
-}
-
-// 执行监听
-func (s *Server) Run() error {
-    if s.handler == nil {
-        return errors.New("start running failed: socket handler not defined")
-    }
-    tcpaddr, err := net.ResolveUDPAddr("udp4", s.address)
-    if err != nil {
-        return err
-    }
-    listen, err := net.ListenUDP("udp", tcpaddr)
-    if err != nil {
-        return err
-    }
-    for {
-        s.handler(listen)
-    }
-    return nil
-}
--- a/g/os/gproc/gproc_comm_receive.go
+++ b/g/os/gproc/gproc_comm_receive.go
@ -65,7 +65,7 @@ func tcpServiceHandler(conn *gtcp.Conn) {
    var retry = gtcp.Retry{3, 10}
    for {
        var result []byte
-        buffer, err := conn.Receive(retry)
+        buffer, err := conn.Receive(-1, retry)
        if len(buffer) > 0 {
            var msgs []*Msg
            for _, msg := range bufferToMsgs(buffer) {
--- a/g/os/gproc/gproc_comm_send.go
+++ b/g/os/gproc/gproc_comm_send.go
@ -46,7 +46,7 @@ func Send(pid int, data []byte, group...string) error {
    for i := gPROC_COMM_FAILURE_RETRY_COUNT; i > 0; i-- {
        if conn, err = getConnByPid(pid); err == nil {
            defer conn.Close()
-            buf, err = conn.SendReceiveWithTimeout(buffer, gPROC_COMM_SEND_TIMEOUT*time.Millisecond)
+            buf, err = conn.SendReceiveWithTimeout(buffer, -1, gPROC_COMM_SEND_TIMEOUT*time.Millisecond)
            if len(buf) > 0 {
                // 如果有返回值，如果不是"ok"，那么表示是错误信息
                if !bytes.EqualFold(buf, []byte("ok")) {
--- a/geg/net/gtcp/gtcp_conn_client.go
+++ b/geg/net/gtcp/gtcp_conn_client.go
@ -1,21 +0,0 @@
-package main
-
-import (
-    "gitee.com/johng/gf/g/net/gtcp"
-    "gitee.com/johng/gf/g/util/gconv"
-    "gitee.com/johng/gf/g/os/glog"
-    "time"
-)
-
-func main() {
-    conn, err := gtcp.NewConn("127.0.0.1:8999")
-    if err != nil {
-        glog.Fatal(err)
-    }
-    for i := 0; i < 10000; i++ {
-        if err := conn.Send([]byte(gconv.String(i))); err != nil {
-            glog.Error(err)
-        }
-        time.Sleep(time.Second)
-    }
-}
--- a/geg/net/gtcp/gtcp_conn_server.go
+++ b/geg/net/gtcp/gtcp_conn_server.go
@ -1,19 +0,0 @@
-package main
-
-import (
-    "net"
-    "gitee.com/johng/gf/g/os/glog"
-    "gitee.com/johng/gf/g/net/gtcp"
-)
-
-func main() {
-    gtcp.NewServer(":8999", func(conn net.Conn) {
-        c := gtcp.NewConnByNetConn(conn)
-        defer c.Close()
-        for {
-            if data, err := c.Receive(); err == nil {
-                glog.Println(string(data))
-            }
-        }
-    }).Run()
-}
--- a/geg/net/gtcp/gtcp_pool1.go
+++ b/geg/net/gtcp/gtcp_pool1.go
@ -15,10 +15,12 @@ func main() {
        c := gtcp.NewConnByNetConn(conn)
        defer c.Close()
        for {
-            if data, _ := c.Receive(); len(data) > 0 {
-                c.Send(append([]byte("> "), data...))
+            if data, _ := c.Receive(-1); len(data) > 0 {
+                if err := c.Send(append([]byte("> "), data...)); err != nil {
+                   fmt.Println(err)
+                }
            }
-            //return
+            // return
        }
    }).Run()

@ -26,8 +28,8 @@ func main() {

    // Client
    for {
-       if conn, err := gtcp.NewConn("127.0.0.1:8999"); err == nil {
-           if b, err := conn.SendReceive([]byte(gtime.Datetime())); err == nil {
+       if conn, err := gtcp.NewPoolConn("127.0.0.1:8999"); err == nil {
+           if b, err := conn.SendReceive([]byte(gtime.Datetime()), -1); err == nil {
               fmt.Println(string(b), conn.LocalAddr(), conn.RemoteAddr())
           } else {
               fmt.Println(err)
--- a/geg/net/gtcp/gtcp_pool2.go
+++ b/geg/net/gtcp/gtcp_pool2.go
@ -15,8 +15,10 @@ func main() {
        c := gtcp.NewConnByNetConn(conn)
        defer c.Close()
        for {
-            if data, _ := c.Receive(); len(data) > 0 {
-                c.Send(append([]byte("> "), data...))
+            if data, _ := c.Receive(-1); len(data) > 0 {
+                if err := c.Send(append([]byte("> "), data...)); err != nil {
+                    fmt.Println(err)
+                }
            }
            return
        }
@ -26,8 +28,8 @@ func main() {

    // Client
    for {
-       if conn, err := gtcp.NewConn("127.0.0.1:8999"); err == nil {
-           if b, err := conn.SendReceive([]byte(gtime.Datetime())); err == nil {
+       if conn, err := gtcp.NewPoolConn("127.0.0.1:8999"); err == nil {
+           if b, err := conn.SendReceive([]byte(gtime.Datetime()), -1); err == nil {
               fmt.Println(string(b), conn.LocalAddr(), conn.RemoteAddr())
           } else {
               fmt.Println(err)
@ -38,4 +40,6 @@ func main() {
       }
       time.Sleep(time.Second)
    }
+
+    select{}
 }
--- a/geg/net/gtcp/gtcp_server.go
+++ b/geg/net/gtcp/gtcp_server.go
@ -1,17 +0,0 @@
-package main
-
-import (
-    "net"
-    "gitee.com/johng/gf/g/net/gtcp"
-)
-
-func main() {
-    gtcp.NewServer(":8999", func(conn net.Conn) {
-        defer conn.Close()
-        for {
-            if data, err := gtcp.Receive(conn); err == nil {
-                gtcp.Send(conn, append([]byte("> "), data...))
-            }
-        }
-    }).Run()
-}
--- a/geg/net/gtcp/gtcp_server_client.go
+++ b/geg/net/gtcp/gtcp_server_client.go
@ -0,0 +1,43 @@
+package main
+
+import (
+    "fmt"
+    "time"
+    "net"
+    "gitee.com/johng/gf/g/net/gtcp"
+    "gitee.com/johng/gf/g/os/glog"
+    "gitee.com/johng/gf/g/os/gtime"
+)
+
+func main() {
+    // Server
+    go gtcp.NewServer("127.0.0.1:8999", func(conn net.Conn) {
+        c := gtcp.NewConnByNetConn(conn)
+        defer c.Close()
+        for {
+            if data, _ := c.Receive(-1); len(data) > 0 {
+                if err := c.Send(append([]byte("> "), data...)); err != nil {
+                  fmt.Println(err)
+                }
+            }
+            //return
+        }
+    }).Run()
+
+    time.Sleep(time.Second)
+
+    // Client
+    for {
+       if conn, err := gtcp.NewConn("127.0.0.1:8999"); err == nil {
+           if b, err := conn.SendReceive([]byte(gtime.Datetime()), -1); err == nil {
+               fmt.Println(string(b), conn.LocalAddr(), conn.RemoteAddr())
+           } else {
+               fmt.Println(err)
+           }
+           conn.Close()
+       } else {
+           glog.Error(err)
+       }
+       time.Sleep(time.Second)
+    }
+}
--- a/geg/net/gtcp/tcp_server_client.go
+++ b/geg/net/gtcp/tcp_server_client.go
@ -1,60 +1,76 @@
-package main
+    package main

-import (
-    "fmt"
-    "time"
-    "net"
-)
+    import (
+        "fmt"
+        "time"
+        "net"
+    )

-func main() {
-    addr := "127.0.0.1:8999"
+    func main() {
+        addr := "127.0.0.1:8999"

-    // Server
-    go func() {
-        tcpaddr, err := net.ResolveTCPAddr("tcp4", addr)
-        if err != nil {
-            panic(err)
-        }
-        listen, err := net.ListenTCP("tcp", tcpaddr)
-        if err != nil {
-            panic(err)
-        }
-        for  {
-            if conn, err := listen.Accept(); err != nil {
+        // Server
+        go func() {
+            tcpaddr, err := net.ResolveTCPAddr("tcp4", addr)
+            if err != nil {
                panic(err)
-            } else if conn != nil {
-                go func(conn net.Conn) {
-                    buffer := make([]byte, 1024)
-                    n, err := conn.Read(buffer)
-                    if err != nil {
-                        fmt.Println(err)
-                    } else {
-                        fmt.Println(">", string(buffer[0 : n]))
-                    }
-                    conn.Close()
-                }(conn)
            }
-        }
-    }()
+            listen, err := net.ListenTCP("tcp", tcpaddr)
+            if err != nil {
+                panic(err)
+            }
+            for  {
+                if conn, err := listen.Accept(); err != nil {
+                    panic(err)
+                } else if conn != nil {
+                    go func(conn net.Conn) {
+                        for {
+                            buffer := make([]byte, 1024)
+                            n, err := conn.Read(buffer)
+                            if err != nil {
+                                fmt.Println(err)
+                                break
+                            } else {
+                                fmt.Println(">", string(buffer[0 : n]))
+                                conn.Close()
+                            }
+                        }

-    time.Sleep(time.Second)
+                    }(conn)
+                }
+            }
+        }()

-    // Client
-    if conn, err := net.Dial("tcp", addr); err == nil {
-        for i := 0; i < 2; i++ {
-            _, err := conn.Write([]byte("hello"))
+        time.Sleep(time.Second)
+
+        // Client
+        if conn, err := net.Dial("tcp", addr); err == nil {
+            // first write
+            _, err := conn.Write([]byte("hello1"))
            if err != nil {
                fmt.Println(err)
                conn.Close()
-                break
+                return
+            } else {
+                fmt.Println("ok")
+            }
+
+            // sleep 10 seconds and re-send
+            time.Sleep(10*time.Second)
+
+            // second write
+            _, err = conn.Write([]byte("hello2"))
+            if err != nil {
+                fmt.Println(err)
+                conn.Close()
+                return
            } else {
                fmt.Println("ok")
            }
            // sleep 10 seconds and re-send
            time.Sleep(10*time.Second)
+        } else {
+            panic(err)
        }
-    } else {
-        panic(err)
-    }

-}
+    }
--- a/geg/net/gudp/gudp_server_client.go
+++ b/geg/net/gudp/gudp_server_client.go
@ -0,0 +1,40 @@
+package main
+
+import (
+    "fmt"
+    "time"
+    "gitee.com/johng/gf/g/os/glog"
+    "gitee.com/johng/gf/g/os/gtime"
+    "gitee.com/johng/gf/g/net/gudp"
+)
+
+func main() {
+    // Server
+    go gudp.NewServer("127.0.0.1:8999", func(conn *gudp.Conn) {
+        defer conn.Close()
+        for {
+            if data, _ := conn.Receive(-1); len(data) > 0 {
+                if err := conn.Send(append([]byte("> "), data...)); err != nil {
+                    glog.Error(err)
+                }
+            }
+        }
+    }).Run()
+
+    time.Sleep(time.Second)
+
+    // Client
+    for {
+        if conn, err := gudp.NewConn("127.0.0.1:8999"); err == nil {
+            if b, err := conn.SendReceive([]byte(gtime.Datetime()), -1); err == nil {
+                fmt.Println(string(b), conn.LocalAddr(), conn.RemoteAddr())
+            } else {
+                glog.Error(err)
+            }
+            conn.Close()
+        } else {
+            glog.Error(err)
+        }
+        time.Sleep(time.Second)
+    }
+}
--- a/geg/net/gudp/udp_client.go
+++ b/geg/net/gudp/udp_client.go
@ -1,11 +1,9 @@
 package main
-
 import (
-    "net"
    "fmt"
+    "net"
    "os"
 )
-
 func main() {
    conn, err := net.Dial("udp", "127.0.0.1:8999")
    defer conn.Close()
@ -13,9 +11,11 @@ func main() {
        os.Exit(1)
    }

-    conn.Write([]byte(""))
-    var msg [20]byte
-    n, err := conn.Read(msg[0:])
+    conn.Write([]byte("Hello world!"))

-    fmt.Println(string(msg[0:n]))
+    buffer := make([]byte, 100)
+
+    conn.Read(buffer)
+
+    fmt.Println(string(buffer))
 }
--- a/geg/net/gudp/udp_server.go
+++ b/geg/net/gudp/udp_server.go
@ -0,0 +1,29 @@
+package main
+import (
+    "fmt"
+    "net"
+)
+func main() {
+    listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 8999})
+    if err != nil {
+        fmt.Println(err)
+        return
+    }
+    fmt.Println("Local:", listener.LocalAddr().String())
+
+    data   := make([]byte, 1024)
+    for {
+        n, remoteAddr, err := listener.ReadFromUDP(data)
+        if err != nil {
+            fmt.Println(err)
+        }
+        fmt.Println(remoteAddr, string(data[ : n]))
+
+
+
+        _, err = listener.WriteToUDP([]byte("world"), remoteAddr)
+        if err != nil {
+            fmt.Printf(err.Error())
+        }
+    }
+}
--- a/geg/net/udp_server.go
+++ b/geg/net/udp_server.go
@ -1,16 +0,0 @@
-package main
-
-import (
-    "fmt"
-    "net"
-    "gitee.com/johng/gf/g/net/gudp"
-)
-
-func main() {
-    gudp.NewServer(":8999", func(conn *net.UDPConn) {
-        buffer := make([]byte, 1024)
-        if length, addr, err := conn.ReadFromUDP(buffer); err == nil {
-            fmt.Println(string(buffer[0 : length]), "from", addr.String())
-        }
-    }).Run()
-}
--- a/geg/other/test.go
+++ b/geg/other/test.go
@ -2,13 +2,73 @@ package main

 import (
    "fmt"
+    "log"
+    "net"
    "time"
 )

 func main() {
-    //t, err := gtime.StrToTime("2018-01-02 13:08:01")
-    //fmt.Println(err)
-    //fmt.Println(t.Format("3:4"))
-    //fmt.Println(time.Now().Format(".000"))
-    fmt.Println(time.Now().Format(".000"))
-}
+    addr := "127.0.0.1:8999"
+
+    tcpaddr, err := net.ResolveTCPAddr("tcp4", addr)
+    if err != nil {
+        log.Fatal(err)
+    }
+    listener, err := net.ListenTCP("tcp", tcpaddr)
+    if err != nil {
+        log.Fatal(err)
+    }
+
+    // Server
+    done := make(chan error)
+    go func(listener net.Listener, done chan<- error) {
+        for {
+            conn, err := listener.Accept()
+            if err != nil {
+                done <- err
+                return
+            }
+            go func(conn net.Conn) {
+                var buffer [1024]byte
+                n, err := conn.Read(buffer[:])
+                if err != nil {
+                    log.Println(err)
+                } else {
+                    log.Println(">", string(buffer[0:n]))
+                }
+                if err := conn.Close(); err != nil {
+                    log.Println("error closing server conn:", err)
+                }
+            }(conn)
+        }
+    }(listener, done)
+
+    // Client
+    conn, err := net.Dial("tcp", addr)
+    if err != nil {
+        log.Fatal(err)
+    }
+    for i := 0; i < 2; i++ {
+        _, err := conn.Write([]byte("hello"))
+        if err != nil {
+            log.Println(err)
+            err = conn.Close()
+            if err != nil {
+                log.Println("error closing client conn:", err)
+            }
+            break
+        }
+        fmt.Println("ok")
+        time.Sleep(2 * time.Second)
+    }
+
+    // Shut the server down and wait for it to report back
+    err = listener.Close()
+    if err != nil {
+        log.Fatal("error closing listener:", err)
+    }
+    err = <-done
+    if err != nil {
+        log.Println("server returned:", err)
+    }
+}
--- a/geg/util/gutil/map_to_struct.go
+++ b/geg/util/gutil/map_to_struct.go
@ -0,0 +1,24 @@
+package main
+
+import (
+    "gitee.com/johng/gf/g/util/gutil"
+    "fmt"
+)
+
+type User struct {
+    Name string
+    Age  int
+    Adds string
+}
+
+func main() {
+    m := map[string]interface{} {
+        "name" : "john",
+        "age"  : 16,
+        "adds" : "test",
+    }
+    o := User{}
+    e := gutil.MapToStruct(m, &o)
+    fmt.Println(e)
+    fmt.Println(o)
+}