Publishing and reading

With MoQ, a participant publishes broadcasts at paths and reads the broadcasts others publish. The SDK hands out the official MoQ objects for this (@moq/net in TypeScript, moq-net in Rust), so what MoQ's documentation says of them holds here.

Paths and rights

A path is /-separated segments, at most 32, relative to the session's root. Which paths an app uses is its own choice; {id}/…, under the participant's own ID, is the usual form, and the token's publish patterns are what guarantee it. A session publishes where its publish patterns match and reads where its subscribe patterns match; a read outside them waits like a read of a path nobody publishes.

A segment that starts with . belongs to Tablebox's programs and system data, such as .aggregation. It is left out of listings unless the app asks for it, read by name like any other path, and covered by the rights as usual.

Publishing

publish(path) gives an official Broadcast.Producer, announced at path. A broadcast holds tracks by name; a track carries groups, and a group carries frames in order. createTrack(name) makes a track; writeString, writeJson and writeFrame write a frame as a group of its own, and appendGroup() starts a group of several frames. To end a broadcast, close its tracks, then the broadcast. After a reconnect the SDK publishes every broadcast again.

A JSON track keeps a value that changes: json.Snapshot.Producer (the official @moq/json) writes a snapshot, then RFC 7396 merge-patch deltas, and a reader that comes late gets the latest value first. json.Stream keeps a log of records instead.

TypeScript, Node · publish.ts

// Publishes {user}/status with a JSON track, status.json, that changes every second.
import { connect, json } from 'tablebox.io';
import { relay } from './relay.ts';
import { sign } from './token.ts';

const user = process.argv[2] ?? 'alice';
const connection = await connect({ ...relay, getToken: () => sign(user, { publish: [`${user}/**`] }) });
const broadcast = connection.publish(`${user}/status`);
const status = new json.Snapshot.Producer<{ online: boolean; count: number }>({ track: broadcast.createTrack('status.json') });
let count = 0;
setInterval(() => status.update({ online: true, count: count++ }), 1_000);

Reading and listing

read(path) waits until something publishes or serves path, then gives its Broadcast.Consumer. track(name).subscribe() gives a track's groups: recvGroup(), then readString(), readJson() or readFrame(). A subscription's maxAge (default 0) is how far back it reaches: 0 starts at the newest group. The broadcast ends when its publisher ends it or leaves; each of its tracks ends cleanly when its publisher closes it, and with an error when the connection drops. Then read again.

list(prefix) gives the paths published at or beneath prefix, now and as they come and go, within the session's read rights: each update is { kind, prefix }, kind being announced, updated or retracted. list(prefix, { hidden: true }) also gives paths with a . segment below prefix. Who publishes under a range is what listing it shows.

TypeScript, Node · read.ts

// Lists every {user}/status as it is published and prints its status.json as it changes.
import { connect, json } from 'tablebox.io';
import { relay } from './relay.ts';
import { sign } from './token.ts';

const connection = await connect({ ...relay, getToken: () => sign('reader', { subscribe: ['**'] }) });
for await (const { kind, prefix } of connection.list('')) {
  if (kind === 'retracted' || !prefix.endsWith('/status')) continue;
  void (async () => {
    const broadcast = await connection.read(prefix);
    const values = new json.Snapshot.Consumer({ track: broadcast.track('status.json').subscribe() });
    for await (const value of values) console.log(`${prefix}: ${JSON.stringify(value)}`);
  })();
}

In Rust, the same calls hand out moq-net's objects, and tablebox::json is moq-json:

Rust · read.rs

//! Reads alice/status and prints its status.json as it changes, with TABLEBOX_TOKEN.
use tablebox::{ConnectOptions, connect, json::snapshot::Consumer};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let env = |name| std::env::var(name).unwrap_or_default();
    let mut options = ConnectOptions::token(env("TABLEBOX_RELAY_URL"), env("TABLEBOX_TOKEN"));
    options.server_certificate_hashes = std::env::var("TABLEBOX_RELAY_CERT_SHA256").into_iter().collect();
    let connection = connect(options)?;
    let broadcast = connection.read("alice/status").await?;
    let track = broadcast.track("status.json")?.subscribe(None).await?;
    let mut values = Consumer::<serde_json::Value>::new(track, Default::default());
    while let Some(value) = values.next().await? {
        println!("alice/status: {value}");
    }
    Ok(())
}

Camera, microphone and screen

In browsers, the official publish and watch elements (publish and watch, from @moq/publish and @moq/watch) work on connection.origin. publish.Broadcast with publish.Source.Camera, Microphone or Screen publishes a hang broadcast (a catalog and a track per rendition), and watch.Player plays one into a canvas, as Get started shows. publish.Video.Encoder with config: { codec: 'avc1' } and publish.Audio.Encoder with codec: 'aac' choose H.264 and AAC, which Egress sends to streaming services as they are.

Previous: Connecting Next: Serving