diff --git a/core/chat/chatclient.go b/core/chat/chatclient.go index d838f8561..fdcd77e38 100644 --- a/core/chat/chatclient.go +++ b/core/chat/chatclient.go @@ -3,6 +3,7 @@ package chat import ( "bytes" "encoding/json" + "sync" "time" log "github.com/sirupsen/logrus" @@ -17,6 +18,7 @@ import ( // Client represents a single chat client. type Client struct { + mu sync.RWMutex id uint accessToken string conn *websocket.Conn @@ -152,11 +154,13 @@ func (c *Client) writePump() { // Optimization: Send multiple events in a single websocket message. // Add queued chat messages to the current websocket message. + c.mu.RLock() n := len(c.send) for i := 0; i < n; i++ { _, _ = w.Write(newline) _, _ = w.Write(<-c.send) } + c.mu.RUnlock() if err := w.Close(); err != nil { return @@ -177,9 +181,12 @@ func (c *Client) handleEvent(data []byte) { func (c *Client) close() { log.Traceln("client closed:", c.User.DisplayName, c.id, c.IPAddress) - _ = c.conn.Close() - c.server.unregister <- c.id + c.mu.Lock() + defer c.mu.Unlock() + if c.send != nil { + _ = c.conn.Close() + c.server.unregister <- c.id close(c.send) c.send = nil } @@ -214,6 +221,9 @@ func (c *Client) sendPayload(payload interface{}) { return } + c.mu.RLock() + defer c.mu.RUnlock() + if c.send != nil { c.send <- data } diff --git a/core/chat/server.go b/core/chat/server.go index 70020c7b9..648713014 100644 --- a/core/chat/server.go +++ b/core/chat/server.go @@ -231,8 +231,8 @@ func (s *Server) Broadcast(payload events.EventPayload) error { return err } - s.mu.Lock() - defer s.mu.Unlock() + s.mu.RLock() + defer s.mu.RUnlock() for _, client := range s.clients { if client == nil { @@ -242,8 +242,7 @@ func (s *Server) Broadcast(payload events.EventPayload) error { select { case client.send <- data: default: - client.close() - delete(s.clients, client.id) + go client.close() } }