diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 8de34cf..9a583e4 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -19,7 +19,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v5 with: - go-version: '1.25' + go-version: '1.27' cache: true - name: Run benchmarks diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index dd55582..9644550 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -34,7 +34,7 @@ jobs: strategy: matrix: os: [ubuntu-latest, macos-latest, windows-latest] - go-version: ['1.25'] # Match go.mod requirement + go-version: ['1.27'] # Match go.mod requirement env: GOEXPERIMENT: jsonv2 # Required for encoding/json/v2 @@ -63,7 +63,7 @@ jobs: run: go test -short -v -race -coverprofile=coverage.txt -covermode=atomic ./... - name: Upload coverage to Codecov - if: matrix.os == 'ubuntu-latest' && matrix.go-version == '1.25' + if: matrix.os == 'ubuntu-latest' && matrix.go-version == '1.27' uses: codecov/codecov-action@v5 with: token: ${{ secrets.CODECOV_TOKEN }} @@ -87,7 +87,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v5 with: - go-version: '1.25' + go-version: '1.27' cache: true - name: Run golangci-lint @@ -110,7 +110,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v5 with: - go-version: '1.25' + go-version: '1.27' cache: true - name: Check formatting diff --git a/README.md b/README.md index 39e580b..cf40e7f 100644 --- a/README.md +++ b/README.md @@ -157,7 +157,7 @@ go get github.com/coregx/stream - ✅ **Text & Binary** - Both message types supported - ✅ **Control Frames** - Ping/Pong, Close handshake - ✅ **Broadcasting Hub** - Efficient multi-client messaging -- ✅ **Connection Management** - Auto cleanup, timeouts +- ✅ **Connection Management** - Auto cleanup - ✅ **Frame Masking** - Client-to-server masking (RFC requirement) - ✅ **84.3% Test Coverage** - 99 tests, production-ready diff --git a/go.mod b/go.mod index d015adb..8449c39 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/coregx/stream -go 1.25 +go 1.27 diff --git a/sse/hub.go b/sse/hub.go index 81c00b1..15b0eb8 100644 --- a/sse/hub.go +++ b/sse/hub.go @@ -186,16 +186,17 @@ func (h *Hub[T]) removeClient(client *Conn) { // } // err = hub.Register(conn) func (h *Hub[T]) Register(conn *Conn) error { - h.mu.RLock() - closed := h.closed - h.mu.RUnlock() - - if closed { + select { + case <-h.done: + return ErrHubClosed + default: + } + select { + case h.register <- conn: + return nil + case <-h.done: return ErrHubClosed } - - h.register <- conn - return nil } // Unregister removes a connection from the hub. @@ -209,16 +210,17 @@ func (h *Hub[T]) Register(conn *Conn) error { // // err := hub.Unregister(conn) func (h *Hub[T]) Unregister(conn *Conn) error { - h.mu.RLock() - closed := h.closed - h.mu.RUnlock() - - if closed { + select { + case <-h.done: + return ErrHubClosed + default: + } + select { + case h.unregister <- conn: + return nil + case <-h.done: return ErrHubClosed } - - h.unregister <- conn - return nil } // Broadcast sends data to all connected clients. @@ -236,16 +238,17 @@ func (h *Hub[T]) Unregister(conn *Conn) error { // // err := hub.Broadcast("Server restarting in 5 minutes") func (h *Hub[T]) Broadcast(data T) error { - h.mu.RLock() - closed := h.closed - h.mu.RUnlock() - - if closed { + select { + case <-h.done: + return ErrHubClosed + default: + } + select { + case h.broadcast <- data: + return nil + case <-h.done: return ErrHubClosed } - - h.broadcast <- data - return nil } // BroadcastJSON sends a JSON-encoded value to all connected clients. diff --git a/websocket/hub.go b/websocket/hub.go index 3b2d2ac..2706efa 100644 --- a/websocket/hub.go +++ b/websocket/hub.go @@ -44,9 +44,10 @@ type Hub struct { broadcast chan []byte // Broadcast message to all // Lifecycle management - done chan struct{} // Shutdown signal - closed bool // Track if hub is closed - wg sync.WaitGroup // Wait for goroutines + done chan struct{} // Shutdown signal + closed bool // Track if hub is closed + started bool // Track if Run() was called + wg sync.WaitGroup // Wait for goroutines // Thread-safety for clients map and closed flag mu sync.RWMutex @@ -62,13 +63,15 @@ type Hub struct { // // Returns a ready-to-use Hub with initialized channels. func NewHub() *Hub { - return &Hub{ + h := &Hub{ clients: make(map[*Conn]bool), register: make(chan *Conn), unregister: make(chan *Conn), broadcast: make(chan []byte, 256), // Buffered for performance done: make(chan struct{}), } + h.wg.Add(1) // Must be before go hub.Run() to prevent race with Close(). + return h } // Run starts the Hub's event loop. @@ -83,35 +86,48 @@ func NewHub() *Hub { // - Graceful shutdown // // Run exits when Close() is called. +// Run must be called in a goroutine: go hub.Run(). +// Call hub.AddRunning() before starting the goroutine if using wg externally. func (h *Hub) Run() { - h.wg.Add(1) + h.mu.Lock() + if h.started || h.closed { + h.mu.Unlock() + return + } + h.started = true + h.mu.Unlock() + defer h.wg.Done() for { select { - case client := <-h.register: - // Register new client + case client, ok := <-h.register: + if !ok { + return + } h.mu.Lock() h.clients[client] = true h.mu.Unlock() - case client := <-h.unregister: - // Unregister client + case client, ok := <-h.unregister: + if !ok { + return + } h.mu.Lock() - if _, ok := h.clients[client]; ok { + if _, exists := h.clients[client]; exists { delete(h.clients, client) - _ = client.Close() // Close connection + _ = client.Close() } h.mu.Unlock() - case message := <-h.broadcast: - // Broadcast to all clients + case message, ok := <-h.broadcast: + if !ok { + return + } h.mu.RLock() for client := range h.clients { - // Send in goroutine to avoid blocking on slow clients go func(c *Conn, msg []byte) { if err := c.Write(BinaryMessage, msg); err != nil { - // Auto-unregister on write failure h.Unregister(c) } }(client, message) @@ -119,7 +135,6 @@ func (h *Hub) Run() { h.mu.RUnlock() case <-h.done: - // Shutdown return } } @@ -136,14 +151,10 @@ func (h *Hub) Run() { // // Thread-safe: can be called from multiple goroutines. func (h *Hub) Register(client *Conn) { - h.mu.RLock() - if h.closed { - h.mu.RUnlock() - return + select { + case h.register <- client: + case <-h.done: } - h.mu.RUnlock() - - h.register <- client } // Unregister removes a client from the Hub. @@ -157,14 +168,10 @@ func (h *Hub) Register(client *Conn) { // Thread-safe: can be called from multiple goroutines. // Safe to call multiple times for the same client (no-op after first call). func (h *Hub) Unregister(client *Conn) { - h.mu.RLock() - if h.closed { - h.mu.RUnlock() - return + select { + case h.unregister <- client: + case <-h.done: } - h.mu.RUnlock() - - h.unregister <- client } // Broadcast sends a message to all connected clients. @@ -181,14 +188,10 @@ func (h *Hub) Unregister(client *Conn) { // Thread-safe: can be called from multiple goroutines. // Non-blocking: queues message and returns immediately. func (h *Hub) Broadcast(message []byte) { - h.mu.RLock() - if h.closed { - h.mu.RUnlock() - return + select { + case h.broadcast <- message: + case <-h.done: } - h.mu.RUnlock() - - h.broadcast <- message } // BroadcastText sends a text message to all connected clients. @@ -259,13 +262,16 @@ func (h *Hub) Close() error { return nil } h.closed = true + started := h.started h.mu.Unlock() // Signal shutdown to event loop close(h.done) - // Wait for event loop to exit - h.wg.Wait() + // Wait for event loop to exit — only if Run() was called. + if started { + h.wg.Wait() + } // Close all client connections h.mu.Lock() @@ -275,10 +281,8 @@ func (h *Hub) Close() error { h.clients = make(map[*Conn]bool) // Clear map h.mu.Unlock() - // Close channels (safe now that event loop exited and no new sends) - close(h.register) - close(h.unregister) - close(h.broadcast) + // Channels are NOT closed — done signal + select handles shutdown. + // Closing channels would cause panic in concurrent senders. return nil }