package znet import ( "errors" "fmt" "io" "net" "sync" "zinx/utils" "zinx/ziface" ) type Connection struct { //当前Conn属于哪个Server TcpServer ziface.IServer //当前连接的socket TCP套接字 Conn *net.TCPConn //当前连接的ID 也可以称作为SessionID,ID全局唯一 ConnID uint32 //当前连接的关闭状态 isClosed bool //消息管理MsgId和对应处理方法的消息管理模块 MsgHandler ziface.IMsgHandle //告知该链接已经退出/停止的channel ExitBuffChan chan bool //无缓冲管道,用于读、写两个goroutine之间的消息通信 msgChan chan []byte //有关冲管道,用于读、写两个goroutine之间的消息通信 msgBuffChan chan []byte //链接属性 property map[string]interface{} //保护链接属性修改的锁 propertyLock sync.RWMutex } //创建连接的方法 func NewConntion(server ziface.IServer, conn *net.TCPConn, connID uint32, msgHandler ziface.IMsgHandle) *Connection { //初始化Conn属性 c := &Connection{ TcpServer: server, Conn: conn, ConnID: connID, isClosed: false, MsgHandler: msgHandler, ExitBuffChan: make(chan bool, 1), msgChan: make(chan []byte), msgBuffChan: make(chan []byte, utils.GlobalObject.MaxMsgChanLen), property: make(map[string]interface{}), } //将新创建的Conn添加到链接管理中 c.TcpServer.GetConnMgr().Add(c) return c } /* 写消息Goroutine, 用户将数据发送给客户端 */ func (c *Connection) StartWriter() { fmt.Println("[Writer Goroutine is running]") defer fmt.Println(c.RemoteAddr().String(), "[conn Writer exit!]") for { select { case data := <-c.msgChan: //有数据要写给客户端 if _, err := c.Conn.Write(data); err != nil { fmt.Println("Send Data error:, ", err, " Conn Writer exit") return } //fmt.Printf("Send data succ! data = %+v\n", data) case data, ok := <-c.msgBuffChan: if ok { //有数据要写给客户端 if _, err := c.Conn.Write(data); err != nil { fmt.Println("Send Buff Data error:, ", err, " Conn Writer exit") return } } else { break fmt.Println("msgBuffChan is Closed") } case <-c.ExitBuffChan: return } } } /* 读消息Goroutine,用于从客户端中读取数据 */ func (c *Connection) StartReader() { fmt.Println("[Reader Goroutine is running]") defer fmt.Println(c.RemoteAddr().String(), "[conn Reader exit!]") defer c.Stop() for { // 创建拆包解包的对象 dp := NewDataPack() //读取客户端的Msg head headData := make([]byte, dp.GetHeadLen()) if _, err := io.ReadFull(c.GetTCPConnection(), headData); err != nil { fmt.Println("read msg head error ", err) break } fmt.Printf("read headData %+v\n", headData) //拆包,得到msgid 和 datalen 放在msg中 msg, err := dp.Unpack(headData) if err != nil { fmt.Println("unpack error ", err) break } //根据 dataLen 读取 data,放在msg.Data中 var data []byte if msg.GetDataLen() > 0 { data = make([]byte, msg.GetDataLen()) if _, err := io.ReadFull(c.GetTCPConnection(), data); err != nil { fmt.Println("read msg data error ", err) break } } msg.SetData(data) //得到当前客户端请求的Request数据 req := Request{ conn: c, msg: msg, } if utils.GlobalObject.WorkerPoolSize > 0 { //已经启动工作池机制,将消息交给Worker处理 c.MsgHandler.SendMsgToTaskQueue(&req) } else { //从绑定好的消息和对应的处理方法中执行对应的Handle方法 go c.MsgHandler.DoMsgHandler(&req) } } } //启动连接,让当前连接开始工作 func (c *Connection) Start() { //1 开启用户从客户端读取数据流程的Goroutine go c.StartReader() //2 开启用于写回客户端数据流程的Goroutine go c.StartWriter() //按照用户传递进来的创建连接时需要处理的业务,执行钩子方法 c.TcpServer.CallOnConnStart(c) } //停止连接,结束当前连接状态M func (c *Connection) Stop() { fmt.Println("Conn Stop()...ConnID = ", c.ConnID) //如果当前链接已经关闭 if c.isClosed == true { return } c.isClosed = true //如果用户注册了该链接的关闭回调业务,那么在此刻应该显示调用 c.TcpServer.CallOnConnStop(c) // 关闭socket链接 c.Conn.Close() //关闭Writer c.ExitBuffChan <- true //将链接从连接管理器中删除 c.TcpServer.GetConnMgr().Remove(c) //关闭该链接全部管道 close(c.ExitBuffChan) close(c.msgBuffChan) } //从当前连接获取原始的socket TCPConn func (c *Connection) GetTCPConnection() *net.TCPConn { return c.Conn } //获取当前连接ID func (c *Connection) GetConnID() uint32 { return c.ConnID } //获取远程客户端地址信息 func (c *Connection) RemoteAddr() net.Addr { return c.Conn.RemoteAddr() } //直接将Message数据发送数据给远程的TCP客户端 func (c *Connection) SendMsg(msgId uint32, data []byte) error { if c.isClosed == true { return errors.New("Connection closed when send msg") } //将data封包,并且发送 dp := NewDataPack() msg, err := dp.Pack(NewMsgPackage(msgId, data)) if err != nil { fmt.Println("Pack error msg id = ", msgId) return errors.New("Pack error msg ") } //写回客户端 c.msgChan <- msg return nil } func (c *Connection) SendBuffMsg(msgId uint32, data []byte) error { if c.isClosed == true { return errors.New("Connection closed when send buff msg") } //将data封包,并且发送 dp := NewDataPack() msg, err := dp.Pack(NewMsgPackage(msgId, data)) if err != nil { fmt.Println("Pack error msg id = ", msgId) return errors.New("Pack error msg ") } //写回客户端 c.msgBuffChan <- msg return nil } //设置链接属性 func (c *Connection) SetProperty(key string, value interface{}) { c.propertyLock.Lock() defer c.propertyLock.Unlock() c.property[key] = value } //获取链接属性 func (c *Connection) GetProperty(key string) (interface{}, error) { c.propertyLock.RLock() defer c.propertyLock.RUnlock() if value, ok := c.property[key]; ok { return value, nil } else { return nil, errors.New("no property found") } } //移除链接属性 func (c *Connection) RemoveProperty(key string) { c.propertyLock.Lock() defer c.propertyLock.Unlock() delete(c.property, key) }