🥮 mooncakes.io
README

bobzhang/crescent/websocket does not have a README file

#
NativeWebSocketOverflowPolicy

source
pub(all) enum NativeWebSocketOverflowPolicy {
  DropOldest
  DropLatest
} derive(Eq, 
Debug)

Policy for handling a full outbound WebSocket message queue.

#
WebSocketAggregatedMessage

source
pub(all) enum WebSocketAggregatedMessage {
  Text(String)
  Binary(Bytes)
}

A fully assembled WebSocket message, either text or binary.

#
WebSocketEvent

source
pub(all) enum WebSocketEvent {
  Open(WebSocketPeer)
  Message(WebSocketPeer, WebSocketAggregatedMessage)
  Close(WebSocketPeer)
}

Events delivered to a WebSocket handler: open, message, or close.

#
WebSocketHandler

source
pub(all) struct WebSocketHandler((WebSocketEvent) -> Unit)

Handler function type for WebSocket route events.

#
WebSocketPeer

source
pub struct WebSocketPeer {
  // private fields
}

Represents a connected WebSocket client with its connection ID, subscribed channels, and route parameters captured from the upgrade URL.

#
WebSocketPeer::WebSocketPeer

source
fn WebSocketPeer::WebSocketPeer(connection_id~ : String, params? : Map[String, String]) -> WebSocketPeer

Creates a new WebSocketPeer with the given connection ID and optional route parameters (extracted from dynamic WebSocket routes like /ws/:room).

#
WebSocketPeer::binary

source
fn WebSocketPeer::binary(self : WebSocketPeer, message : Bytes) -> Unit

Sends a binary message to this WebSocket peer.

#
WebSocketPeer::connection_id

source
fn WebSocketPeer::connection_id(self : WebSocketPeer) -> String

Returns the unique connection ID assigned to this peer by the runtime. Stable for the lifetime of the connection.

#
WebSocketPeer::param

source
fn WebSocketPeer::param(self : WebSocketPeer, name : String) -> String?

Returns a route parameter captured from the WebSocket upgrade URL, or None if the parameter was not present.

For example, a route /ws/:room matched against /ws/lobby makes peer.param("room") return Some("lobby").

#
WebSocketPeer::publish

source
fn WebSocketPeer::publish(self : WebSocketPeer, channel : String, message : String) -> Unit

Publishes a text message to a pub/sub channel on behalf of this peer.

#
WebSocketPeer::subscribe

source
fn WebSocketPeer::subscribe(self : WebSocketPeer, channel : String) -> Unit

Subscribes this WebSocket peer to the given pub/sub channel.

#
WebSocketPeer::text

source
fn WebSocketPeer::text(self : WebSocketPeer, message : String) -> Unit

Sends a text message to this WebSocket peer.

#
WebSocketPeer::to_string

source
fn WebSocketPeer::to_string(self : WebSocketPeer) -> String

Returns a string representation of this WebSocket peer.

#
WebSocketPeer::unsubscribe

source
fn WebSocketPeer::unsubscribe(self : WebSocketPeer, channel : String) -> Unit

Unsubscribes this WebSocket peer from the given pub/sub channel.

#
DEFAULT_OUTGOING_QUEUE_CAPACITY

source
let DEFAULT_OUTGOING_QUEUE_CAPACITY : Int

Default per-connection capacity for the outbound WebSocket message queue, used when NativeServeOptions::websocket_outgoing_queue_capacity is None.

#
channel_member_count

source
fn channel_member_count(channel : String) -> Int

Returns the number of subscribers to the given channel across all runtimes.

#
channel_member_count_in_runtime

source
fn channel_member_count_in_runtime(runtime_id : String, channel : String) -> Int

Returns the number of subscribers to the given channel within a specific runtime, or 0 if the runtime is not registered.

#
cleanup_runtime

source
fn cleanup_runtime(runtime_id : String) -> Unit

Tears down all hub state associated with a runtime ID: closes outgoing queues, removes channel memberships, and drops the hub. Called via defer from App::serve_on when serving ends, and used by tests to reset state between cases. Safe to call on a runtime ID that was never registered.

#
connection_is_registered

source
fn connection_is_registered(connection_id : String) -> Bool

Returns true if the given connection ID is registered with any runtime.

#
handle_route_async

source
async fn handle_route_async(runtime_id : String, request : 
Request, conn : 
ServerConnection, handler : WebSocketHandler, params : Map[String, String], max_message_bytes : Int?, outgoing_queue_capacity : Int, overflow_policy : NativeWebSocketOverflowPolicy, read_timeout_ms : Int?) -> Unit

Runs one accepted WebSocket upgrade to completion: handshake, hub registration, the Open → Message* → Close user-handler lifecycle, and cleanup.

Called by App::handle_request's WebSocket dispatch branch. Unlike HTTP handle_request (which runs once per request), this function is long-lived — it returns only when the peer loop exits (client close, read timeout, oversized message) or when the enclosing task group is cancelled (server shutdown).

Lifecycle phases:

  1. Handshake. @async_websocket.from_http_server validates the upgrade headers and writes the 101 Switching Protocols response. InvalidHandshake is swallowed silently — no Open/Close events fire, the underlying conn closes via defer ws.close(). Other errors propagate (surfaces to the keep-alive loop, which closes the conn).
  2. Registration. Allocate a connection_id, create a bounded outbound queue (Blocking(outgoing_queue_capacity)), register with the hub. The overflow_policy is recorded in the hub so ws_publish can drop oldest/latest on a full queue.
  3. Writer task. write_native_ws_outgoing is spawned as a sibling inside the task group: it drains outgoing into the socket. It exits when outgoing.close() fires (normal exit) or when the task group is cancelled.
  4. Peer loop. Fire Open once, then loop: recv_native_ws_message reads a frame header (respecting read_timeout_ms); read_native_ws_message_contents aggregates continuation frames up to max_message_bytes (oversized → 1009 close sent internally, returns None → we break). Each complete message fires Message. Either helper returning None ends the session.
  5. Shutdown. outgoing.close() signals the writer task to exit. handler(Close) fires (see invariants below).

Key invariants:

  • Close fires exactly once whenever Open did. Normal exit reaches the final handler(Close(...)) at the bottom; error/cancellation goes through the catch block, which calls Close and re-raises. If the handshake fails (InvalidHandshake), neither Open nor Close fires.

  • Cancellation sends a polite GoingAway close frame before the socket dies. send_native_ws_shutdown_close uses protect_from_cancel so the frame actually reaches the wire even though we're already in a cancelled task. Without this, clients would see an abrupt TCP reset on server shutdown instead of a 1001 close.

  • The two defers guarantee cleanup on every path, including user handler exceptions during Close. defer ws.close() drops the socket; defer unregister_native_ws_connection removes the hub entry, channel memberships, and any subscriptions the user added during Close. Running unregister only on the happy path would leak those on any exception inside the Close handler.

#
registered_connection_count

source
fn registered_connection_count() -> Int

Returns the number of registered WebSocket connections across all runtimes.

#
registered_runtime_count

source
fn registered_runtime_count() -> Int

Returns the number of currently registered WebSocket runtimes.

#
registered_runtime_ids

source
fn registered_runtime_ids() -> Array[String]

Returns all currently registered runtime IDs. Intended for tests and debugging; production code should not need this.

#
runtime_id_for_connection

source
fn runtime_id_for_connection(connection_id : String) -> String?

Returns the runtime ID that owns this connection, or None if the connection is not registered.

#
runtime_is_registered

source
fn runtime_is_registered(runtime_id : String) -> Bool

Returns true if the given runtime ID has a registered hub.

Source Files

Powered by MoonBit

Site sourceReport issuePackagesBuild queueSkillsStatistics

© 2026 mooncakes.io