client.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137
  1. package websocket
  2. import (
  3. "fmt"
  4. "github.com/gogf/gf/v2/container/garray"
  5. "github.com/gogf/gf/v2/util/guid"
  6. "github.com/gorilla/websocket"
  7. "runtime/debug"
  8. )
  9. const (
  10. // 用户连接超时时间
  11. heartbeatExpirationTime = 6 * 60
  12. )
  13. // 用户登录
  14. type login struct {
  15. UserId uint64
  16. Client *Client
  17. }
  18. // GetKey 读取客户端数据
  19. func (l *login) GetKey() (key string) {
  20. key = GetUserKey(l.UserId)
  21. return
  22. }
  23. // Client 客户端连接
  24. type Client struct {
  25. Addr string // 客户端地址
  26. ID string // 连接唯一标识
  27. Socket *websocket.Conn // 用户连接
  28. Send chan *WResponse // 待发送的数据
  29. SendClose bool // 发送是否关闭
  30. UserId uint64 // 用户ID,用户登录以后才有
  31. FirstTime uint64 // 首次连接事件
  32. HeartbeatTime uint64 // 用户上次心跳时间
  33. LoginTime uint64 // 登录时间 登录以后才有
  34. isApp bool // 是否是app
  35. tags garray.StrArray // 标签
  36. }
  37. // NewClient 初始化
  38. func NewClient(addr string, socket *websocket.Conn, firstTime uint64) (client *Client) {
  39. client = &Client{
  40. Addr: addr,
  41. ID: guid.S(),
  42. Socket: socket,
  43. Send: make(chan *WResponse, 100),
  44. SendClose: false,
  45. FirstTime: firstTime,
  46. HeartbeatTime: firstTime,
  47. }
  48. return
  49. }
  50. // 读取客户端数据
  51. func (c *Client) read() {
  52. defer func() {
  53. if r := recover(); r != nil {
  54. fmt.Println("write stop", string(debug.Stack()), r)
  55. }
  56. }()
  57. defer func() {
  58. c.close()
  59. }()
  60. for {
  61. _, message, err := c.Socket.ReadMessage()
  62. if err != nil {
  63. return
  64. }
  65. // 处理程序
  66. fmt.Println(message)
  67. ProcessData(c, message)
  68. }
  69. }
  70. // 向客户端写数据
  71. func (c *Client) write() {
  72. defer func() {
  73. if r := recover(); r != nil {
  74. fmt.Println("write stop", string(debug.Stack()), r)
  75. }
  76. }()
  77. defer func() {
  78. clientManager.Unregister <- c
  79. _ = c.Socket.Close()
  80. }()
  81. for {
  82. select {
  83. case message, ok := <-c.Send:
  84. if !ok {
  85. // 发送数据错误 关闭连接
  86. return
  87. }
  88. _ = c.Socket.WriteJSON(message)
  89. }
  90. }
  91. }
  92. // SendMsg 发送数据
  93. func (c *Client) SendMsg(msg *WResponse) {
  94. if c == nil || c.SendClose {
  95. return
  96. }
  97. defer func() {
  98. if r := recover(); r != nil {
  99. fmt.Println("SendMsg stop:", r, string(debug.Stack()))
  100. }
  101. }()
  102. c.Send <- msg
  103. }
  104. // Heartbeat 心跳更新
  105. func (c *Client) Heartbeat(currentTime uint64) {
  106. c.HeartbeatTime = currentTime
  107. return
  108. }
  109. // IsHeartbeatTimeout 心跳是否超时
  110. func (c *Client) IsHeartbeatTimeout(currentTime uint64) (timeout bool) {
  111. if c.HeartbeatTime+heartbeatExpirationTime <= currentTime {
  112. timeout = true
  113. }
  114. return
  115. }
  116. // 关闭客户端
  117. func (c *Client) close() {
  118. if c.SendClose {
  119. return
  120. }
  121. c.SendClose = true
  122. close(c.Send)
  123. }