Streams & Channels

Overview

A stream delivers server-sent events; a channel is a WebSocket subscription, which the caller can also send on when it names .send(). Either one is served by the process that defines it with .implement(), or describes where another process serves it with .get(), .connect() or .via().

import { courier, defineGroup } from "@noverax/courier/define"
import { z } from "zod"

const { channel } = courier()

export const chat = defineGroup({
  prefix: "chat",
  endpoints: {
    room: channel
      .variables(z.object({ room: z.string(), nick: z.string() }))
      .topic((v) => ({ room: v.room }))
      .send(z.object({ text: z.string() }))
      .response(z.object({ from: z.string(), text: z.string() }))
      .implement((v) => ({
        onJoin(ctx) {
          ctx.send({ from: "server", text: `welcome to ${v.room}` })
        },
        onMessage(ctx, message) {
          ctx.broadcast(ctx.topic, { from: v.nick, text: message.text })
        },
      })),
  },
})

Serving a stream

A stream's .implement() takes an async generator, handed the subscription's variables, a signal that aborts when the client leaves, the lastEventId a reconnecting client resumes after, and the request's headers. Each event it yields is { data }, with an event name or an id where it needs them; tracked(id, data) gives the event an id, so a client that reconnects is handed only what it missed:

import { courier, defineGroup } from "@noverax/courier/define"
import { ForbiddenError } from "@noverax/courier"
import { tracked } from "@noverax/courier/realtime"
import { z } from "zod"

const { stream, channel } = courier()

export const orders = defineGroup({
  prefix: "orders",
  endpoints: {
    updates: stream
      .variables(z.object({ shop: z.string() }))
      .response(z.object({ id: z.string(), state: z.string() }))
      .implement(async function* (ctx) {
        if (!(await canWatch(ctx.headers, ctx.variables.shop))) {
          throw new ForbiddenError("Not a member of this shop")
        }
        for await (const order of orderChanges(ctx.variables.shop, ctx.lastEventId, ctx.signal)) {
          yield tracked(order.id, order)
        }
      }),
    // A channel nobody sends on is served the same way; it yields the values themselves
    ticker: channel.response(z.number()).implement(async function* () {
      yield 1
    }),
  },
})

Returning ends the stream: the client stops reading and does not reconnect. Throwing fails it. A RealtimeError — ForbiddenError, AuthError, RateLimitError, TimeoutError, or one built with its own code — reaches the client with its code and message; any other error reaches it as code 5000, "Stream failed", and in full only the server's log.

Reaching a stream or a channel

A stream or channel served with .implement() is reached through the server that serves it — the proxy routes of your framework adapter, or a standalone server — so a proxy caller opens it: createProxyCaller, or createProxyQueryCaller for the React hooks.

A direct caller, createCaller or createQueryCaller, sends everything to the API's baseURL, which is the upstream's address. Nothing serves this process's streams and channels there, so calling one through a direct caller throws ConfigurationError("DIRECT_CALL_TO_SERVED_ENDPOINT"), naming it, when it is called — the React hooks report it through error — and the rest of the API stays callable. A stream or channel addressed with .get(), .connect() or .via() is another process's, and a direct caller opens it at baseURL.

A proxy caller reaches another process's stream or channel too: the proxy relays it, opening it where its definition says. A stream is opened with the source's headers, the request headers the proxy lets through, the user's session, and whatever the source's middleware sets. A browser reaches a third party's stream through its own server, with no CORS to configure and no key of its own to hold. A channel's upstream socket is opened with the same headers, as a direct caller in the server's process would open it; the subscriptions relayed with the same headers share it, so one user's channels share a socket and no two users' do. The proxy relays the streams and channels it would forward as endpoints: one a defineBoundary leaves out stays unreachable.

• A stream's events pass on as they came: ids, event names, and data that is not JSON as its text. A dropped upstream connection is reopened from the last event id, while the client's stays open. The client's Last-Event-ID is sent upstream when it resumes.
• The stream ends as the upstream's ends. It fails as the upstream's fails: a refusal carries the upstream's status as code 4000 + status, so a 401 arrives as 4401. A client that leaves closes the upstream stream.
• A channel's frames pass both ways, and it ends as its upstream subscription ends, with its code — a connection the upstream refuses included. A message the upstream subscription has no room to hold, while it joins or comes back, is answered with a 4429 error on the channel, which goes on. Subscribers of the server with equal variables share one upstream subscription.
• A path parameter that is empty, . or .. is refused before anything is sent, as it is for an HTTP call. Otherwise it would reach a route above the declared one with the server's credentials.

How a subscription ends

A subscription ends in one of three ways, and the client can tell them apart.

• The client closes it — close(), or a React hook unmounting. Nothing is reported.
• The server finishes it — a stream's generator returned, or a channel closed itself. Iterating a stream ends (for await leaves its loop); a channel's status becomes disconnected. Neither reconnects.
• The server fails it — the handler threw, the subscription or the join was refused, its variables failed their schema, or a limit was reached. Iterating a stream rejects with a RealtimeError carrying the code and message the server sent; a channel emits it on error, then its status becomes disconnected. A refusal answered with an HTTP status arrives as code 4000 + status, so a 401 is 4401.

A dropped connection is none of these: the client reconnects, resumes a stream after the last event id it saw, and subscribes its channels again. When it has spent its retries, iterating the stream rejects with a ConnectionError.

import { isRealtimeError } from "@noverax/courier"

const updates = client.orders.updates({ shop: "north" })
try {
  for await (const event of updates) render(event.data)
  // the server finished the stream
} catch (error) {
  if (isRealtimeError(error) && error.code === 4403) showNoAccess(error.message)
  else throw error
}

const room = client.chat.room({ room: "general", nick: "ada" })
room.on("error", (error) => notify(error.message)) // a refused message: the channel goes on
room.on("status", (status) => setOnline(status === "connected"))

A channel's error event also reports what it refused without ending: a message that failed its schema (4422), one larger than the server accepts (4413), or an onMessage that threw. send() returns whether the message went out or waits for the channel; false means it was dropped, with a warning.

Topics

Every subscription to a channel is on a topic: the endpoint, source.group.method, qualified by a key. ctx.topic is the subscriber's own, and ctx.broadcast(topic, data) sends to every other subscriber on it — every replica's, with a pubsub adapter.

Without .topic(), the key is all of a subscriber's variables: subscribers share a topic only when their variables are equal. That is the safe default for a feed whose variables pick what it shows, but it splits a room into as many topics as it has members as soon as a variable belongs to one member — a nickname, a cursor colour.

.topic() names what subscribers share a topic by. It reads the variables as the schema produced them, and returns what JSON can carry. In the channel above, everyone in room r1 is on one topic, so a broadcast reaches all of them, while each keeps its own nick in v and ctx.variables. A key that throws ends that subscription with code 5000.

On one connection, subscriptions whose key is equal are one subscription: the newer subscribe replaces the older, whose onLeave runs, and both of the client's channels receive what is sent on the topic. Two browsers, or two tabs, are two connections.

.topic() is named before .implement(). A channel another process serves has its topics decided there, so naming it before .connect() or .via() throws ConfigurationError("TOPIC_ON_REMOTE_CHANNEL").

What a channel sends

A stream's events, a push channel's values and what a channel's handlers pass to ctx.send() and ctx.broadcast() are held to the endpoint's .response() schema before they leave the server, and the client receives the schema's output — defaults filled in, transforms applied. A channel's ctx.send() or ctx.broadcast() of a value that does not match throws a RealtimeValidationError where it was called and sends nothing; a channel's response schema must validate synchronously. Validation has the details.

Message size

A server accepts messages up to its maxMessageSize, 64 KB by default, and answers a larger one with a 4413 error on a connection that stays open. The client does not send one: a channel's send of a message whose encoded frame is longer than its own maxMessageSize — 64 KB unless the caller's ws says otherwise — puts nothing on the wire: send answers false, and the channel's error event reads code 4413. The channel and everything else on the socket go on.

A server whose WebSocket layer closes the socket over a frame too large to read, with close code 1009, loses that one message only: the client reconnects, and each channel on the socket hears a 1009 error.

Connection options

Every channel and push subscription of one caller on one URL shares a socket. The caller's ws option sets that socket up, in createCaller, createProxyCaller, createQueryCaller and createProxyQueryCaller alike:

import { createProxyCaller } from "@noverax/courier"
import type { api } from "@/api"
import { boundary } from "./boundary"

export const client = createProxyCaller<typeof api>("api", {
  boundary,
  ws: {
    auth: async () => ({ token: await session.accessToken() }),
    backoff: { initialMs: 500, maxMs: 10_000 },
    heartbeat: { intervalMs: 25_000, timeoutMs: 10_000 },
    maxMessageSize: 65_536,
  },
})

• auth — what the client tells the server about itself, as socket.io's auth does: an object, or a function that returns one, called anew before every connect and reconnect so the token in it is current. It is sent in the socket's first frame, never in its URL, where proxies and servers log it, and the server's onConnect reads it as auth — see Who a connection is. It is sent again, asked for anew, when the server asks with sendAuthRequired; a caller without it leaves that request unanswered, and the server closes the connection at its deadline.
• backoff — the delay before each reconnect: initialMs (1 s), maxMs (30 s), factor (2), jitter (on).
• heartbeat — the server's ping interval and pong timeout; a server silent for longer than the two together is taken for gone, and the client reconnects.
• messageQueue — how much is kept to send while reconnecting: maxCount (100) and maxBytes (1 MB). A message beyond it is dropped with a warning, and its send answers false.
• protocols — WebSocket sub-protocols for the handshake.
• maxMessageSize — the largest message sent; see Message size.

A caller with ws options opens a socket of its own, beside the one another caller uses on the same URL, so its options are never dropped in favour of the other's.

Who a connection is

A browser's WebSocket carries the page's cookies on its handshake, and nothing else: no script can put an Authorization or an API key on it, and a token put in the URL ends up in the logs of every proxy and server it passes. So a Courier client says who it is in a frame of its own, the first it sends on every socket — before its subscriptions, on every reconnect too:

• its ws.auth, which the server's onConnect reads as auth;
• the headers it could not put on the handshake, which the server reads as the handshake's own: onConnect's request carries them, and so does every subscription's middleware and session. A header the handshake has keeps its value, and none a browser or a proxy sets itself — Cookie, Origin, Host, the Sec- and forwarding headers — is taken from the frame.

Those headers are the ones the caller's calls send: the API's headers, the caller's headers, and the user's session — its bearer token, and on a server its cookies and forward's headers. Where the platform's WebSocket takes headers on its handshake — on a server, in a React Native app — they go there instead, and the frame carries auth alone.

const server = await createCourierServer({
  sources,
  onConnect: async ({ request, auth }) => {
    const user = await users.byToken(auth?.token ?? bearerOf(request))
    if (!user) throw new Error("Unauthorized")
    return { userId: user.id }
  },
})

The server holds what a client sends until onConnect has admitted its connection, then handles it in order, so a subscription made as the socket opens is not lost while the user is looked up. A client that sends no auth frame — one that is not Courier's — is admitted on its handshake alone once its first frame comes, or after authTimeoutMs (3 s).

To have a client prove who it is again on an open connection — its token about to expire — call sendAuthRequired(connections, connectionId, timeoutMs): the client sends its credentials again, asked for anew, and onReauth reads them as onConnect does and returns the connection's new metadata. Credentials it refuses, or none in time, close the connection with 4401.

Next steps

• React: Streams & Channels — follow a stream, keep a query up to date, and send on a channel from components
• Standalone Server — serve streams and channels from their own process
• Redis Adapters — broadcast across replicas, resume streams, track presence