Skip to content
MAATDocumentationLatest

Typed Realtime Clients

A Maat handler broadcasts an event; a Flutter widget rebuilds. Between those two lines are a WebSocket, a signature, a reconnect and a token that expires. package:horus_client/realtime.dart is that middle.

final Stream<OrderUpdated> updates = realtime.stream(orderUpdated(7));

Read Broadcasting for the server half and WebSockets with Thoth for the server itself.

The three pieces

import 'package:horus_client/realtime.dart';


// 1. The typed event
final class OrderUpdated {
  const OrderUpdated({required this.id, required this.status});


  factory OrderUpdated.fromJson(Map<String, Object?> json) => OrderUpdated(
    id: json['id']! as int,
    status: json['status']! as String,
  );


  final int id;
  final String status;
}


// 2. Where it arrives. The channel matches the event's broadcastOn(),
//    prefix included; the event name matches its broadcastAs().
ChannelEvent<OrderUpdated> orderUpdated(int id) => ChannelEvent(
  channel: 'private-orders.$id',
  event: 'OrderUpdated',
  decode: (json) => OrderUpdated.fromJson(json! as Map<String, Object?>),
);


// 3. The connection
final realtime = HorusRealtime(
  socketUri: Uri.parse('ws://localhost:6001/app/$appKey'),
  client: api,
);

ChannelEvent<T> is the realtime twin of Endpoint<T>: a location and a decoder. The two halves of a client read alike on purpose.

In Flutter

StreamBuilder<OrderUpdated>(
  stream: realtime.stream(orderUpdated(order.id)),
  builder: (context, snapshot) => switch (snapshot) {
    AsyncSnapshot(hasError: true) => const Text('Reconnecting…'),
    AsyncSnapshot(:final data?) => Text(data.status),
    _ => Text(order.status),
  },
);

Nothing in the module imports Flutter, and nothing imports dart:io on a path the browser reaches — package:web_socket_channel picks the browser or the IO implementation — so the same code runs on mobile, desktop and web.

A channel is subscribed when its first listener arrives and dropped when its last one cancels, so a screen that pops stops paying for it. Two widgets on one channel share a single subscription.

Private channels

A private or presence channel needs a signature from the application before Thoth will accept the subscription. HorusRealtime gets one by POSTing to /broadcasting/auth through the HorusClient you gave it — the same base URL and the same credentials as every other call, which is why it takes a client rather than a URL and a token of its own.

final api = HorusClient(
  Uri.parse('https://orders.example.com'),
  headersProvider: () async => {'authorization': 'Bearer ${await tokens.access()}'},
);
final realtime = HorusRealtime(socketUri: socketUri, client: api);

The server decides who may subscribe, in a provider:

Broadcast.channel('orders.{id}', (user, params) =>
    ordersOwnedBy(user).contains(int.parse(params['id']!)));

A refusal errors that one stream — as a ChannelAuthorizationException — and leaves the rest of the connection running. One order a customer may not watch is not a reason to drop the four they may.

Public channels need no client at all.

Reconnect, and why it re-authorizes

Thoth signs a subscription as HMAC(secret, "$socketId:$channelName"). A reconnect gets a new socket id, so every signature the client is holding is already dead.

HorusRealtime therefore does not replay stored subscribe frames. On each reconnect it waits for the new socket id, re-authorizes every channel that still has listeners, and resubscribes with fresh signatures. A channel whose last listener cancelled while the connection was down is not resubscribed.

Retries use exponential backoff with full jitter — 500 ms doubling to a 30 s cap. The jitter is not decoration: without it, every client dropped by one server restart comes back at the same instant and knocks it over again.

HorusRealtime(
  socketUri: socketUri,
  client: api,
  backoff: exponentialBackoff(base: Duration(seconds: 1), cap: Duration(minutes: 1)),
);

A refusal that retrying cannot fix — a wrong app key (4001) or an unsupported protocol (4007) — stops instead of spinning and errors every listener so the application can say so. A capacity refusal (4100) keeps the client on its normal reconnect backoff because room may become available.

Watch the connection with realtime.statuses, a Stream<RealtimeStatus> of connecting, connected, reconnecting and disconnected.

Expired tokens

onAuthFailure runs once when /broadcasting/auth answers 401 or 403, then the authorization is retried:

HorusRealtime(
  socketUri: socketUri,
  client: api,
  onAuthFailure: () => tokens.refresh(),
);

Because headersProvider is consulted per request, the refreshed token reaches the retry with nothing else to wire up. A second refusal is a real refusal, not an expired token: that channel’s stream gets the error and the rest of the socket carries on.

Not hearing your own events

A client that just made a change already has the new state in the response. Receiving the broadcast too makes it apply the change twice.

Give the HTTP client the connection’s socket id and the problem goes away:

api.socketIdProvider = () => realtime.socketId;

Every request then carries X-Socket-ID, which is what PendingBroadcast.toOthers(request) reads on the server:

await broadcast(OrderUpdated(...)).toOthers(request).send();

The exclusion is per socket, not per channel — every other subscriber still receives it. While the connection is down socketId is null and no header is sent, so a stale id never excludes a socket that no longer exists.

Not covered yet

  • Presence member lists and client- events. This client ignores those frames rather than half-supporting them. The gap is client-side only: Thoth implements both — presence member lists at thoth_handler.dart:178,221-230,332-345 and client- event forwarding at :395 — so a JavaScript client talking to the same server gets them today. See Realtime in the Browser for Horus, the browser client that does.
  • Generating ChannelEvents from the server. The typed event is declared by hand. The shape above is the one a generator would emit, so adopting one later changes nothing at the call site. See Building JSON APIs for the request half.
  • Redis and multi-node fan-out. Thoth’s registry is in memory and one process is the supported deployment; see WebSockets with Thoth.

The orders example runs all of this against a real Thoth, a real Maat application and a real socket.