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.

#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. 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());