Data — Cookbook - ClutchCall

This page has small, task-focused snippets for the data modality. Each snippet assumes a constructed client:

import { Data } from "@clutchcall/sdk/data";

const data = new Data({
  token:    process.env.CLUTCHCALL_CREDENTIALS!,
  clientId: "device-7",
});

Publish a reading (lossy, default)

The default lane is the QUIC datagram lane. It has the lowest latency. Use it for high-rate telemetry where the freshest value wins.

await data.publish({
  topic:   "sensors/room1/temperature",
  payload: new TextEncoder().encode("23.5"),
});

Publish an event that must arrive (reliable lane)

Set reliable: true for application events, commands, or settlements. This lane is ordered and retried per publisher.

await data.publish({
  topic:    "events/orders/shipped",
  payload:  new TextEncoder().encode(JSON.stringify({ orderId: "A-91" })),
  reliable: true,
});

+ matches exactly one path segment. Here, it matches any room.

const sub = await data.subscribe(
  { topicFilter: "sensors/+/temperature" },
  (msg) => console.log(msg.topic, new TextDecoder().decode(msg.payload)),
);

# matches the rest of the path and must be the trailing segment. One subscription covers everything under events/.

await data.subscribe({ topicFilter: "events/#" }, (msg) => {
  console.log("event:", msg.topic, "from", msg.fromClientId);
});

Publish retained current state

The relay caches a retained message and delivers it to every future subscriber on attach. Use it for “current state.”

await data.publish({
  topic:    "devices/device-7/state",
  payload:  new TextEncoder().encode(JSON.stringify({ online: true, version: "1.4.2" })),
  retained: true,
});

Clear a retained topic

Publish a zero-length payload with retained: true. Do this typically on clean shutdown. Dashboards then show “offline” without a heartbeat timeout.

await data.publish({
  topic:    "devices/device-7/state",
  payload:  new Uint8Array(0),
  retained: true,
});

Tell a bootstrap value from a live one

Retained values that the relay re-emits on attach arrive with msg.retained === true. Render the snapshot. Then track live updates.

await data.subscribe({ topicFilter: "devices/+/state" }, (msg) => {
  if (msg.retained) applySnapshot(msg);   // cached current state
  else               applyUpdate(msg);    // live change
});

Fan in from a whole fleet to one dashboard

Every message carries fromClientId. One subscription thus demuxes many publishers without a track per device.

const latest = new Map<string, number>();
await data.subscribe({ topicFilter: "sensors/+/temperature" }, (msg) => {
  latest.set(msg.fromClientId, Number(new TextDecoder().decode(msg.payload)));
});

Publish and consume JSON

Payloads are opaque bytes. Encode and decode JSON yourself.

const enc = (o: unknown) => new TextEncoder().encode(JSON.stringify(o));
const dec = (b: Uint8Array) => JSON.parse(new TextDecoder().decode(b));

await data.publish({ topic: "events/users/signup", payload: enc({ userId: "u-7" }) });

await data.subscribe({ topicFilter: "events/users/#" }, (msg) => {
  const body = dec(msg.payload);
  console.log("signup", body.userId);
});

Route by topic with a dispatch table

Use one subscription on a subtree and many handlers keyed by sub-filter. This is the classic event-bus shape.

import { topicMatches } from "@clutchcall/sdk/data";

const routes: [string, (m: any) => void][] = [
  ["events/orders/shipped", (m) => onShipped(m)],
  ["events/users/signup",   (m) => onSignup(m)],
];

await data.subscribe({ topicFilter: "events/#" }, (msg) => {
  for (const [pattern, fn] of routes)
    if (topicMatches(msg.topic, pattern)) return fn(msg);
});

Stop a single subscription

const sub = await data.subscribe({ topicFilter: "sensors/#" }, handle);
// later…
sub.close();

Watch connection state

Pass onState to react to reconnects. The SDK replays subscriptions and retained values automatically after a drop.

const data = new Data({
  token, clientId: "device-7",
  onState: (state, reason) => console.log("data session:", state, reason ?? ""),
});

Publish from Python

The Python surface mirrors the TS one with snake_case keywords.

from clutchcall.data import Data

data = Data(token=token, client_id="device-7")
data.publish(topic="sensors/room1/temperature", payload=b"23.5")
sub = data.subscribe(
    topic_filter="sensors/+/temperature",
    on_message=lambda m: print(m.topic, m.from_client_id, m.payload),
)

Stay wire-compatible without the client

Use the exported codec to build or parse frames yourself.

import { encodeDataFrame, decodeDataFrame } from "@clutchcall/sdk/data";

const frame   = encodeDataFrame("device-7", "sensors/room1/temp", payload);
const decoded = decodeDataFrame(frame);   // { fromClientId, topic, payload }

Keep the top-level topic segment concrete and meaningful (sensors, events, devices). It selects the underlying track. It is thus also your coarse fan-out and routing boundary.

​Publish a reading (lossy, default)

​Publish an event that must arrive (reliable lane)

​Subscribe with a + (single-segment) wildcard

​Subscribe to a whole subtree with #

​Publish retained current state

​Clear a retained topic

​Tell a bootstrap value from a live one

​Fan in from a whole fleet to one dashboard

​Publish and consume JSON

​Route by topic with a dispatch table

​Stop a single subscription

​Watch connection state

​Publish from Python

​Stay wire-compatible without the client

Publish a reading (lossy, default)

Publish an event that must arrive (reliable lane)

Subscribe with a `+` (single-segment) wildcard

Subscribe to a whole subtree with `#`

Publish retained current state

Clear a retained topic

Tell a bootstrap value from a live one

Fan in from a whole fleet to one dashboard

Publish and consume JSON

Route by topic with a dispatch table

Stop a single subscription

Watch connection state

Publish from Python

Stay wire-compatible without the client