You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

282 lines
6.8 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
  1. package znet
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "net"
  8. "sync"
  9. "github.com/aceld/zinx/utils"
  10. "github.com/aceld/zinx/ziface"
  11. )
  12. //Connection 链接
  13. type Connection struct {
  14. //当前Conn属于哪个Server
  15. TCPServer ziface.IServer
  16. //当前连接的socket TCP套接字
  17. Conn *net.TCPConn
  18. //当前连接的ID 也可以称作为SessionID,ID全局唯一
  19. ConnID uint32
  20. //消息管理MsgID和对应处理方法的消息管理模块
  21. MsgHandler ziface.IMsgHandle
  22. //告知该链接已经退出/停止的channel
  23. ctx context.Context
  24. cancel context.CancelFunc
  25. //无缓冲管道,用于读、写两个goroutine之间的消息通信
  26. msgChan chan []byte
  27. //有缓冲管道,用于读、写两个goroutine之间的消息通信
  28. msgBuffChan chan []byte
  29. sync.RWMutex
  30. //链接属性
  31. property map[string]interface{}
  32. ////保护当前property的锁
  33. propertyLock sync.Mutex
  34. //当前连接的关闭状态
  35. isClosed bool
  36. }
  37. //NewConnection 创建连接的方法
  38. func NewConnection(server ziface.IServer, conn *net.TCPConn, connID uint32, msgHandler ziface.IMsgHandle) *Connection {
  39. //初始化Conn属性
  40. c := &Connection{
  41. TCPServer: server,
  42. Conn: conn,
  43. ConnID: connID,
  44. isClosed: false,
  45. MsgHandler: msgHandler,
  46. msgChan: make(chan []byte),
  47. msgBuffChan: make(chan []byte, utils.GlobalObject.MaxMsgChanLen),
  48. property: nil,
  49. }
  50. //将新创建的Conn添加到链接管理中
  51. c.TCPServer.GetConnMgr().Add(c)
  52. return c
  53. }
  54. //StartWriter 写消息Goroutine, 用户将数据发送给客户端
  55. func (c *Connection) StartWriter() {
  56. fmt.Println("[Writer Goroutine is running]")
  57. defer fmt.Println(c.RemoteAddr().String(), "[conn Writer exit!]")
  58. for {
  59. select {
  60. case data := <-c.msgChan:
  61. //有数据要写给客户端
  62. if _, err := c.Conn.Write(data); err != nil {
  63. fmt.Println("Send Data error:, ", err, " Conn Writer exit")
  64. return
  65. }
  66. //fmt.Printf("Send data succ! data = %+v\n", data)
  67. case data, ok := <-c.msgBuffChan:
  68. if ok {
  69. //有数据要写给客户端
  70. if _, err := c.Conn.Write(data); err != nil {
  71. fmt.Println("Send Buff Data error:, ", err, " Conn Writer exit")
  72. return
  73. }
  74. } else {
  75. fmt.Println("msgBuffChan is Closed")
  76. break
  77. }
  78. case <-c.ctx.Done():
  79. return
  80. }
  81. }
  82. }
  83. //StartReader 读消息Goroutine,用于从客户端中读取数据
  84. func (c *Connection) StartReader() {
  85. fmt.Println("[Reader Goroutine is running]")
  86. defer fmt.Println(c.RemoteAddr().String(), "[conn Reader exit!]")
  87. defer c.Stop()
  88. // 创建拆包解包的对象
  89. for {
  90. select {
  91. case <-c.ctx.Done():
  92. return
  93. default:
  94. //读取客户端的Msg head
  95. headData := make([]byte, c.TCPServer.Packet().GetHeadLen())
  96. if _, err := io.ReadFull(c.Conn, headData); err != nil {
  97. fmt.Println("read msg head error ", err)
  98. return
  99. }
  100. //fmt.Printf("read headData %+v\n", headData)
  101. //拆包,得到msgID 和 datalen 放在msg中
  102. msg, err := c.TCPServer.Packet().Unpack(headData)
  103. if err != nil {
  104. fmt.Println("unpack error ", err)
  105. return
  106. }
  107. //根据 dataLen 读取 data,放在msg.Data中
  108. var data []byte
  109. if msg.GetDataLen() > 0 {
  110. data = make([]byte, msg.GetDataLen())
  111. if _, err := io.ReadFull(c.Conn, data); err != nil {
  112. fmt.Println("read msg data error ", err)
  113. return
  114. }
  115. }
  116. msg.SetData(data)
  117. //得到当前客户端请求的Request数据
  118. req := Request{
  119. conn: c,
  120. msg: msg,
  121. }
  122. if utils.GlobalObject.WorkerPoolSize > 0 {
  123. //已经启动工作池机制,将消息交给Worker处理
  124. c.MsgHandler.SendMsgToTaskQueue(&req)
  125. } else {
  126. //从绑定好的消息和对应的处理方法中执行对应的Handle方法
  127. go c.MsgHandler.DoMsgHandler(&req)
  128. }
  129. }
  130. }
  131. }
  132. //Start 启动连接,让当前连接开始工作
  133. func (c *Connection) Start() {
  134. c.ctx, c.cancel = context.WithCancel(context.Background())
  135. //1 开启用户从客户端读取数据流程的Goroutine
  136. go c.StartReader()
  137. //2 开启用于写回客户端数据流程的Goroutine
  138. go c.StartWriter()
  139. //按照用户传递进来的创建连接时需要处理的业务,执行钩子方法
  140. c.TCPServer.CallOnConnStart(c)
  141. }
  142. //Stop 停止连接,结束当前连接状态M
  143. func (c *Connection) Stop() {
  144. c.Lock()
  145. defer c.Unlock()
  146. //如果用户注册了该链接的关闭回调业务,那么在此刻应该显示调用
  147. c.TCPServer.CallOnConnStop(c)
  148. //如果当前链接已经关闭
  149. if c.isClosed == true {
  150. return
  151. }
  152. fmt.Println("Conn Stop()...ConnID = ", c.ConnID)
  153. // 关闭socket链接
  154. c.Conn.Close()
  155. //关闭Writer
  156. c.cancel()
  157. //将链接从连接管理器中删除
  158. c.TCPServer.GetConnMgr().Remove(c)
  159. //关闭该链接全部管道
  160. close(c.msgBuffChan)
  161. //设置标志位
  162. c.isClosed = true
  163. }
  164. //GetTCPConnection 从当前连接获取原始的socket TCPConn
  165. func (c *Connection) GetTCPConnection() *net.TCPConn {
  166. return c.Conn
  167. }
  168. //GetConnID 获取当前连接ID
  169. func (c *Connection) GetConnID() uint32 {
  170. return c.ConnID
  171. }
  172. //RemoteAddr 获取远程客户端地址信息
  173. func (c *Connection) RemoteAddr() net.Addr {
  174. return c.Conn.RemoteAddr()
  175. }
  176. //SendMsg 直接将Message数据发送数据给远程的TCP客户端
  177. func (c *Connection) SendMsg(msgID uint32, data []byte) error {
  178. c.RLock()
  179. defer c.RUnlock()
  180. if c.isClosed == true {
  181. return errors.New("connection closed when send msg")
  182. }
  183. //将data封包,并且发送
  184. dp := c.TCPServer.Packet()
  185. msg, err := dp.Pack(NewMsgPackage(msgID, data))
  186. if err != nil {
  187. fmt.Println("Pack error msg ID = ", msgID)
  188. return errors.New("Pack error msg ")
  189. }
  190. //写回客户端
  191. c.msgChan <- msg
  192. return nil
  193. }
  194. //SendBuffMsg 发生BuffMsg
  195. func (c *Connection) SendBuffMsg(msgID uint32, data []byte) error {
  196. c.RLock()
  197. defer c.RUnlock()
  198. if c.isClosed == true {
  199. return errors.New("Connection closed when send buff msg")
  200. }
  201. //将data封包,并且发送
  202. dp := c.TCPServer.Packet()
  203. msg, err := dp.Pack(NewMsgPackage(msgID, data))
  204. if err != nil {
  205. fmt.Println("Pack error msg ID = ", msgID)
  206. return errors.New("Pack error msg ")
  207. }
  208. //写回客户端
  209. c.msgBuffChan <- msg
  210. return nil
  211. }
  212. //SetProperty 设置链接属性
  213. func (c *Connection) SetProperty(key string, value interface{}) {
  214. c.propertyLock.Lock()
  215. defer c.propertyLock.Unlock()
  216. if c.property == nil {
  217. c.property = make(map[string]interface{})
  218. }
  219. c.property[key] = value
  220. }
  221. //GetProperty 获取链接属性
  222. func (c *Connection) GetProperty(key string) (interface{}, error) {
  223. c.propertyLock.Lock()
  224. defer c.propertyLock.Unlock()
  225. if value, ok := c.property[key]; ok {
  226. return value, nil
  227. }
  228. return nil, errors.New("no property found")
  229. }
  230. //RemoveProperty 移除链接属性
  231. func (c *Connection) RemoveProperty(key string) {
  232. c.propertyLock.Lock()
  233. defer c.propertyLock.Unlock()
  234. delete(c.property, key)
  235. }
  236. //返回ctx,用于用户自定义的go程获取连接退出状态
  237. func (c *Connection) Context() context.Context {
  238. return c.ctx
  239. }