ws_client.go 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107
  1. package handlers
  2. import (
  3. "encoding/json"
  4. "log/slog"
  5. "time"
  6. "github.com/gorilla/websocket"
  7. "golang.org/x/time/rate"
  8. )
  9. const (
  10. writeWait = 10 * time.Second
  11. pongWait = 60 * time.Second
  12. pingPeriod = (pongWait * 9) / 10
  13. maxMessageSize = 4096
  14. visitorUpdateLimit = 10
  15. visitorUpdateBurst = 1
  16. )
  17. type wsMessage struct {
  18. Type string `json:"type"`
  19. Data json.RawMessage `json:"data"`
  20. }
  21. type Client struct {
  22. hub *WSHub
  23. conn *websocket.Conn
  24. send chan []byte
  25. userID string
  26. lat float64
  27. lng float64
  28. limiter *rate.Limiter
  29. }
  30. func newClient(hub *WSHub, conn *websocket.Conn) *Client {
  31. return &Client{
  32. hub: hub,
  33. conn: conn,
  34. send: make(chan []byte, 256),
  35. limiter: rate.NewLimiter(visitorUpdateLimit, visitorUpdateBurst),
  36. }
  37. }
  38. func (c *Client) readPump() {
  39. defer func() {
  40. c.hub.unregister <- c
  41. c.conn.Close()
  42. }()
  43. c.conn.SetReadLimit(maxMessageSize)
  44. c.conn.SetReadDeadline(time.Now().Add(pongWait))
  45. c.conn.SetPongHandler(func(string) error {
  46. c.conn.SetReadDeadline(time.Now().Add(pongWait))
  47. return nil
  48. })
  49. for {
  50. var msg wsMessage
  51. if err := c.conn.ReadJSON(&msg); err != nil {
  52. break
  53. }
  54. switch msg.Type {
  55. case "visitor_update":
  56. if !c.limiter.Allow() {
  57. slog.Warn("ws visitor_update rate limit exceeded", "user_id", c.userID)
  58. continue
  59. }
  60. var dot VisitorDot
  61. if err := json.Unmarshal(msg.Data, &dot); err != nil {
  62. continue
  63. }
  64. c.userID = dot.UserID
  65. c.lat = dot.Lat
  66. c.lng = dot.Lng
  67. c.hub.broadcastVisitors()
  68. }
  69. }
  70. }
  71. func (c *Client) writePump() {
  72. ticker := time.NewTicker(pingPeriod)
  73. defer func() {
  74. ticker.Stop()
  75. c.conn.Close()
  76. }()
  77. for {
  78. select {
  79. case message, ok := <-c.send:
  80. c.conn.SetWriteDeadline(time.Now().Add(writeWait))
  81. if !ok {
  82. c.conn.WriteMessage(websocket.CloseMessage, []byte{})
  83. return
  84. }
  85. if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil {
  86. return
  87. }
  88. case <-ticker.C:
  89. c.conn.SetWriteDeadline(time.Now().Add(writeWait))
  90. if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
  91. return
  92. }
  93. }
  94. }
  95. }