Guides

WebSocket & server-sent events

Live data comes into the core like everything else: through a port. The core reads at its own pace and decides when to reconnect; the platform owns the socket.

Turn it on#

WebSocket and Sse are opt-in ports: a core asks for them with a cargo feature, and a core that does not keeps the schema hash and the wasm size it had. Each platform answers with its own client: URLSessionWebSocketTask, the Kotlin runtime's client on the JVM and Android, the browser's WebSocket.

[dependencies]
undra = { version = "0.1", features = ["websocket", "sse"] }

A connection#

The core connects, sends and reads. Messages arrive as a stream the core pulls: it asks for 16 at a time, so a core that stops reading stops the socket, on the platforms that can pause one.

use undra::ports::ws::{WsConnection, WsError, WsMessage, WsOptions};

let conn = WsConnection::connect(&ctx, "wss://chat.example.com/room", WsOptions::default()).await?;
conn.send_text("hello").await?;
let mut messages = conn.messages();
while let Some(next) = undra::ports::next(&mut messages).await {
    match next {
        Ok(WsMessage::Text(text)) => room.push(text),
        Ok(WsMessage::Binary(_)) => {}
        Err(WsError::Closed { code: 1000, .. }) => break, // the server said goodbye
        Err(error) => return Err(error),
    }
}

Every end is typed#

Browsers cannot send headers with a WebSocket, so the browser adapter refuses a connect that has some instead of dropping them: put the credential in the URL or a subprotocol.

Reconnecting is yours#

The port never reconnects on its own: which close codes to retry, how long to wait and what to resend are your app's rules, and they run in the core, where a test drives them with a fake clock. Backoff gives the delay, a weak context sleeps.

let backoff = Backoff::default(); // 500 ms doubling to 30 s, with jitter from the Rng port
let mut attempt = 0;
while weak.sleep(backoff.delay(attempt, &*ctx.rng())).await.is_ok() {
    attempt += 1;
    // connect again, re-subscribe, reset `attempt` once a message arrives
}

Server-sent events#

sse::subscribe is a stream of events parsed as the HTML standard says. It ends with Ended when the server finishes the response; subscribe again with the last event id to resume.

let mut events = sse::subscribe(&ctx, "https://api.example.com/feed", vec![], last_id.clone());
while let Some(Ok(event)) = undra::ports::next(&mut events).await {
    last_id = event.id.clone().or(last_id);
    feed.apply(event.data);
}

In tests#

FakeWebSocket and FakeSse are scripted servers: accept or refuse, push, close, drop, echo. They record what the core sent and how much it pulled, so backpressure is an assertion.

let fakes = fakes::install(&t);
fakes.web_socket.echo(true);
// ... the core connects and sends ...
assert!(fakes.web_socket.delivered(conn) <= consumed + 24);

On the web the adapters are an opt-in import, @undra/runtime/realtime, so an app that does not use them ships none of them. Design: ADR-047.