streamResource

#A first stream

Give it a request function (a URL string is fine) and pick a transport. sse() for Server-Sent Events, websocket() for a socket, or your own StreamTransport. The URL is reactive, same as a query, so changing symbol() tears the old connection down and opens a new one.

import { streamResource, sse, websocket } from '@mmstack/resource';

readonly prices = streamResource<PriceTick>(
  () => `/api/prices/${this.symbol()}/stream`,
  {
    transport: sse(), // or websocket(), or your own StreamTransport
  },
);

// prices.value()     -> the latest message
// prices.connected() -> live-connection indicator

value() holds the most recent message and connected() is a live-connection indicator, the little dot in the corner. Note that those two are separate on purpose, which the status rules below explain.

#Status and connection are different signals

status() stays 'loading' until the first message arrives, because a connection with no data yet honestly isn't ready to render. Once a message lands it flips to 'resolved' and value() tracks every message after that.

connected() is the transport's own state, independent of whether data has arrived. Use status() to decide placeholder versus content, and connected() to show a live or reconnecting badge. A stream can be connected but still 'loading' (open, no message yet), or disconnected while 'resolved' (dropped, holding the last value).

#Reconnection and offline

A live connection's job is to stay alive, so drops reconnect with exponential backoff by default (1s base, 30s cap, persistent). Through the outage the last value stays readable, and only genuinely exhausted retries surface as status: 'error'. reload() starts a fresh attempt budget.

readonly chat = streamResource<ChatMessage>(() => '/api/chat/stream', {
  transport: sse(),
  // reconnect: persistent by default; tune or disable it
  // reconnect: { max: 5, backoff: 2_000 },
  // reconnect: 0, // single-shot
});

It is offline-aware too: while the browser is offline nothing burns attempts, and regaining the network reconnects immediately on a fresh ladder. If you want a single-shot connection instead of a persistent one, pass reconnect: 0.

#Disabling and stopping

Two levers, and they mean different things. Return undefined from the request function to disconnect (status: 'idle'), the same disable pattern as a query. abort() disconnects and stays disconnected, keeping the current value (status: 'local') until a reload() or a source change brings it back. That is what scope.abortPending() reaches, so a stream cancels cleanly with its transition scope.

Streams never connect on the server. A stream never settles, so connecting during SSR would wedge serialization. They are client-only by design, and there is nothing to configure for that.

#Pausing

A socket kept open for a subtree nobody is looking at is wasted network and wasted server fan-out. The pause option closes the live connection while a condition holds and reconnects the moment it lifts, on a fresh backoff ladder. The current value and status stay put, so nothing flickers on resume. It takes the same forms as the query option: true follows the surrounding Activity boundary, a predicate or Signal<boolean> is read directly.

readonly prices = streamResource<PriceTick>(() => '/api/prices/stream', {
  transport: sse(),
  pause: true, // closed while the Activity boundary is paused
  // pause: this.tabHidden, // or any Signal<boolean> / predicate
});

While paused, connected() is false and a send() behaves exactly as it does while disconnected.

#Sending messages

A WebSocket goes both ways, so websocket() is a bidirectional transport and the resource it produces has a send() method. Server-Sent Events are read-only, and so is the resource you get from sse(); the return type follows the transport. Because the socket behind a reconnecting stream changes over time, you never hold it yourself: send() always writes to whichever connection is live.

readonly chat = streamResource<ChatEvent, ChatCommand>(
  () => '/api/chat/socket',
  {
    transport: websocket(), // bidirectional, so the ref has send()
    outbox: true, // queue while reconnecting instead of dropping
  },
);

sendMessage(text: string) {
  const delivered = this.chat.send({ type: 'message', text });
  // false only without outbox, when nothing was there to hand it to
}

// readonly feed = streamResource<Tick>(() => url, { transport: sse() });
// feed.send(...) does not exist: sse() is a read-only transport

With no open connection (connecting, offline, paused, aborted), send() drops the message and returns false. Opt into outbox: true to queue instead. A queued message is addressed to a connection, not to the resource: it flushes in order on the next open of the same source, a source change, reload(), abort() or destroy() discards it, and once retries are exhausted send() returns false rather than queueing into a connection that will never come. That is a deliberate opt-in, because after a long reconnect wait the whole backlog reaches the server at once. Outgoing messages default to JSON.stringify; pass serialize for binary frames or another wire format.

#Parsing and custom transports

Messages default to JSON.parse. Both built-in transports take a deserialize function; sse() also takes an event name and websocket() takes protocols.

streamResource<PriceTick>(() => '/api/prices/stream', {
  transport: sse({
    event: 'tick', // named SSE event
    deserialize: (raw) => JSON.parse(raw) as PriceTick,
  }),
});

The transport option is the extension point. A custom StreamTransport maps any connection-shaped thing (a shared STOMP client's topic, a worker port) onto emit, open, and fail, and the reconnect and status machinery comes with it. Return a connection with a send method (a BidiStreamTransport) and the resource gains send() too. If a consumer wants events rather than the latest-value shape, bridge with toObservable(res.value).

#Recipe: live price ticker

Register the stream as an indicator so a surrounding suspense boundary shows the held value with a busy state rather than a placeholder, and read connected() for a reconnecting hint.

readonly prices = streamResource<PriceTick>(
  () => `/api/prices/${this.symbol()}/stream`,
  {
    transport: sse(),
    register: 'indicator', // hold + busy state inside a suspense boundary
  },
);

readonly live = computed(() => this.prices.connected());