DocsServer features
WebSockets
An rpc is a question with one answer, and a stream is an answer that arrives in pieces. A socket is the third shape: both sides may speak, at any time, for as long as the page is open.
Optional, and absent unless asked for
The #[socket] attribute and the rahti::ws module exist only when the application's rahti dependency enables the ws cargo feature. A project that did not ask compiles none of the socket wire — the feature is what keeps tokio-tungstenite out of an application that never opens a connection.
# Cargo.toml — the attribute and rahti::ws exist only behind this feature.
rahti = { version = "0.1", features = ["ws"] }cargo rahti new my-app --ws # a new project
cargo rahti upgrade --ws # a project that started without itThe config key is what the tooling reads to know the feature is wanted, so a project whose config says "ws": true and whose manifest does not name the feature is repaired by running upgrade again.
The Rust signature
use rahti::ws::Socket;
#[socket]
pub async fn chat(room: String, mut socket: Socket) -> rahti::Result<()> {
while let Some(message) = socket.recv::<String>().await? {
if !socket.send(&message).await {
break; // The browser is gone.
}
}
Ok(())
}#[socket]takesauth, or nothing.#[socket(auth)]refuses the handshake with a 401 unless the browser carries a session.- the function must be
async— the conversation only exists behind an await; - the last parameter is the
Socket. Everything before it arrives in the connection's first frame, read by the parameter's own name under the rpc payload rules; - return nothing, or
rahti::Result<()>. A socket answers through its frames, so a returned value would have nowhere to go — the macro refuses one. AnErrreaches the browser as an error frame, then the close; - the original function stays callable from Rust, but only the wire builds a
Socket.
The socket
| Call | Meaning |
|---|---|
| socket.recv::<T>().await | The next frame as T. Ok(None) is the connection closing — the natural loop is while let Some(v) = socket.recv().await?. |
| socket.send(&value).await | One frame out. Returns false once the browser is gone: a signal to stop producing, not an error. |
| socket.text().await | The next frame as it arrived, for a handler that parses it itself. |
| socket.sender() | A cloneable SocketSender another task may hold. |
| socket.close().await | Say goodbye first. Dropping the socket closes the connection too. |
| rahti::auth::session() | Answers inside a handler as it does in an rpc. sign_in/sign_out do not: there is no response left to write a cookie onto. |
Connections close themselves when the process is told to stop, so an open socket never holds Ctrl+C.
The client
const sock = pp.socket("chat", { room: "lobby" }, {
onOpen: () => setStatus("open"),
onMessage: (value) => append(value),
onError: (error) => showError(error),
onClose: ({ code, reason, wasClean }) => setStatus("closed"),
});
sock.send({ text: "hello" });
sock.close();pp.socket returns its handle synchronously. Frames sent before the connection opens are buffered and flushed after the argument frame, so opening and sending on consecutive lines works. sock.send returns false once the connection is closed.
The wire
GET /__pulsepoint/ws?name=<function> upgraded to a WebSocket
first text frame -> { "room": "lobby" } the arguments
every frame after -> one JSON value, either direction
{ "error": "…" } -> reserved: failure inside an open connectionOnly the name travels in the URL; the arguments are the first text frame — one JSON object, exactly the payload pp.rpc would have posted — because a URL is logged by every proxy on the way and an argument is data. A frame carrying the error key alone is reserved: it is how the server reports failure inside an open connection, where no HTTP status line exists any more, and the client routes it to onError.
Everything refusable is refused before the upgrade, as HTTP: an unnamed or unknown socket, a cross-origin page, a missing session on #[socket(auth)].
Who may connect
The same-origin policy does not cover WebSockets: another site's page can open one here, and the browser attaches the cookies. What that page cannot do is forge its Origin header, so the handshake is refused when Origin disagrees with Host — the socket counterpart of the rpc CSRF check. Schemes are not compared, because a TLS-terminating proxy rewrites them, and a request with no Origin at all is a non-browser client carrying nothing to forge.
Broadcast: one message, every browser
A handler's Socket speaks to the one browser that connected, so an echo answers privately and two tabs never hear each other. That is the default on purpose — a socket is a conversation. A chat room is the other shape, built out of socket.sender(): keep one cloned sender per connection in shared state, and a broadcast is a walk down the list.
fn room() -> &'static Mutex<Vec<SocketSender>> {
static ROOM: OnceLock<Mutex<Vec<SocketSender>>> = OnceLock::new();
ROOM.get_or_init(|| Mutex::new(Vec::new()))
}
#[socket]
pub async fn chat_room(name: String, mut socket: Socket) -> Result<()> {
room().lock().unwrap().push(socket.sender());
while let Some(said) = socket.recv::<Said>().await? {
// Clone the list out from under the lock before sending: a Mutex
// guard must not be held across an `.await`.
broadcast(&name, said.text).await;
}
Ok(())
}Two rules keep it honest:
- clone the list out from under the lock before sending — a
Mutexguard must not be held across an await; - a send that returns
falseis a browser that left. Prune on that answer (senders.retain(SocketSender::is_open)) rather than keeping a ledger.
The room state is the application's, like the database: Rahti provides the senders and says nothing about how they are grouped. Rooms-by-name is a HashMap<String, Vec<SocketSender>> keyed by an argument from the first frame; the shape above is the one-room case.