feat(websocket): add event helper package

Author: ReneWerner87

v3/socketio/go.mod

@@ -4,7 +4,7 @@ go 1.25.0
 
 require (
 	github.com/fasthttp/websocket v1.5.12
-	github.com/gofiber/contrib/v3/websocket v1.0.0
+	github.com/gofiber/contrib/v3/websocket v1.1.4
 	github.com/gofiber/fiber/v3 v3.2.0
 	github.com/google/uuid v1.6.0
 	github.com/stretchr/testify v1.11.1

v3/socketio/legacy/README.md

@@ -31,3 +31,5 @@ legacy.On(legacy.EventMessage, func(ep *legacy.EventPayload) {
 
 app.Get("/ws", legacy.New(func(kws *legacy.Websocket) {}))
 ```
+
+Set tuning globals such as `PongTimeout`, `SendQueueSize`, or `MaxSendRetry` on `github.com/gofiber/contrib/v3/websocket/event` directly before accepting connections.

v3/socketio/legacy/legacy.go

@@ -5,8 +5,6 @@
 package legacy
 
 import (
-	"time"
-
 	"github.com/gofiber/contrib/v3/websocket"
 	wsevent "github.com/gofiber/contrib/v3/websocket/event"
 	"github.com/gofiber/fiber/v3"
@@ -41,18 +39,10 @@ var (
 	ErrorUUIDDuplication   = wsevent.ErrorUUIDDuplication
 )
 
-var (
-	PongTimeout      = wsevent.PongTimeout
-	RetrySendTimeout = wsevent.RetrySendTimeout
-	MaxSendRetry     = wsevent.MaxSendRetry
-	ReadTimeout      = wsevent.ReadTimeout
-)
-
 // New returns the legacy plain WebSocket event handler.
 //
 // Deprecated: use github.com/gofiber/contrib/v3/websocket/event.New directly.
 func New(callback func(kws *Websocket), config ...websocket.Config) fiber.Handler {
-	syncConfig()
 	return wsevent.New(callback, config...)
 }
 
@@ -90,17 +80,3 @@ func Broadcast(message []byte, mType ...int) {
 func Fire(event string, data []byte) {
 	wsevent.Fire(event, data)
 }
-
-func syncConfig() {
-	wsevent.PongTimeout = durationOrDefault(PongTimeout, wsevent.PongTimeout)
-	wsevent.RetrySendTimeout = durationOrDefault(RetrySendTimeout, wsevent.RetrySendTimeout)
-	wsevent.MaxSendRetry = MaxSendRetry
-	wsevent.ReadTimeout = durationOrDefault(ReadTimeout, wsevent.ReadTimeout)
-}
-
-func durationOrDefault(value, fallback time.Duration) time.Duration {
-	if value == 0 {
-		return fallback
-	}
-	return value
-}

v3/socketio/legacy/legacy_test.go

@@ -1,6 +1,7 @@
 package legacy
 
 import (
+	"errors"
 	"net"
 	"testing"
 	"time"
@@ -47,7 +48,7 @@ func TestLegacyPlainWebSocketEventShim(t *testing.T) {
 		},
 		HandshakeTimeout: 5 * time.Second,
 	}
-	conn, _, err := dialer.Dial("ws://"+ln.Addr().String(), nil)
+	conn, err := dialWebSocket(t, dialer, "ws://"+ln.Addr().String())
 	require.NoError(t, err)
 	defer func() { _ = conn.Close() }()
 
@@ -58,3 +59,21 @@ func TestLegacyPlainWebSocketEventShim(t *testing.T) {
 	require.Equal(t, TextMessage, messageType)
 	require.Equal(t, "legacy:ping", string(message))
 }
+
+func dialWebSocket(t *testing.T, dialer *websocket.Dialer, url string) (*websocket.Conn, error) {
+	t.Helper()
+
+	var lastErr error
+	for range 20 {
+		conn, _, err := dialer.Dial(url, nil)
+		if err == nil {
+			return conn, nil
+		}
+		lastErr = err
+		time.Sleep(10 * time.Millisecond)
+	}
+	if lastErr == nil {
+		lastErr = errors.New("websocket dial did not run")
+	}
+	return nil, lastErr
+}

v3/websocket/event/README.md

@@ -35,6 +35,18 @@ func Broadcast(message []byte, mType ...int)
 func Fire(name string, data []byte)
 ```
 
+## Configuration
+
+Set these package-level variables before accepting connections.
+
+| Variable           | Default | Description |
+|:-------------------|:--------|:------------|
+| `PongTimeout`      | `1s`    | Interval for server-sent WebSocket pong frames. |
+| `RetrySendTimeout` | `20ms`  | Backoff between retries while the connection is not ready. |
+| `MaxSendRetry`     | `5`     | Max retries for transient socket write readiness issues. |
+| `SendQueueSize`    | `100`   | Per-connection outbound message queue capacity. |
+| `ReadTimeout`      | `10ms`  | Deprecated; reads now block until a frame arrives or the connection closes. |
+
 ## Example
 
 ```go

v3/websocket/event/event.go

@@ -64,7 +64,10 @@ var (
 	RetrySendTimeout = 20 * time.Millisecond
 	// MaxSendRetry defines the max retries for transient socket write issues.
 	MaxSendRetry = 5
-	// ReadTimeout controls the pause between read attempts.
+	// SendQueueSize controls the per-connection outbound message queue size.
+	SendQueueSize = 100
+	// ReadTimeout is deprecated and no longer used; reads block until a
+	// message arrives or the connection is closed.
 	ReadTimeout = 10 * time.Millisecond
 )
 
@@ -214,9 +217,7 @@ func (l *safeListeners) get(event string) []eventCallback {
 		return make([]eventCallback, 0)
 	}
 
-	ret := make([]eventCallback, 0, len(l.list[event]))
-	ret = append(ret, l.list[event]...)
-	return ret
+	return append([]eventCallback(nil), l.list[event]...)
 }
 
 var listeners = safeListeners{
@@ -241,7 +242,7 @@ func New(callback func(kws *Websocket), config ...websocket.Config) fiber.Handle
 			Cookies: func(key string, defaultValue ...string) string {
 				return c.Cookies(key, defaultValue...)
 			},
-			queue:      make(chan message, 100),
+			queue:      make(chan message, sendQueueSize()),
 			done:       make(chan struct{}, 1),
 			attributes: make(map[string]interface{}),
 			isAlive:    true,
@@ -422,6 +423,7 @@ func (kws *Websocket) Emit(message []byte, mType ...int) {
 func (kws *Websocket) Close() {
 	kws.write(CloseMessage, []byte("Connection closed"))
 	kws.fireEvent(EventClose, nil, nil)
+	kws.disconnected(nil)
 }
 
 // IsAlive reports whether the connection is active.
@@ -475,15 +477,25 @@ func (kws *Websocket) send(ctx context.Context) {
 		case msg := <-kws.queue:
 			if !kws.hasConn() {
 				if msg.retries <= MaxSendRetry {
-					go func(msg message) {
-						time.Sleep(RetrySendTimeout)
-						msg.retries++
-						select {
-						case kws.queue <- msg:
-						case <-ctx.Done():
-						case <-kws.done:
-						}
-					}(msg)
+					retryTimer := time.NewTimer(RetrySendTimeout)
+					select {
+					case <-retryTimer.C:
+					case <-ctx.Done():
+						stopTimer(retryTimer)
+						return
+					case <-kws.done:
+						stopTimer(retryTimer)
+						return
+					}
+
+					msg.retries++
+					select {
+					case kws.queue <- msg:
+					case <-ctx.Done():
+						return
+					case <-kws.done:
+						return
+					}
 				}
 				continue
 			}
@@ -522,42 +534,35 @@ func (kws *Websocket) run() {
 	wg.Wait()
 }
 
-func (kws *Websocket) read(ctx context.Context) {
-	timeoutTicker := time.NewTicker(ReadTimeout)
-	defer timeoutTicker.Stop()
+func (kws *Websocket) read(_ context.Context) {
 	for {
-		select {
-		case <-timeoutTicker.C:
-			if !kws.hasConn() {
-				continue
-			}
-
-			mType, msg, err := kws.Conn.ReadMessage()
+		if !kws.hasConn() {
+			return
+		}
 
-			if mType == PingMessage {
-				kws.fireEvent(EventPing, nil, nil)
-				continue
-			}
+		mType, msg, err := kws.Conn.ReadMessage()
 
-			if mType == PongMessage {
-				kws.fireEvent(EventPong, nil, nil)
-				continue
-			}
+		if mType == PingMessage {
+			kws.fireEvent(EventPing, nil, nil)
+			continue
+		}
 
-			if mType == CloseMessage {
-				kws.disconnected(nil)
-				return
-			}
+		if mType == PongMessage {
+			kws.fireEvent(EventPong, nil, nil)
+			continue
+		}
 
-			if err != nil {
-				kws.disconnected(err)
-				return
-			}
+		if mType == CloseMessage {
+			kws.disconnected(nil)
+			return
+		}
 
-			kws.fireEvent(EventMessage, msg, nil)
-		case <-ctx.Done():
+		if err != nil {
+			kws.disconnected(err)
 			return
 		}
+
+		kws.fireEvent(EventMessage, msg, nil)
 	}
 }
 
@@ -597,6 +602,22 @@ func (kws *Websocket) randomUUID() string {
 	return uuid.New().String()
 }
 
+func sendQueueSize() int {
+	if SendQueueSize <= 0 {
+		return 1
+	}
+	return SendQueueSize
+}
+
+func stopTimer(timer *time.Timer) {
+	if !timer.Stop() {
+		select {
+		case <-timer.C:
+		default:
+		}
+	}
+}
+
 func fireGlobalEvent(event string, data []byte, err error) {
 	for _, kws := range pool.all() {
 		kws.fireEvent(event, data, err)

v3/websocket/event/event_test.go

@@ -306,6 +306,19 @@ func TestWebsocketSetUUIDUpdatesPool(t *testing.T) {
 	require.Equal(t, kws, poolEntry)
 }
 
+func TestWebsocketCloseRemovesConnectionFromPool(t *testing.T) {
+	resetState()
+
+	kws := createWS()
+	pool.set(kws)
+
+	kws.Close()
+
+	require.False(t, kws.IsAlive())
+	_, err := pool.get(kws.GetUUID())
+	require.ErrorIs(t, err, ErrorInvalidConnection)
+}
+
 func createWS() *Websocket {
 	kws := &Websocket{
 		Conn: nil,
@@ -321,7 +334,7 @@ func createWS() *Websocket {
 		Cookies: func(key string, defaultValue ...string) string {
 			return ""
 		},
-		queue:      make(chan message),
+		queue:      make(chan message, 1),
 		done:       make(chan struct{}, 1),
 		attributes: make(map[string]interface{}),
 		isAlive:    true,