README
¶
go-wskit
WebSocket hub-and-spoke server on coder/websocket with optional Redis Pub/Sub for multi-instance broadcast.
Install
go get github.com/wahrwelt-kit/go-wskit
import "github.com/wahrwelt-kit/go-wskit"
Features
- Hub - single-goroutine dispatcher with register/unregister/broadcast channels, graceful shutdown via context, atomic subscriber count, and explicit operation errors
- Client - ReadPump (optional inbound message handler) + WritePump (send + ping), safe concurrent Send with close protection
- SSEClient - Server-Sent Events subscriber;
AcceptSSEregisters it with the hub and writes protocol-safedata:frames until disconnect or shutdown - Redis Pub/Sub - BroadcastEvent delivers locally first, publishes to Redis for other instances, and avoids duplicate local echo from the same hub
- Event envelope -
Event{Type, Payload, Timestamp}standard JSON format, BroadcastJSON helper - Accept / AcceptSSE helpers - upgrade HTTP to WebSocket or SSE + create client + register in one call
- Functional options - configurable timeouts, buffer sizes, connect/disconnect/drop/error callbacks
Example
hub := wskit.NewHub(
wskit.WithRedis(redisClient, "ws:events"),
wskit.WithOnDrop(func(sub wskit.Subscriber, data []byte) {
log.Printf("dropped websocket payload: %d bytes", len(data))
}),
wskit.WithOnError(func(op string, err error) {
log.Printf("websocket %s: %v", op, err)
}),
wskit.WithOnConnect(func(sub wskit.Subscriber) {
data, _ := json.Marshal(wskit.NewEvent("connected", nil))
sub.Send(data)
}),
)
go hub.Run(ctx)
go hub.SubscribeToRedis(ctx)
// WebSocket endpoint
http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
client, err := wskit.Accept(r.Context(), w, r, hub, nil)
if err != nil {
return
}
go client.ReadPump()
go client.WritePump()
})
// SSE endpoint
http.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
if err := wskit.AcceptSSE(w, r, hub); err != nil {
return
}
})
if err := hub.BroadcastJSON(ctx, "notification", map[string]string{"message": "hello"}); err != nil {
log.Printf("broadcast: %v", err)
}
Options
Hub: WithRedis, WithBroadcastBuf, WithRegisterBuf, WithChannelTimeout, WithOnTimeout, WithOnDrop, WithOnError, WithOnConnect, WithOnDisconnect
Client: WithWriteWait, WithPingInterval, WithMaxMessageSize, WithSendBufSize, WithMessageHandler
Documentation
¶
Overview ¶
Package wskit provides a WebSocket and SSE hub-and-spoke server built on coder/websocket
Hub and Subscriber ¶
Create a Hub with NewHub (optionally WithRedis for multi-instance broadcast), run Hub.Run(ctx) in a goroutine, then use Accept to upgrade HTTP connections and register WebSocket clients, or AcceptSSE for Server-Sent Events Any type implementing the Subscriber interface (Send + Close) can participate in hub broadcasts. Register, Unregister, and Broadcast return errors for nil input, stopped hubs, or channel operation timeouts
WebSocket Clients ¶
Use Accept to upgrade HTTP connections and register clients. Run Client.ReadPump and Client.WritePump in separate goroutines per connection. Use WithMessageHandler when inbound WebSocket messages should be handled instead of only drained
SSE Clients ¶
Use AcceptSSE to handle SSE connections. It registers an SSEClient with the hub, writes protocol-safe data frames, and blocks until the client disconnects or the hub shuts down
Event envelope ¶
Event and NewEvent provide a standard JSON envelope (type, payload, timestamp). Use Hub.BroadcastEvent or Hub.BroadcastJSON to send to all subscribers
Redis Pub/Sub ¶
WithRedis(client, channel) enables local-first broadcast plus Redis fanout on BroadcastEvent/BroadcastJSON. Other instances run SubscribeToRedis(ctx) to receive and broadcast locally. SubscribeToRedis automatically reconnects with exponential backoff and skips messages originating from the same hub
Options ¶
Hub: WithRedis, WithBroadcastBuf, WithRegisterBuf, WithChannelTimeout, WithOnTimeout, WithOnDrop, WithOnError, WithOnConnect, WithOnDisconnect Client: WithWriteWait, WithPingInterval, WithMaxMessageSize, WithSendBufSize, WithMessageHandler
Index ¶
- Constants
- Variables
- func AcceptSSE(w http.ResponseWriter, r *http.Request, hub *Hub) error
- type Client
- type ClientConfig
- type ClientOption
- type Event
- type Hub
- func (h *Hub) Broadcast(data []byte) error
- func (h *Hub) BroadcastEvent(ctx context.Context, event any) error
- func (h *Hub) BroadcastJSON(ctx context.Context, typ string, payload any) error
- func (h *Hub) Register(sub Subscriber) error
- func (h *Hub) Run(ctx context.Context)
- func (h *Hub) SubscribeToRedis(ctx context.Context)
- func (h *Hub) SubscriberCount() int
- func (h *Hub) Unregister(sub Subscriber) error
- type HubOption
- func WithBroadcastBuf(n int) HubOption
- func WithChannelTimeout(d time.Duration) HubOption
- func WithOnConnect(fn func(Subscriber)) HubOption
- func WithOnDisconnect(fn func(Subscriber)) HubOption
- func WithOnDrop(fn func(Subscriber, []byte)) HubOption
- func WithOnError(fn func(op string, err error)) HubOption
- func WithOnTimeout(fn func(op string)) HubOption
- func WithRedis(client *redis.Client, channel string) HubOption
- func WithRegisterBuf(n int) HubOption
- type MessageHandler
- type SSEClient
- type Subscriber
Constants ¶
const ( DefaultWriteWait = 10 * time.Second // Timeout for writing a single message or ping frame to the WebSocket connection DefaultPingInterval = 30 * time.Second // Interval between outgoing ping frames to keep the connection alive DefaultMaxMessageSize = 64 * 1024 // Maximum allowed size of a single incoming message in bytes DefaultSendBufSize = 256 // Default send channel buffer size (number of messages) for Client and SSEClient )
DefaultWriteWait is the timeout for writing a single message or ping frame to the WebSocket connection
const ( DefaultBroadcastBuf = 256 // Default broadcast channel buffer size (number of messages) DefaultRegisterBuf = 64 // Default register/unregister channel buffer size (number of operations) DefaultChannelTimeout = 5 * time.Second // Default timeout for Register, Unregister, and Broadcast channel operations )
DefaultBroadcastBuf is the default broadcast channel buffer size (number of messages)
Variables ¶
var ( // ErrHubStopped is returned when a hub operation is attempted after shutdown ErrHubStopped = errors.New("wskit: hub is stopped") // ErrFlusherNotSupported is returned by AcceptSSE when the ResponseWriter does not implement http.Flusher ErrFlusherNotSupported = errors.New("wskit: http.Flusher not supported") // ErrNilContext is returned when a required context is nil ErrNilContext = errors.New("wskit: context is nil") // ErrNilHub is returned when a required hub is nil ErrNilHub = errors.New("wskit: hub is nil") // ErrNilConn is returned when a required WebSocket connection is nil ErrNilConn = errors.New("wskit: websocket connection is nil") // ErrNilRequest is returned when a required HTTP request is nil ErrNilRequest = errors.New("wskit: http request is nil") // ErrNilResponseWriter is returned when a required HTTP response writer is nil ErrNilResponseWriter = errors.New("wskit: http response writer is nil") // ErrNilSubscriber is returned when a required subscriber is nil ErrNilSubscriber = errors.New("wskit: subscriber is nil") // ErrOperationTimeout is returned when a hub channel operation times out ErrOperationTimeout = errors.New("wskit: operation timeout") // ErrSubscriberClosed is returned when sending to a closed subscriber ErrSubscriberClosed = errors.New("wskit: subscriber is closed") // ErrSubscriberBufferFull is returned when a subscriber send buffer is full ErrSubscriberBufferFull = errors.New("wskit: subscriber buffer is full") // ErrRedisSubscriptionClosed is reported when a Redis subscription channel closes unexpectedly ErrRedisSubscriptionClosed = errors.New("wskit: redis subscription closed") )
Functions ¶
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client represents a single WebSocket connection attached to a Hub It implements the Subscriber interface
func Accept ¶
func Accept(ctx context.Context, w http.ResponseWriter, r *http.Request, hub *Hub, acceptOpts *websocket.AcceptOptions, clientOpts ...ClientOption) (*Client, error)
Accept upgrades the HTTP connection to WebSocket, creates a Client, and registers it with the hub. Caller should run ReadPump and WritePump in goroutines. acceptOpts may be nil for default upgrade options
func NewClient ¶
func NewClient(hub *Hub, conn *websocket.Conn, ctx context.Context, opts ...ClientOption) (*Client, error)
NewClient creates a client for the given hub and connection. Call Register on the hub, then run ReadPump and WritePump in separate goroutines
func (*Client) Close ¶ added in v0.3.0
func (c *Client) Close()
Close signals the client to shut down. It is idempotent and safe to call from any goroutine. The underlying WebSocket connection is closed by WritePump/ReadPump defers
func (*Client) ReadPump ¶
func (c *Client) ReadPump()
ReadPump reads messages from the connection until it closes or errors. On exit it unregisters the client and closes the connection. Run in a goroutine
func (*Client) Send ¶
Send enqueues data for writing. Non-blocking; returns false if the send buffer is full or the client has been closed
type ClientConfig ¶
type ClientConfig struct {
WriteWait time.Duration
PingInterval time.Duration
MaxMessageSize int64
SendBufSize int
OnMessage MessageHandler
}
ClientConfig holds configuration parameters for a Client or SSEClient
type ClientOption ¶
type ClientOption func(*ClientConfig)
ClientOption configures a Client
func WithMaxMessageSize ¶
func WithMaxMessageSize(n int64) ClientOption
WithMaxMessageSize sets the maximum size of a single incoming message
func WithMessageHandler ¶ added in v0.4.0
func WithMessageHandler(fn MessageHandler) ClientOption
WithMessageHandler sets the inbound WebSocket message handler used by ReadPump.
func WithPingInterval ¶
func WithPingInterval(d time.Duration) ClientOption
WithPingInterval sets the interval between ping frames
func WithSendBufSize ¶
func WithSendBufSize(n int) ClientOption
WithSendBufSize sets the send channel buffer size
func WithWriteWait ¶
func WithWriteWait(d time.Duration) ClientOption
WithWriteWait sets the timeout for writing a message or ping
type Event ¶
type Event struct {
Type string `json:"type"`
Payload any `json:"payload"`
Timestamp time.Time `json:"timestamp"`
}
Event is the default envelope for WebSocket messages: type, payload, and timestamp
type Hub ¶
type Hub struct {
// contains filtered or unexported fields
}
Hub is the central dispatcher for subscribers. Run one goroutine with Run(ctx)
func (*Hub) Broadcast ¶
Broadcast sends data to all connected subscribers. Non-blocking with timeout.
func (*Hub) BroadcastEvent ¶
BroadcastEvent marshals event as JSON, broadcasts it locally, and publishes it to Redis when configured.
func (*Hub) BroadcastJSON ¶
BroadcastJSON marshals Event{typ, payload, now} and broadcasts it via the hub. Returns an error if JSON marshaling fails
func (*Hub) Register ¶
func (h *Hub) Register(sub Subscriber) error
Register adds the subscriber to the hub. Non-blocking with timeout.
func (*Hub) SubscribeToRedis ¶
SubscribeToRedis subscribes to the hub's Redis channel and broadcasts received messages to all clients. It automatically reconnects with exponential backoff if the subscription is lost. Run in a goroutine; it returns when ctx is cancelled
func (*Hub) SubscriberCount ¶ added in v0.3.0
SubscriberCount returns the number of registered subscribers
func (*Hub) Unregister ¶
func (h *Hub) Unregister(sub Subscriber) error
Unregister removes the subscriber from the hub. Non-blocking with timeout.
type HubOption ¶
type HubOption func(*Hub)
HubOption configures a Hub
func WithBroadcastBuf ¶
WithBroadcastBuf sets the broadcast channel buffer size
func WithChannelTimeout ¶
WithChannelTimeout sets the timeout for Register, Unregister, and Broadcast operations
func WithOnConnect ¶
func WithOnConnect(fn func(Subscriber)) HubOption
WithOnConnect sets the callback invoked when a subscriber registers
func WithOnDisconnect ¶ added in v0.3.0
func WithOnDisconnect(fn func(Subscriber)) HubOption
WithOnDisconnect sets the callback invoked when a subscriber unregisters
func WithOnDrop ¶ added in v0.4.0
func WithOnDrop(fn func(Subscriber, []byte)) HubOption
WithOnDrop sets a callback invoked when a subscriber rejects a broadcast message
func WithOnError ¶ added in v0.4.0
WithOnError sets a callback invoked when Redis or hub operations fail asynchronously
func WithOnTimeout ¶
WithOnTimeout sets a callback when a channel operation times out (e.g. for logging)
func WithRedis ¶
WithRedis configures Redis Pub/Sub for multi-instance broadcast. If client is nil, Redis is disabled
func WithRegisterBuf ¶
WithRegisterBuf sets the register/unregister channel buffer size
type MessageHandler ¶ added in v0.4.0
type MessageHandler func(ctx context.Context, client *Client, messageType websocket.MessageType, data []byte) error
MessageHandler handles inbound WebSocket messages read by Client.ReadPump. Returning an error stops the client read loop and unregisters the client.
type SSEClient ¶ added in v0.3.0
type SSEClient struct {
// contains filtered or unexported fields
}
SSEClient represents a Server-Sent Events connection attached to a Hub It implements the Subscriber interface
func NewSSEClient ¶ added in v0.3.0
NewSSEClient creates an SSE subscriber for the given hub with the specified buffer size
func (*SSEClient) Close ¶ added in v0.3.0
func (c *SSEClient) Close()
Close signals the SSE client to shut down. It is idempotent and safe to call from any goroutine. AcceptSSE returns once the done channel is closed
type Subscriber ¶ added in v0.3.0
type Subscriber interface {
// Send enqueues data for delivery. Returns true if accepted, false if dropped
Send(data []byte) bool
// Close signals the subscriber to shut down. Must be idempotent
Close()
}
Subscriber is the interface that any client (WebSocket, SSE, etc.) must implement to participate in hub broadcasts
Source Files