package tcp2 import ( "crypto/rand" "encoding/binary" "errors" "fmt" "io" "log" "net" "strings" "time" "git.me9.top/git/tinymq/config" "git.me9.top/git/tinymq/conn" "git.me9.top/git/tinymq/util" "git.me9.top/git/tinymq/util/pool" ) const PROTO string = "tcp" const VERSION uint8 = 2 // 第一级数据包的最大长度,如果超过则自动增加一个地址位到4个字节 const MAX_LENGTH = 0xFFFF // 由于都是内部使用,感觉没有必要加这个限制 // const MAX2_LENGTH = 0x1FFFFFFF // 500 M,避免申请过大内存 type Tcp2 struct { cf *config.Config conn net.Conn cipher *util.Cipher // 记录当前的加解密类 compress bool // 是否启用压缩 } // 服务端 // hash 格式 encryptMethod:encryptKey func Server(cf *config.Config, bind string, hash string, fn conn.ServerConnectFunc) (err error) { var ci *util.CipherInfo var encryptKey string if hash != "" { i := strings.Index(hash, ":") if i <= 0 { err = errors.New("hash is invalid") return } encryptMethod := hash[0:i] encryptKey = hash[i+1:] if c, ok := util.CipherMethod[encryptMethod]; ok { ci = c } else { return errors.New("Unsupported encryption method: " + encryptMethod) } } log.Printf("Listening and serving tcp on %s\n", bind) l, err := net.Listen("tcp", bind) if err != nil { log.Println("[tcp2 Server ERROR]", err) return } go func(l net.Listener) { defer l.Close() for { conn, err := l.Accept() if err != nil { log.Println("[accept ERROR]", err) return } go func(conn net.Conn) { if ci == nil { c := &Tcp2{ cf: cf, conn: conn, } fn(c) return } var eiv []byte var div []byte if ci.IvLen > 0 { // 服务端 IV eiv = pool.Get(ci.IvLen) defer pool.Put(eiv) _, err = rand.Read(eiv) if err != nil { log.Println("[tcp2 Server rand.Read ERROR]", err) return } // 发送 IV conn.SetWriteDeadline(time.Now().Add(time.Duration(cf.WriteWait) * time.Millisecond)) if _, err := conn.Write(eiv); err != nil { log.Println("[tcp2 Server conn.Write ERROR]", err) return } // 读取 IV err = conn.SetReadDeadline(time.Now().Add(time.Millisecond * time.Duration(cf.ReadWait))) if err != nil { log.Println("[tcp2 Server SetReadDeadline ERROR]", err) return } div = pool.Get(ci.IvLen) defer pool.Put(div) _, err := io.ReadFull(conn, div) if err != nil { log.Println("[tcp2 Server ReadFull ERROR]", err) return } } cipher, err := util.NewCipher(ci, encryptKey, eiv, div) if err != nil { log.Println("[tcp2 NewCipher ERROR]", err) return } // 初始化 c := &Tcp2{ cf: cf, conn: conn, cipher: cipher, } fn(c) }(conn) } }(l) return } // 客户端,新建一个连接 func Dial(cf *config.Config, addr string, hash string) (conn.Connect, error) { // 没有加密的情况 if hash == "" { conn, err := net.DialTimeout("tcp", addr, time.Duration(cf.ConnectTimeout)*time.Millisecond) if err != nil { return nil, err } c := &Tcp2{ cf: cf, conn: conn, } return c, nil } i := strings.Index(hash, ":") if i <= 0 { return nil, errors.New("hash is invalid") } encryptMethod := hash[0:i] encryptKey := hash[i+1:] ci, ok := util.CipherMethod[encryptMethod] if !ok { return nil, errors.New("Unsupported encryption method: " + encryptMethod) } conn, err := net.DialTimeout("tcp", addr, time.Duration(cf.ConnectTimeout)*time.Millisecond) if err != nil { return nil, err } var eiv []byte var div []byte if ci.IvLen > 0 { // 客户端 IV eiv = pool.Get(ci.IvLen) defer pool.Put(eiv) _, err = rand.Read(eiv) if err != nil { log.Println("[tcp2 Client rand.Read ERROR]", err) return nil, err } // 发送 IV conn.SetWriteDeadline(time.Now().Add(time.Duration(cf.WriteWait) * time.Millisecond)) if _, err := conn.Write(eiv); err != nil { log.Println("[tcp2 Client conn.Write ERROR]", err) return nil, err } // 读取 IV err = conn.SetReadDeadline(time.Now().Add(time.Millisecond * time.Duration(cf.ReadWait))) if err != nil { log.Println("[tcp2 Client SetReadDeadline ERROR]", err) return nil, err } div = pool.Get(ci.IvLen) defer pool.Put(div) _, err := io.ReadFull(conn, div) if err != nil { log.Println("[tcp2 Client ReadFull ERROR]", err) return nil, err } } cipher, err := util.NewCipher(ci, encryptKey, eiv, div) if err != nil { log.Println("[tcp2 NewCipher ERROR]", err) return nil, err } // 初始化 c := &Tcp2{ cf: cf, conn: conn, cipher: cipher, compress: true, } return c, nil } // 发送数据包到网络 // recycle 指示是否需要将raw放回内存池 func (c *Tcp2) WriteRawPackage(raw []byte, recycle bool) (err error) { var buf []byte var index int dlen := len(raw) if dlen >= MAX_LENGTH { buf = pool.Get(dlen + 2 + 4 + 1) index = 2 + 4 binary.BigEndian.PutUint16(buf[:2], MAX_LENGTH) binary.BigEndian.PutUint32(buf[2:6], uint32(dlen)) } else { buf = pool.Get(dlen + 2 + 1) index = 2 binary.BigEndian.PutUint16(buf[:2], uint16(dlen)) } copy(buf[index:index+dlen], raw) index += dlen buf[index] = util.CRC8(raw) if recycle { pool.Put(raw) } defer pool.Put(buf) if c.cipher != nil { c.cipher.Encrypt(buf, buf) } c.conn.SetWriteDeadline(time.Now().Add(time.Duration(c.cf.WriteWait) * time.Millisecond)) for { n, err := c.conn.Write(buf) if err != nil { return err } if n < len(buf) { buf = buf[n:] } else { return nil } } } // 从连接中读取数据包,已经剥离了头部长度和尾部crc // recycle 指示是否需要手动将缓存放回内存池 func (c *Tcp2) ReadRawPackage(deadline int) (buf []byte, recycle bool, err error) { err = c.conn.SetReadDeadline(time.Now().Add(time.Millisecond * time.Duration(deadline))) if err != nil { return nil, false, err } buf = pool.Get(2) // 读取数据流长度 _, err = io.ReadFull(c.conn, buf) if err != nil { pool.Put(buf) return nil, false, err } // 将读出来的数据进行解密 if c.cipher != nil { c.cipher.Decrypt(buf, buf) } dlen := uint32(binary.BigEndian.Uint16(buf)) pool.Put(buf) if dlen < 2 { return nil, false, errors.New("length is less to 2") } if dlen >= MAX_LENGTH { // 数据包比较大,通过后面的4位长度来表示实际长度 buf := pool.Get(4) _, err := io.ReadFull(c.conn, buf) if err != nil { pool.Put(buf) return nil, false, err } if c.cipher != nil { c.cipher.Decrypt(buf, buf) } dlen = binary.BigEndian.Uint32(buf) pool.Put(buf) if dlen < MAX_LENGTH { return nil, false, errors.New("wrong length in read message") } } // 读取指定长度的数据 buf = pool.Get(int(dlen) + 1) // 最后一个是crc的值 _, err = io.ReadFull(c.conn, buf) if err != nil { pool.Put(buf) return nil, false, err } if c.cipher != nil { c.cipher.Decrypt(buf, buf) } // 检查CRC8 if util.CRC8(buf[:dlen]) != buf[dlen] { pool.Put(buf) return nil, false, errors.New("CRC error") } return buf[:dlen], true, nil } // 发送Auth信息 // 建立连接后第一个发送的消息 func (c *Tcp2) WriteAuthInfo(channel string, auth []byte) (err error) { buf := conn.AuthPackageEncode(PROTO, VERSION, channel, auth, c.compress) return c.WriteRawPackage(buf, true) } // 获取Auth信息 func (c *Tcp2) ReadAuthInfo() (proto string, version uint8, channel string, auth []byte, err error) { msg, recycle, err := c.ReadRawPackage(c.cf.ReadWait) if err != nil { return } var compress bool proto, version, compress, channel, auth, err = conn.AuthPackageDecode(msg, recycle) if err != nil { return } if proto != PROTO { err = fmt.Errorf("wrong proto: %s", proto) return } if version != VERSION { err = fmt.Errorf("require version %d, get version: %d", VERSION, version) return } c.compress = compress return } // 发送请求数据包到网络 func (c *Tcp2) WriteRequest(id uint16, cmd string, data []byte) error { buf := conn.RequestPackageEncode(id, cmd, data, c.compress, c.cf) return c.WriteRawPackage(buf, true) } // 发送响应数据包到网络 // 网络格式:[id, stateCode, data] func (c *Tcp2) WriteResponse(id uint16, state uint8, data []byte) error { buf := conn.ResponsePackageEncode(id, state, data, c.compress, c.cf) return c.WriteRawPackage(buf, true) } // 发送ping包 func (c *Tcp2) WritePing(id uint16) error { buf := conn.PingPackageEncode(id) return c.WriteRawPackage(buf, true) } // 获取信息 func (c *Tcp2) ReadMessage(deadline int) (msgType conn.MsgType, id uint16, cmd string, state uint8, data []byte, err error) { var msg []byte var recycle bool msg, recycle, err = c.ReadRawPackage(deadline) if err != nil { return } return conn.PackageDecode(msg, recycle, c.compress, c.cf) } // 获取远程的地址 func (c *Tcp2) RemoteIP() net.IP { return util.AddrToIP(c.conn.RemoteAddr()) } // 获取本地的地址 func (c *Tcp2) LocalIP() net.IP { return util.AddrToIP(c.conn.LocalAddr()) } func (c *Tcp2) Close() error { return c.conn.Close() }