utd_channels 0.2.0

Standalone Pusher-style realtime SDK for UTD-Stream-Engine: channels (public/private/presence), event pub/sub, presence member lists, and typed connection lifecycle. Pure Dart — no Flutter, no UI, no [...]

utd_channels

A standalone, Pusher-style realtime SDK for UTD-Stream-Engine: channels (public / private- / presence-), event pub/sub as Dart streams, live presence member lists, and ack-based client events with typed errors. Pure Dart — no Flutter dependency, no UI, no UTD kits required — so it runs in Flutter apps, server-side Dart, and plain tests alike. Channel authorization is byte-compatible with Pusher's HTTP contract: your existing pusher-http-node/-php/Laravel auth endpoint works unchanged.

A complete chat app in ~60 lines

import 'package:utd_channels/utd_channels.dart';

class ChatMessage {
  ChatMessage({required this.text, this.sentAt});

  factory ChatMessage.fromJson(Map<String, dynamic> json) => ChatMessage(
        text: json['text'] as String? ?? '',
        sentAt: DateTime.tryParse(json['sent_at'] as String? ?? ''),
      );

  final String text;
  final DateTime? sentAt;
}

Future<void> main() async {
  final client = UTDChannelsClient(
    appId: 'app_9f2c',
    // Dev quickstart: publishable key, client-asserted identity.
    // Production: UTDAuth.tokenProvider / UTDAuth.endpoint — see below.
    auth: const UTDAuth.insecureDevelopment(
      appKey: 'key_x',
      identity: 'alice',
      name: 'Alice',
    ),
    // Dev: needs `channels_insecure_dev` enabled on the project. Production:
    // UTDChannelAuth.endpoint(Uri.parse('https://api.my.app/utd/channel-auth')).
    channelAuth: const UTDChannelAuth.insecureDevelopment(),
  );

  client.connectionStates
      .listen((c) => print('connection: ${c.previous} -> ${c.current}'));

  // Subscribing is synchronous: bind immediately, the subscription is queued
  // until the socket connects and re-authorized after every reconnect.
  final room = client.subscribe('presence-room-7') as UTDPresenceChannel;

  // Raw events — `userId` is server-attested, never taken from the payload.
  room.bind('client-message')
      .listen((e) => print('${e.userId}: ${e.dataAsMap['text']}'));

  // Or land payloads directly in your own model class.
  room.bindJson<ChatMessage>('client-message', ChatMessage.fromJson)
      .listen((m) => print('decoded: ${m.text} at ${m.sentAt}'));

  room.memberAdded.listen((m) => print('${m.id} joined (${m.info?['name']})'));
  room.memberRemoved.listen((m) => print('${m.id} left'));

  await client.connect();
  await room.whenSubscribed;
  print('${room.members.length} online, me = ${room.me?.id}');

  // Ack-based (unlike Pusher's fire-and-forget): completes when the server
  // accepts, throws typed exceptions when it refuses.
  await room.trigger('client-message', {
    'text': 'hello',
    'sent_at': DateTime.now().toIso8601String(),
  });

  // When you're done: closes every channel stream and the socket.
  // await client.dispose();
}

A runnable two-terminal version of this app lives in example/utd_channels_example.dart.

Authentication

Two independent concerns, two knobs:

Concern	Constructor parameter	Answers
Connection auth	auth: (an UTDAuth)	Who is this socket? Mints the session token the socket authenticates with.
Channel auth	channelAuth: (a UTDChannelAuth)	May this socket join private-x? Signs individual subscriptions.

Connection auth (UTDAuth)

Development — UTDAuth.insecureDevelopment(appKey:, identity:, name:, deviceId:) mints a session straight from the client using the publishable app key (POST {mintBaseUrl}/api/v1/auth/session). The identity is client-asserted and not verified — anyone holding the publishable key can claim any identity. The name is the warning; never ship it.

Production — your backend verifies the user, mints the session, and hands the client only the result:

auth: UTDAuth.tokenProvider(() async {
  final r = await myApi.post('/realtime/token'); // your own auth applies
  return UTDConnectionToken(
    token: r['token'] as String,
    wsUrl: r['ws_url'] as String,
    expiresAt: DateTime.tryParse(r['expires_at'] as String? ?? ''),
  );
}),
// or, if the endpoint speaks the contract below directly:
auth: UTDAuth.endpoint(
  Uri.parse('https://api.myapp.com/realtime/token'),
  headers: () async => {'Authorization': 'Bearer $mySessionJwt'},
),

The provider/endpoint is called on every (re)connect, so expired tokens refresh transparently; when expiresAt is known the socket re-mints and reconnects proactively before expiry.

Backend mint contract — your token endpoint:

1. Authenticates the caller with your app's own session mechanism.
2. Calls the engine mint, POST https://api.udt-stream.com/api/v1/auth/session, headers X-App-Id / X-App-Key, JSON body {"identity": "<verified user id>", "name"?: "...", "device_id"?: "..."}.
3. Relays the engine response — {"user_token": "...", "ws_url": "...", "expires_in": 3600} — to the client. UTDAuth.endpoint expects 200 {"token": ..., "ws_url": ...} and accepts user_token/url as aliases plus optional expires_in (seconds) or expires_at (ISO-8601), so relaying the engine body verbatim works as-is.

The difference from dev mode is where the identity comes from: your backend asserts an identity it verified, so the client can never impersonate.

Channel auth (UTDChannelAuth)

Required for any private-/presence- subscription (subscribe throws ArgumentError without one). Public channels need none.

// The Pusher-compatible HTTP contract (recipes below):
channelAuth: UTDChannelAuth.endpoint(
  Uri.parse('https://api.myapp.com/utd/channel-auth'),
  headers: () async => {'Authorization': 'Bearer $mySessionJwt'},
),

// Full control (batching, caching, non-HTTP transports):
channelAuth: UTDChannelAuth.authorizer((request) async {
  final grant = await myBackend.authorize(request.socketId, request.channelName);
  return UTDChannelGrant(auth: grant.auth, channelData: grant.channelData);
}),

// Dev only — sends no grant; the server accepts it solely for projects with
// `settings.channels_insecure_dev` enabled:
channelAuth: const UTDChannelAuth.insecureDevelopment(),

With UTDChannelAuth.endpoint the SDK POSTs application/x-www-form-urlencoded fields socket_id and channel_name (exactly pusher-js's field names) and expects 200 {"auth": "<appKey>:<HMAC-SHA256 hex>", "channel_data"?: "<json string>"}. Grants are bound to the socket id, which changes on every reconnect — the SDK re-authorizes automatically when it resubscribes.

Channels & presence

The channel name prefix decides its type (Pusher rules):

Name	Type	Subscription auth	Client events (trigger)	Member list
news	public	none	no	no
private-orders-42	private	grant required	yes	no
presence-room-7	presence	grant + channel_data	yes	yes

Names are 1–164 characters from [a-zA-Z0-9_\-=@,.;]. subscribe(name) returns UTDChannel, UTDPrivateChannel, or UTDPresenceChannel by prefix (cast to reach the presence surface). Client events must be named client-…, may only be triggered on private/presence channels you are currently subscribed to, and payloads are capped at 10 KB — all enforced locally before any round trip, then again by the server.

Presence semantics

• Snapshot on subscribe: members and me are populated from the subscribe ack, so both are complete once whenSubscribed resolves.
• Deduped by member id: a user with two devices/tabs appears once; memberAdded fires on their first socket, memberRemoved when their last socket unsubscribes or disconnects.
• Member identity is signed: UTDMember.id/info come from the channel_data your auth endpoint signed ({"user_id": ..., "user_info"?: ...}), never from the client. Likewise UTDChannelEvent.userId on client events is server-attested from the sender's subscription.
• Caps: 100 members per presence channel (subscribing to a full channel fails with UTDErrorCode.channelFull), 1 KB of user_info per member.
• Reconnects: the member list is cleared and rebuilt from the fresh snapshot after every reconnect; grants are re-issued automatically because the socket id changed.

Subscription lifecycle

subscribe() is synchronous and idempotent — it returns the existing channel for a name you already hold. Each channel exposes subscriptionState (queued → subscribing → subscribed, or failed) and whenSubscribed, a future that is re-issued per (re)subscribe cycle. unsubscribe(name) tears the channel down and closes its streams.

Error handling

Every failure is a UTDStreamException carrying a stable machine code (the UTDErrorCode constants) — branch on codes, never on message text:

try {
  await room.trigger('client-message', {'text': 'hi'});
} on UTDRateLimitedException catch (e) {
  scheduleRetry(e.retryAfter ?? const Duration(seconds: 1));
} on UTDStreamException catch (e) {
  switch (e.code) {
    case UTDErrorCode.notSubscribed:   // subscription not live yet
    case UTDErrorCode.payloadTooLarge: // > 10 KB
    case UTDErrorCode.notConnected:    // socket down mid-flight
      showTransientError(e.message);
    default:
      rethrow;
  }
}

Typed subclasses when you need more than a code:

Exception	When	Extras
UTDSubscriptionException	channel auth rejected / malformed grant / bad signature	channelName, plus your endpoint's raw status + body
UTDRateLimitedException	server throttled the op	retryAfter, isTakeoverCooldown
UTDTokenException	mint returned an unusable token/url pair	—
UTDBannedException	the identity is banned	isBanned

Codes you will actually branch on: channel_auth_failed, channel_full, not_subscribed, payload_too_large, rate_limited, feature_disabled (channels not enabled on the project's plan — isFeatureDisabled / isForbidden helpers), not_connected, timeout, network. The full registry is in UTDErrorCode's dartdoc and the protocol spec.

Where errors surface:

• Requests (trigger, unsubscribe, connect, raw request) throw.
• Subscriptions: channel.whenSubscribed completes with the error; the same error is also emitted on client.errors (failed subscriptions retry automatically on the next reconnect).
• client.errors: post-connect server errors and fatal-close notices.
• client.forceExit: terminal ends — UTDForceExitReason.signedInElsewhere (single-active-session takeover, close code 4003) or reconnectTimeout (reconnect attempts exhausted). No auto-reconnect after these.

Close codes 4000–4099 are fatal by protocol convention and never auto-retried; everything else reconnects with capped exponential backoff + full jitter (tunable via ReconnectPolicy, dead connections detected via KeepAlivePolicy).

Migrating from Pusher

Pusher (pusher_channels_flutter / pusher-js)	utd_channels
Pusher.getInstance() + pusher.init(apiKey: ..., cluster: ...)	UTDChannelsClient(appId: ..., auth: ..., channelAuth: ...) — plain constructor, multi-instance, real dispose()
pusher.connect()	await client.connect()
pusher.subscribe(channelName: 'room')	client.subscribe('room') — synchronous; await channel.whenSubscribed when you need it live
channel.bind('event', (e) { ... }) / onEvent: callback	channel.bind('event') → Stream<UTDChannelEvent>; plus bindJson<T>('event', T.fromJson) and bindAll()
channel.trigger('client-x', data) (fire-and-forget)	await channel.trigger('client-x', data) — completes on server ack, throws typed exceptions
onAuthorizer: (channelName, socketId, options) async => {...}	channelAuth: UTDChannelAuth.endpoint(...) or UTDChannelAuth.authorizer((request) async => UTDChannelGrant(...))
pusher.getSocketId()	client.socketId
onConnectionStateChange: (current, previous) { ... }	client.connectionStates → Stream<UTDConnectionStateChange> (+ client.state)
channel.members / channel.me / onMemberAdded	room.members / room.me / room.memberAdded / room.memberRemoved streams
pusher.unsubscribe(channelName: 'room')	await client.unsubscribe('room')
server: pusher.trigger(channel, event, data)	POST /api/v1/channels/events with X-App-Id/X-App-Secret (single + batch, socket_id exclusion)

Deliberate deviations, all on purpose:

• No Pusher wire compatibility — the transport is a clean {type, id?, data?} + ack JSON protocol, not Pusher protocol 7 with its double-encoded data strings. What is byte-compatible is the channel-auth signature — the only part your backend touches.
• trigger() is ack-based, so rate limits and permission errors are catchable exceptions instead of silent drops.
• Streams instead of callback registries; no singleton; bindJson<T> decodes events into your own model classes.

Auth endpoint recipes

The grant is exactly Pusher's: auth = "<APP_KEY>:<signature>" where signature is HMAC-SHA256 (hex, key = your project secret) over socket_id:channel_name for private channels, or socket_id:channel_name:channel_data for presence channels (channel_data is the verbatim JSON string {"user_id": ..., "user_info"?: ...} you return alongside). Always authenticate the request with your own session mechanism before signing.

Node — reuse pusher-http-node unchanged

const Pusher = require("pusher");
const pusher = new Pusher({
  appId: "unused", cluster: "unused",         // only key + secret matter here
  key: process.env.UTD_APP_KEY, secret: process.env.UTD_PROJECT_SECRET,
});

app.post("/utd/channel-auth", (req, res) => {
  const { socket_id, channel_name } = req.body; // x-www-form-urlencoded
  if (!req.user) return res.sendStatus(403);    // your session check
  if (channel_name.startsWith("presence-")) {
    return res.json(pusher.authorizeChannel(socket_id, channel_name, {
      user_id: req.user.id, user_info: { name: req.user.name },
    }));
  }
  res.json(pusher.authorizeChannel(socket_id, channel_name));
});

Laravel — reuse Broadcasting

Configure the pusher broadcaster with your UTD key/secret (cluster can be anything), keep authorizing channels in routes/channels.php, and point the SDK at the stock endpoint:

channelAuth: UTDChannelAuth.endpoint(
  Uri.parse('https://api.myapp.com/broadcasting/auth'),
  headers: () async => {'Authorization': 'Bearer $token'},
),

Plain PHP

$socketId    = $_POST['socket_id'];
$channelName = $_POST['channel_name'];
if (str_starts_with($channelName, 'presence-')) {
    $channelData = json_encode(['user_id' => $user->id, 'user_info' => ['name' => $user->name]]);
    $sig = hash_hmac('sha256', "$socketId:$channelName:$channelData", $secret);
    echo json_encode(['auth' => "$appKey:$sig", 'channel_data' => $channelData]);
} else {
    $sig = hash_hmac('sha256', "$socketId:$channelName", $secret);
    echo json_encode(['auth' => "$appKey:$sig"]);
}

Python

import hashlib, hmac, json

def channel_auth(socket_id, channel_name, app_key, secret, user):
    if channel_name.startswith("presence-"):
        channel_data = json.dumps({"user_id": user.id, "user_info": {"name": user.name}})
        to_sign = f"{socket_id}:{channel_name}:{channel_data}"
        sig = hmac.new(secret.encode(), to_sign.encode(), hashlib.sha256).hexdigest()
        return {"auth": f"{app_key}:{sig}", "channel_data": channel_data}
    sig = hmac.new(secret.encode(), f"{socket_id}:{channel_name}".encode(), hashlib.sha256).hexdigest()
    return {"auth": f"{app_key}:{sig}"}

Wire protocol

The full client↔engine contract — frame envelope, channel.* ops, push frames, ack/close code registries, the channel-auth signature, and the server-side trigger REST API — is specified in doc/PROTOCOL.md. Useful if you self-host the engine, write an SDK in another language, or use the escape hatches (client.request(...) / client.frames) to speak raw protocol on the shared socket.