SnapWS v1.1.3

Bug Fixes

• Fixed a “superfluous response WriteHeader call” caused when a MiddlewareErr was returned. Middleware errors now correctly return after writing the response.

Additions

• Added batching example

Internal Improvements

• Normal (non-batch) message broadcasting is optimized: a single frame with headers and payload is now queued to each connection. This avoids per-connection frame creation, payload copying, and potential fragmentation that previously increased syscalls.
• broadcastQueue channel draining after closing the connection. This might speed up the GC in some scenarios.

Message Batching System

Message batching first introduced in v1.1.0, but i added somethings, so i will include the notes of v1.1.0 in here and add to them the new things. (think of this as the stable release of message batching).

Overview

In typical real-time applications, sending individual messages results in substantial protocol overhead. Each WebSocket message requires frame headers, system calls, and network packets. The new batching system aggregates messages over configurable time intervals, reducing this overhead and improving overall throughput.

Implementation Details

The batching system operates through a background flusher that periodically combines queued messages and sends them as single WebSocket frames. Messages are collected in per-connection batches and automatically flushed based on time intervals or size thresholds.

Core Components:

• batchFlusher: Manages periodic flushing across all connections
• messageBatch: Per-connection message accumulation
• BatchStrategy: Configurable message encoding formats

Batching Strategies

JSON Array Strategy
Messages are encoded as JSON arrays and sent as binary WebSocket messages:

upgrader.EnableJSONBatching(ctx, 50*time.Millisecond)

Output format: [message1, message2, message3, ...]

Length-Prefixed Binary Strategy
Messages are prefixed with 4-byte big-endian length headers:

upgrader.EnablePrefixBatching(ctx, 50*time.Millisecond)

Output format: [uint32_length][message1][uint32_length][message2]...

Custom Strategy
Allows implementation of application-specific batching formats:

upgrader.EnableBatching(ctx, flushInterval, customSendFunc)

API Methods

When batching is enabled, you have to use batch-specific methods, regular methods will still send messages instantly even if batching is enabled.

Upgrader Methods

• EnableJSONBatching(ctx, flushEvery) - Configure JSON array batching
• EnablePrefixBatching(ctx, flushEvery) - Configure binary length-prefixed batching
• EnableBatching(ctx, flushEvery, sendFunc) - Configure custom batching

Conn methods

• Batch(data) - Adds raw bytes to the batch (doesnt validate and doesnt marshal)
• BatchJSON(v) - Marshals to JSON then calls Batch(data []byte). This is just a convenience method that marshals and calls Batch() under the hood.

Broadcasting methods
These methods are available for rooms and managers.
These methods respect BroadcastBackpressure field in options, check: options.go.

• BatchBroadcast(ctx, data, exclude) (int, error) - Adds a message to all connections batch.
• BatchBroadcastJSON(ctx, data, exclude) (int, error) - Marshals and adds a message to all connections batch.

Configuration

Default Limits:

• Maximum bytes per batch: 1MB
• Default flush interval: 50ms

Note: all these defaults can be changed by you.

Prefix/Suffix data:
Some applications might need to send additional data with each batch, weather that's at the beginning of the message or at the end. For example if JSON array batching is enabled: [prefix-data, message-1, message-2, ......., message-n, suffix-data].

To do the Flusher exposes 2 fields:

• BatchPrefixFunc
• BatchSuffixFunc
  which are of of type: type PrefixSuffixFunc func(conn *Conn, messages [][]byte) []byte

Upgrader.EnableJSONBatching(context.TODO(), time.Second*3)

Upgrader.Flusher.BatchPrefixFunc = func(conn *snapws.Conn, messages [][]byte) []byte {
   	d, _ := json.Marshal(fmt.Sprintf("This is a prefix, %d messages are after this", len(messages)))
   	return d
}

Upgrader.Flusher.BatchSuffixFunc = func(conn *snapws.Conn, messages [][]byte) []byte {
   	d, _ := json.Marshal(fmt.Sprintf("This is a suffix, %d messages came before this", len(messages)))
   	return d
}

Performance Characteristics

Batching provides the most benefit for applications with:

• High message frequency
• Small to medium message sizes
• Multiple concurrent connections
• Real-time requirements with acceptable latency

Backward Compatibility

The batching system is entirely additive. Existing applications continue to function without modification. Batching is opt-in and can be enabled per-upgrader instance.

Technical Notes

• All batching operations are thread-safe
• Context cancellation is respected throughout the batching pipeline
• Connections are automatically cleaned up when closed
• Memory usage is bounded by the configured batch limits
• Graceful shutdown ensures all pending batches are flushed

Other changes

• The logic of UTF-8 validation was reversed, the validation happened in send methods not in read methods. Now the if SkipUTF8Validation is not set to true, UTF-8 will happen in read methods not on writes.
• Added BroadcastJSON() convenience method to Manager and Room.

v1.1.1

• Added GetRoomKeys() for room managers. It returns all the keys of the rooms as a slice of the manager generic key type.

SnapWS v1.1.0

Message Batching System

This release mainly introduces a comprehensive message batching system designed to significantly improve performance in high-throughput WebSocket applications. Message batching addresses the overhead associated with sending numerous small messages by combining multiple messages into single WebSocket frames.

Overview

In typical real-time applications, sending individual messages results in substantial protocol overhead. Each WebSocket message requires frame headers, system calls, and network packets. The new batching system aggregates messages over configurable time intervals, reducing this overhead and improving overall throughput.

Implementation Details

The batching system operates through a background flusher that periodically combines queued messages and sends them as single WebSocket frames. Messages are collected in per-connection batches and automatically flushed based on time intervals or size thresholds.

Core Components:

• batchFlusher: Manages periodic flushing across all connections
• messageBatch: Per-connection message accumulation
• BatchStrategy: Configurable message encoding formats

Batching Strategies

JSON Array Strategy
Messages are encoded as JSON arrays and sent as binary WebSocket messages:

upgrader.EnableJSONBatching(ctx, 50*time.Millisecond)

Output format: [message1, message2, message3, ...]

Length-Prefixed Binary Strategy
Messages are prefixed with 4-byte big-endian length headers:

upgrader.EnablePrefixBatching(ctx, 50*time.Millisecond)

Output format: [uint32_length][message1][uint32_length][message2]...

Custom Strategy
Allows implementation of application-specific batching formats:

upgrader.EnableBatching(ctx, flushInterval, customSendFunc)

API Methods

When batching is enabled, you have to use batch-specific methods, regular methods will still send messages instantly even if batching is enabled.

Upgrader Methods

• EnableJSONBatching(ctx, flushEvery) - Configure JSON array batching
• EnablePrefixBatching(ctx, flushEvery) - Configure binary length-prefixed batching
• EnableBatching(ctx, flushEvery, sendFunc) - Configure custom batching

Conn methods

• Batch(data) - Adds raw bytes to the batch (doesnt validate and doesnt marshal)
• BatchJSON(v) - Marshals to JSON then calls Batch(data []byte). This is just a convenience method that marshals and calls Batch() under the hood.

Broadcasting methods
These methods are available for rooms and managers.
These methods respect BroadcastBackpressure field in options, check: [Backpressure Handling] after this.

• BatchBroadcast(ctx, data, exclude) (int, error) - Adds a message to all connections batch.
• BatchBroadcastJSON(ctx, data, exclude) (int, error) - Marshals and adds a message to all connections batch.

Configuration

Default Limits:

• Maximum bytes per batch: 1MB
• Default flush interval: 50ms

Note: all these defaults can be changed by you.

Backpressure Handling
The system supports three strategies when connection queues become full:

• BackpressureDrop (default): Discard messages for slow clients
• BackpressureClose: Terminate slow connections
• BackpressureWait (not recommended): Block until queue space is available

Performance Characteristics

Batching provides the most benefit for applications with:

• High message frequency
• Small to medium message sizes
• Multiple concurrent connections
• Real-time requirements with acceptable latency

Backward Compatibility

The batching system is entirely additive. Existing applications continue to function without modification. Batching is opt-in and can be enabled per-upgrader instance.

Technical Notes

• All batching operations are thread-safe
• Context cancellation is respected throughout the batching pipeline
• Connections are automatically cleaned up when closed
• Memory usage is bounded by the configured batch limits
• Graceful shutdown ensures all pending batches are flushed

Broadcasting

Broadcasting changed from spawning workers and killing them after the broadcast finishes, to having a broadcastQueue buffered channel per connection, and a go routine that listens for this channel.
This change was due to a Reddit comment i read, and then spent the next 2 days benchmarking and decided to change the broadcasting approach to this.
This change eliminates the overhead of goroutine creation/destruction for each broadcast operation, resulting in better performance and more predictable resource usage.

API Changes

• Removed broadcastWorkers from options
• Added BroadcastChannelsSize (default 8) in options
• Added BroadcastBackpressure (default BackpressureDrop). Broadcasting strategies already explained above, in the batching section.

Manager

• Changed GetConn(key KeyType) (*ManagedConn[KeyType], bool) to: Get(key KeyType) *ManagedConn[KeyType]. This change was made to ensure consistency between room methods and manager methods.
• Removed GetAllConnsWithExclude(exclude ...KeyType)
• Replaced GetAllConns() with GetAllConns(exclude ...KeyType)
• Changed GetAllConnsWithExcludeAsConn(exclude ...KeyType) To GetAllConnsAsConn(exclude ...KeyType)

Additional changes

• In NextWriter(ctx, msgType) method, there was an if check !conn.writer.closed, this mean that if the writer was taken it will error. This is unnecessary and causes unwanted errors. So this if check is removed now. The user should ensure all writers are closed and provide a ctx that times out, because now if the writer is not closed and another goroutine is trying to get the next writer, it will hang forever.

Changes

• Added args to room hooks.
• Updated the room-chat example to match the new feature.

Changes

• Removed unused import of testify.
• In newManagedConn(conn *Conn, key KeyType) i was setting conn.onClose without protecting it with the mutex. So i added lock/unlock of the onCloseMu.
• Added OnJoin/OnLeave hooks for rooms, and DefaultOnJoin/Leave for the room manager.
• In the room struct the "key" field was private, but it was meant to be public. So now it is public.