From e606486cc543c31cceb239d0aeaf5bbf9c4c8385 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 23:49:03 +0000 Subject: [PATCH 1/7] feat(hang/publish/watch): publish & subscribe arbitrary tracks in a broadcast Add a backwards-compatible way to carry arbitrary application tracks (e.g. a meta.json metadata track) within an existing broadcast, alongside audio/video, without spinning up a separate broadcast. Catalog: new optional `data` section, a flat map of track name to a small descriptor (mime, description). Omitted from the wire when empty so existing catalogs are byte-identical. Mirrored in js/hang and rs/hang. Publish: `Broadcast.publishJson(name, opts)` serves a fan-out JSON track (seeding late joiners with the latest value) and adds the catalog `data` entry; `Broadcast.publishTrack(name, serve)` is the low-level escape hatch for arbitrary payloads. The unknown-track branch now consults this registry instead of always rejecting. Extracted a reusable `JsonProducer` that `CatalogProducer` reuses. Watch: `Broadcast.subscribeJson(name, opts)` returns a `JsonConsumer` whose latest value is a signal, following the active broadcast across reconnects. Both web components already expose `.broadcast`, so this is reachable as `el.broadcast.publishJson(...)` / `el.broadcast.subscribeJson(...)`. Docs: concept/layer/hang plus the @moq/publish and @moq/watch lib pages. --- doc/concept/layer/hang.md | 14 ++++++ doc/lib/js/@moq/publish.md | 16 +++++++ doc/lib/js/@moq/watch.md | 21 +++++++++ js/hang/src/catalog/data.ts | 31 ++++++++++++++ js/hang/src/catalog/index.ts | 1 + js/hang/src/catalog/root.test.ts | 7 +++ js/hang/src/catalog/root.ts | 5 +++ js/publish/src/broadcast.ts | 68 +++++++++++++++++++++++++++++- js/publish/src/catalog.ts | 35 ++++----------- js/publish/src/index.ts | 1 + js/publish/src/json.test.ts | 46 ++++++++++++++++++++ js/publish/src/json.ts | 68 ++++++++++++++++++++++++++++++ js/watch/src/broadcast.ts | 20 +++++++++ js/watch/src/index.ts | 1 + js/watch/src/json.test.ts | 60 ++++++++++++++++++++++++++ js/watch/src/json.ts | 61 +++++++++++++++++++++++++++ rs/hang/src/catalog/data.rs | 26 ++++++++++++ rs/hang/src/catalog/mod.rs | 2 + rs/hang/src/catalog/root.rs | 55 ++++++++++++++++++++++-- rs/moq-mux/src/catalog/hang/ext.rs | 1 + rs/moq-mux/src/catalog/producer.rs | 2 + 21 files changed, 509 insertions(+), 32 deletions(-) create mode 100644 js/hang/src/catalog/data.ts create mode 100644 js/publish/src/json.test.ts create mode 100644 js/publish/src/json.ts create mode 100644 js/watch/src/json.test.ts create mode 100644 js/watch/src/json.ts create mode 100644 rs/hang/src/catalog/data.rs diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index 0753b749b6..7c928ea3ec 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -83,6 +83,20 @@ The catalog is a JSON document published through the merge-patch helper (`@moq/j This keeps application-specific sections in the application layer while the base catalog stays generic. +### Data tracks + +Beyond `audio`/`video`, the catalog has an optional `data` section: a flat map of track name to a small descriptor (`mime`, `description`). It advertises arbitrary application-defined tracks (for example a `meta.json` metadata track) carried within the same broadcast, so a consumer can discover and subscribe to them without a separate broadcast. The catalog says nothing about how to decode a data track; the descriptor fields are hints for the consumer. + +```json +{ + "data": { + "meta.json": { "mime": "application/json", "description": "broadcast metadata" } + } +} +``` + +The web components expose this directly: on the publish side, `broadcast.publishJson("meta.json")` serves a JSON track (seeding late joiners with the latest value) and adds the `data` entry; on the watch side, `broadcast.subscribeJson("meta.json")` follows the active broadcast and exposes the latest value as a signal. The section is omitted from the wire when empty, so existing catalogs are unaffected. + ## Container The catalog also contains a `container` field for each rendition used to denote the encoding of each track. diff --git a/doc/lib/js/@moq/publish.md b/doc/lib/js/@moq/publish.md index babdbc2762..b92da56f2f 100644 --- a/doc/lib/js/@moq/publish.md +++ b/doc/lib/js/@moq/publish.md @@ -127,6 +127,22 @@ broadcast.video.device.set("screen"); broadcast.name.set("bob.hang"); ``` +### Custom data tracks + +Beyond audio and video, you can publish arbitrary application tracks within the +same broadcast (no separate broadcast needed). `publishJson` serves a JSON track +(seeding late joiners with the latest value) and advertises it in the catalog +[`data` section](/concept/layer/hang#data-tracks): + +```typescript +const meta = broadcast.publishJson<{ title: string }>("meta.json"); +meta.update({ title: "Live from the moon" }); +``` + +For non-JSON payloads, `publishTrack(name, serve)` is the low-level escape hatch: +`serve` runs whenever a subscriber requests the named track. The component +exposes both via its `broadcast` property (`el.broadcast.publishJson(...)`). + ## Related Packages - **[@moq/watch](/lib/js/@moq/watch)** — Subscribe to and render MoQ broadcasts diff --git a/doc/lib/js/@moq/watch.md b/doc/lib/js/@moq/watch.md index 89f0c57612..cab4af13f3 100644 --- a/doc/lib/js/@moq/watch.md +++ b/doc/lib/js/@moq/watch.md @@ -139,6 +139,27 @@ el.catalog = myCatalog; > `"msf"`) tears down the previous fetch loop, which clears `catalog`. Set the > catalog *after* switching to `"manual"`, not before. +### Custom data tracks + +A broadcast can carry arbitrary application tracks (for example a `meta.json` +metadata track) alongside the media, listed in the catalog +[`data` section](/concept/layer/hang#data-tracks). Use `subscribeJson` to read +one: it follows the active broadcast across reconnects and exposes the latest +value as a signal. Call `close()` when done (it's also closed when the broadcast +closes). + +```typescript +const meta = broadcast.subscribeJson<{ title: string }>("meta.json"); +meta.value.subscribe((value) => console.log("meta", value)); +``` + +The component exposes the same method via its `broadcast` property: + +```typescript +const el = document.querySelector("moq-watch")!; +const meta = el.broadcast.subscribeJson("meta.json"); +``` + ## UI Overlay Import `@moq/watch/ui` for a Web Component overlay with buffering indicator, stats panel, and playback controls: diff --git a/js/hang/src/catalog/data.ts b/js/hang/src/catalog/data.ts new file mode 100644 index 0000000000..b66afc3bcf --- /dev/null +++ b/js/hang/src/catalog/data.ts @@ -0,0 +1,31 @@ +import * as z from "zod/mini"; + +/** + * A custom, application-defined data track in the catalog. + * + * Unlike `audio`/`video`, the catalog says nothing about how to decode a data track. It just + * advertises that the track exists (keyed by name in {@link DataSchema}) so a consumer can discover + * and subscribe to it. The optional fields are hints for the consumer, not instructions for hang. + */ +export const DataTrackSchema = z.object({ + /** The MIME type of each frame's payload, e.g. `"application/json"`. Informational. */ + mime: z.optional(z.string()), + + /** A free-form description of the track's contents, for humans. */ + description: z.optional(z.string()), +}); + +/** A custom, application-defined data track in the catalog. */ +export type DataTrack = z.infer; + +/** + * The `data` catalog section: a map of track name to {@link DataTrack}. + * + * Each key is the name of a track within the broadcast (e.g. `"meta.json"`). The tracks are + * independent of each other, so this is a flat map rather than the rendition groups used by + * `audio`/`video`. + */ +export const DataSchema = z.record(z.string(), DataTrackSchema); + +/** The `data` catalog section: a map of track name to {@link DataTrack}. */ +export type Data = z.infer; diff --git a/js/hang/src/catalog/index.ts b/js/hang/src/catalog/index.ts index 6a0c787b54..a4429c8ab6 100644 --- a/js/hang/src/catalog/index.ts +++ b/js/hang/src/catalog/index.ts @@ -7,6 +7,7 @@ export * from "./audio"; export * from "./container"; +export * from "./data"; export * from "./format"; export * from "./integers"; export * from "./priority"; diff --git a/js/hang/src/catalog/root.test.ts b/js/hang/src/catalog/root.test.ts index 5e2a2f3bc2..80bce585dc 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -12,6 +12,13 @@ test("base catalog preserves unknown sections", () => { expect(parsed.scte35).toEqual({ spliceId: 42 }); }); +test("parses the data section of custom tracks", () => { + const parsed = RootSchema.parse({ + data: { "meta.json": { mime: "application/json", description: "broadcast metadata" } }, + }); + expect(parsed.data?.["meta.json"]).toEqual({ mime: "application/json", description: "broadcast metadata" }); +}); + test("extended schema validates app sections", () => { const Scte35Schema = z.object({ spliceId: z.number() }); const ExtendedSchema = z.extend(RootSchema, { scte35: z.optional(Scte35Schema) }); diff --git a/js/hang/src/catalog/root.ts b/js/hang/src/catalog/root.ts index 9f7be7f121..72ea18492b 100644 --- a/js/hang/src/catalog/root.ts +++ b/js/hang/src/catalog/root.ts @@ -1,6 +1,7 @@ import * as z from "zod/mini"; import { AudioSchema } from "./audio"; +import { DataSchema } from "./data"; import { VideoSchema } from "./video"; /** @@ -10,10 +11,14 @@ import { VideoSchema } from "./video"; * application can add its own sections (e.g. `scte35`) without modifying hang. A base consumer * ignores the extra sections; an extended consumer validates them with its own schema, typically * built via `z.extend(RootSchema, { ... })`. + * + * The `data` section lists arbitrary application-defined tracks (e.g. a `meta.json` track) carried + * alongside the media within the same broadcast. */ export const RootSchema = z.looseObject({ video: z.optional(VideoSchema), audio: z.optional(AudioSchema), + data: z.optional(DataSchema), }); /** The root catalog object, with optional video and audio sections plus any app extensions. */ diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index ae2638d578..0edccb3b4b 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -1,8 +1,10 @@ import * as Catalog from "@moq/hang/catalog"; +import type * as Json from "@moq/json"; import * as Moq from "@moq/net"; import { Effect, Signal } from "@moq/signals"; import * as Audio from "./audio"; import { CatalogProducer } from "./catalog"; +import { JsonProducer } from "./json"; import * as Video from "./video"; export type BroadcastProps = { @@ -13,6 +15,17 @@ export type BroadcastProps = { video?: Video.Props; }; +/** Serves a custom track when a subscriber requests it, scoped to the subscription's lifetime. */ +export type ServeTrack = (track: Moq.Track, effect: Effect) => void; + +/** Options for {@link Broadcast.publishJson}, extending the {@link Json.Config} for the value. */ +export type PublishJsonOptions = Json.Config & { + /** MIME type recorded in the catalog `data` entry. Defaults to `"application/json"`. */ + mime?: string; + /** Human-readable description recorded in the catalog `data` entry. */ + description?: string; +}; + export class Broadcast { static readonly CATALOG_TRACK = "catalog.json"; @@ -28,6 +41,10 @@ export class Broadcast { // sections (e.g. `scte35`) by locking it too. readonly catalog = new CatalogProducer(); + // Handlers for custom tracks registered via `publishTrack`, keyed by track name. Persists across + // reconnects so a new `Moq.Broadcast` still serves them. + #tracks = new Map(); + signals = new Effect(); constructor(props?: BroadcastProps) { @@ -100,15 +117,64 @@ export class Broadcast { case Video.Root.TRACK_SD: this.video.sd.serve(request.track, effect); break; - default: + default: { + const serve = this.#tracks.get(request.track.name); + if (serve) { + serve(request.track, effect); + break; + } console.error("received subscription for unknown track", request.track.name); request.track.close(new Error(`Unknown track: ${request.track.name}`)); break; + } } }); } } + /** + * Serve a custom track within this broadcast, identified by name. + * + * When a subscriber requests a track with this name, `serve` runs with the track and an effect + * scoped to that subscription (cleaned up when the subscriber goes away). The handler persists + * across reconnects. This is the low-level escape hatch for arbitrary payloads; see + * {@link publishJson} for a JSON track that also advertises itself in the catalog. + * + * Returns a function that unregisters the handler. Note this does not close already-served + * subscriptions, nor touch the catalog. + */ + publishTrack(name: string, serve: ServeTrack): () => void { + this.#tracks.set(name, serve); + return () => { + if (this.#tracks.get(name) === serve) this.#tracks.delete(name); + }; + } + + /** + * Publish a custom JSON track within this broadcast and advertise it in the catalog `data` + * section. + * + * Returns a {@link JsonProducer}: call `update`/`mutate` to set the value, which is served to + * every subscriber (seeding late joiners with the latest value). The track lives for the + * lifetime of the broadcast. + * + * ```ts + * const meta = broadcast.publishJson("meta.json"); + * meta.update({ title: "Live from the moon" }); + * ``` + */ + publishJson(name: string, options?: PublishJsonOptions): JsonProducer { + const producer = new JsonProducer(options); + this.publishTrack(name, (track, effect) => producer.serve(track, effect)); + + this.catalog.mutate((catalog) => { + catalog.data ??= {}; + catalog.data[name] = { mime: options?.mime ?? "application/json", description: options?.description }; + }); + + return producer; + } + close() { this.signals.close(); this.audio.close(); diff --git a/js/publish/src/catalog.ts b/js/publish/src/catalog.ts index 7cad29cb90..7bce178f1c 100644 --- a/js/publish/src/catalog.ts +++ b/js/publish/src/catalog.ts @@ -1,38 +1,19 @@ import type * as Catalog from "@moq/hang/catalog"; -import * as Json from "@moq/json"; -import type * as Moq from "@moq/net"; -import type { Effect } from "@moq/signals"; + +import { JsonProducer } from "./json"; /** * A stable catalog producer that fans out to on-demand subscription tracks. * * Unlike a raw track producer, this exists independently of any subscription: edit it at any time - * with {@link mutate}, and each subscriber (including a relay that reconnects) is seeded with the - * current catalog before receiving updates. Independent owners (the base `video`/`audio` and an + * with `mutate`, and each subscriber (including a relay that reconnects) is seeded with the current + * catalog before receiving updates. Independent owners (the base `video`/`audio` and an * application's own sections, e.g. `scte35`) each edit only their own keys, so their sections * compose instead of clobbering one another. */ -export class CatalogProducer { - #value: Catalog.Root = {}; - #outputs = new Set>(); - - /** Edit the catalog in place; the result is published to all current subscribers. */ - mutate(fn: (catalog: Catalog.Root) => void): void { - const value = structuredClone(this.#value); - fn(value); - this.#value = value; - for (const output of this.#outputs) output.update(value); - } - - /** Serve a subscription request: seed it with the current catalog, then forward updates. */ - serve(track: Moq.Track, effect: Effect): void { - const output = new Json.Producer(track); - output.update(this.#value); - - this.#outputs.add(output); - effect.cleanup(() => { - this.#outputs.delete(output); - output.finish(); - }); +export class CatalogProducer extends JsonProducer { + /** Create a catalog producer seeded with an empty catalog. */ + constructor() { + super({ initial: {} }); } } diff --git a/js/publish/src/index.ts b/js/publish/src/index.ts index 082e2f005f..f81cec91f7 100644 --- a/js/publish/src/index.ts +++ b/js/publish/src/index.ts @@ -6,6 +6,7 @@ export * as Signals from "@moq/signals"; export * as Audio from "./audio"; export * from "./broadcast"; export * from "./catalog"; +export * from "./json"; export * as Preview from "./preview"; export * as Source from "./source"; export * as Video from "./video"; diff --git a/js/publish/src/json.test.ts b/js/publish/src/json.test.ts new file mode 100644 index 0000000000..45ef46119c --- /dev/null +++ b/js/publish/src/json.test.ts @@ -0,0 +1,46 @@ +import { expect, test } from "bun:test"; +import * as Json from "@moq/json"; +import { Track } from "@moq/net"; +import { Effect } from "@moq/signals"; +import { JsonProducer } from "./json.ts"; + +test("seeds late subscribers with the current value and fans out updates", async () => { + const producer = new JsonProducer<{ title?: string; count?: number }>(); + + // Set a value before anyone subscribes: it is retained, not lost. + producer.update({ title: "hello" }); + expect(producer.value).toEqual({ title: "hello" }); + + const effect = new Effect(); + const track = new Track("meta.json"); + producer.serve(track, effect); + const consumer = new Json.Consumer<{ title?: string; count?: number }>(track); + + // A new subscriber is seeded with the current value. + expect(await consumer.next()).toEqual({ title: "hello" }); + + // Subsequent updates fan out to the subscriber. + producer.update({ title: "world" }); + expect(await consumer.next()).toEqual({ title: "world" }); + + effect.close(); +}); + +test("mutate composes from the last value", async () => { + const producer = new JsonProducer>({ initial: {} }); + + // mutate works before any update because of the configured initial value. + producer.mutate((v) => { + v.a = 1; + }); + producer.mutate((v) => { + v.b = 2; + }); + + expect(producer.value).toEqual({ a: 1, b: 2 }); +}); + +test("mutate without a value or initial throws", () => { + const producer = new JsonProducer>(); + expect(() => producer.mutate(() => {})).toThrow(); +}); diff --git a/js/publish/src/json.ts b/js/publish/src/json.ts new file mode 100644 index 0000000000..1bcd5f0fd9 --- /dev/null +++ b/js/publish/src/json.ts @@ -0,0 +1,68 @@ +import * as Json from "@moq/json"; +import type * as Moq from "@moq/net"; +import type { Effect } from "@moq/signals"; + +/** + * A stable JSON value that fans out to on-demand subscription tracks. + * + * Unlike a raw {@link Json.Producer}, this exists independently of any subscription: set the value + * at any time with {@link update} or {@link mutate}, and each subscriber (including a relay that + * reconnects) is seeded with the current value before receiving updates. Multiple independent owners + * can share one instance and each edit only their own keys via {@link mutate}, so their sections + * compose instead of clobbering one another. + * + * This backs both the broadcast catalog and arbitrary custom JSON tracks (see + * `Broadcast.publishJson`). + */ +export class JsonProducer { + #value: T | undefined; + #outputs = new Set>(); + #config: Json.Config; + + /** Create a producer, optionally seeding an initial value and per-track config. */ + constructor(config: Json.Config = {}) { + this.#config = config; + this.#value = config.initial; + } + + /** The current value, or `undefined` if nothing has been published yet. */ + get value(): T | undefined { + return this.#value; + } + + /** Replace the value; the result is published to all current subscribers. No-op if unchanged. */ + update(value: T): void { + this.#value = value; + for (const output of this.#outputs) output.update(value); + } + + /** + * Mutate the current value in place and publish the result. + * + * The callback receives a deep clone of the last value (falling back to the configured `initial`, + * throwing if neither exists). Edit it in place; on return the result is published via + * {@link update}. + */ + mutate(fn: (value: T) => void): void { + const base = this.#value ?? this.#config.initial; + if (base === undefined) { + throw new Error("mutate() requires a prior update() or `initial` in the config"); + } + + const value = structuredClone(base) as T; + fn(value); + this.update(value); + } + + /** Serve a subscription request: seed it with the current value, then forward updates. */ + serve(track: Moq.Track, effect: Effect): void { + const output = new Json.Producer(track, this.#config); + if (this.#value !== undefined) output.update(this.#value); + + this.#outputs.add(output); + effect.cleanup(() => { + this.#outputs.delete(output); + output.finish(); + }); + } +} diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 2b05cc6f4a..bae001a9b0 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -5,6 +5,7 @@ import type * as Moq from "@moq/net"; import { Path } from "@moq/net"; import { Effect, type Getter, Signal } from "@moq/signals"; +import { JsonConsumer, type SubscribeJsonOptions } from "./json"; import { toHang } from "./msf"; // Watch supports the two on-the-wire catalog formats from @moq/hang plus a @@ -190,6 +191,25 @@ export class Broadcast { }); } + /** + * Subscribe to a custom JSON track within this broadcast (e.g. a `meta.json` track listed in the + * catalog `data` section). + * + * Follows the active broadcast across reconnects and exposes the latest reconstructed value as a + * Signal on the returned consumer. The consumer is closed when this broadcast closes; close it + * sooner via its `close()` to stop subscribing. + * + * ```ts + * const meta = broadcast.subscribeJson<{ title: string }>("meta.json"); + * meta.value.subscribe((v) => console.log("meta", v)); + * ``` + */ + subscribeJson(name: string, options?: SubscribeJsonOptions): JsonConsumer { + const consumer = new JsonConsumer(this.active, name, options); + this.signals.cleanup(() => consumer.close()); + return consumer; + } + close() { this.signals.close(); } diff --git a/js/watch/src/index.ts b/js/watch/src/index.ts index 6103f721f7..2fa90e3b19 100644 --- a/js/watch/src/index.ts +++ b/js/watch/src/index.ts @@ -6,6 +6,7 @@ export * as Signals from "@moq/signals"; export * as Audio from "./audio"; export * from "./backend"; export * from "./broadcast"; +export * from "./json"; export * as Mse from "./mse"; export * from "./sync"; export * as Video from "./video"; diff --git a/js/watch/src/json.test.ts b/js/watch/src/json.test.ts new file mode 100644 index 0000000000..ce064c5d91 --- /dev/null +++ b/js/watch/src/json.test.ts @@ -0,0 +1,60 @@ +import { expect, test } from "bun:test"; +import * as Json from "@moq/json"; +import { Broadcast as MoqBroadcast } from "@moq/net"; +import { type Getter, Signal } from "@moq/signals"; +import { JsonConsumer } from "./json.ts"; + +// Resolve once the signal holds a defined value. +function nextDefined(signal: Getter): Promise { + return new Promise((resolve) => { + const dispose = signal.subscribe((value) => { + if (value === undefined) return; + dispose(); + resolve(value); + }); + }); +} + +test("reads a custom JSON track from the active broadcast", async () => { + const broadcast = new MoqBroadcast(); + const active = new Signal(broadcast); + + const consumer = new JsonConsumer<{ title: string }>(active, "meta.json"); + + // The consumer subscribes; serve the request from the producer side. + const request = await broadcast.requested(); + if (!request) throw new Error("expected a track request"); + expect(request.track.name).toBe("meta.json"); + const producer = new Json.Producer<{ title: string }>(request.track); + producer.update({ title: "hello" }); + + expect(await nextDefined(consumer.value)).toEqual({ title: "hello" }); + + consumer.close(); +}); + +test("clears the value when the broadcast goes away", async () => { + const broadcast = new MoqBroadcast(); + const active = new Signal(broadcast); + + const consumer = new JsonConsumer<{ title: string }>(active, "meta.json"); + + const request = await broadcast.requested(); + if (!request) throw new Error("expected a track request"); + const producer = new Json.Producer<{ title: string }>(request.track); + producer.update({ title: "hello" }); + await nextDefined(consumer.value); + + // Dropping the active broadcast tears down the subscription and clears the value. + await new Promise((resolve) => { + const dispose = consumer.value.subscribe((value) => { + if (value !== undefined) return; + dispose(); + resolve(); + }); + active.set(undefined); + }); + expect(consumer.value.peek()).toBeUndefined(); + + consumer.close(); +}); diff --git a/js/watch/src/json.ts b/js/watch/src/json.ts new file mode 100644 index 0000000000..b4b0cbc2d2 --- /dev/null +++ b/js/watch/src/json.ts @@ -0,0 +1,61 @@ +import * as Catalog from "@moq/hang/catalog"; +import * as Json from "@moq/json"; +import type * as Moq from "@moq/net"; +import { Effect, type Getter, Signal } from "@moq/signals"; + +/** Options for {@link Broadcast.subscribeJson}, extending the {@link Json.Config} for the value. */ +export type SubscribeJsonOptions = Json.Config & { + /** + * Subscription priority. Defaults to {@link Catalog.PRIORITY.catalog} so metadata arrives ahead + * of media. + */ + priority?: number; +}; + +/** + * Consumes a custom JSON track from a broadcast, following the active broadcast across reconnects. + * + * The latest reconstructed value is exposed as a {@link Signal} via {@link value}. Call + * {@link close} when done (or let the owning watch {@link Broadcast} close it). + */ +export class JsonConsumer { + /** The latest reconstructed value, or `undefined` before the first frame or while offline. */ + readonly value = new Signal(undefined); + + #signals = new Effect(); + + /** Subscribe to `name` on whatever broadcast `active` currently holds. */ + constructor(active: Getter, name: string, options?: SubscribeJsonOptions) { + const priority = options?.priority ?? Catalog.PRIORITY.catalog; + + this.#signals.run((effect) => { + const broadcast = effect.get(active); + if (!broadcast) return; + + const track = broadcast.subscribe(name, priority); + effect.cleanup(() => track.close()); + + // Clear the value when this subscription tears down so a stale value from a previous + // broadcast doesn't linger after a reconnect. + effect.cleanup(() => this.value.set(undefined)); + + const consumer = new Json.Consumer(track, options); + effect.spawn(async () => { + try { + for (;;) { + const next = await Promise.race([effect.cancel, consumer.next()]); + if (next === undefined) break; + this.value.set(next); + } + } catch (err) { + console.warn("error fetching json track", name, err); + } + }); + }); + } + + /** Stop subscribing and release resources. */ + close(): void { + this.#signals.close(); + } +} diff --git a/rs/hang/src/catalog/data.rs b/rs/hang/src/catalog/data.rs new file mode 100644 index 0000000000..f72b790eeb --- /dev/null +++ b/rs/hang/src/catalog/data.rs @@ -0,0 +1,26 @@ +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; + +/// A custom, application-defined data track in the catalog. +/// +/// Unlike `audio`/`video`, the catalog says nothing about how to decode a data track. It just +/// advertises that the track exists (keyed by name in the [`Data`] map) so a consumer can discover +/// and subscribe to it. The optional fields are hints for the consumer, not instructions for hang. +#[serde_with::skip_serializing_none] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(rename_all = "camelCase")] +pub struct DataTrack { + /// The MIME type of each frame's payload, e.g. `"application/json"`. Informational. + pub mime: Option, + + /// A free-form description of the track's contents, for humans. + pub description: Option, +} + +/// The `data` catalog section: a map of track name to [`DataTrack`]. +/// +/// Each key is the name of a track within the broadcast (e.g. `"meta.json"`). The tracks are +/// independent of each other, so this is a flat map rather than the rendition groups used by +/// `audio`/`video`. +pub type Data = BTreeMap; diff --git a/rs/hang/src/catalog/mod.rs b/rs/hang/src/catalog/mod.rs index bab4f5ffeb..def874e573 100644 --- a/rs/hang/src/catalog/mod.rs +++ b/rs/hang/src/catalog/mod.rs @@ -6,10 +6,12 @@ mod audio; mod container; +mod data; mod root; mod video; pub use audio::*; pub use container::*; +pub use data::*; pub use root::*; pub use video::*; diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index 9c68c5c7a8..4162cfccb0 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -1,12 +1,13 @@ //! This module contains the structs and functions for the MoQ catalog format use crate::Result; -use crate::catalog::{Audio, Video}; +use crate::catalog::{Audio, Data, Video}; use serde::{Deserialize, Serialize}; /// A catalog track, created by a broadcaster to describe the tracks available in a broadcast. /// -/// The base catalog carries only the media sections (`video`, `audio`). Applications extend it with -/// their own root sections (e.g. `scte35`) by flattening this struct into their own with +/// The base catalog carries the media sections (`video`, `audio`) plus an optional `data` section +/// listing arbitrary application-defined tracks (e.g. a `meta.json` track). Applications extend it +/// with their own root sections (e.g. `scte35`) by flattening this struct into their own with /// `#[serde(flatten)]`. The catalog does not deny unknown fields, so a base consumer ignores the /// extra sections and an extended catalog stays wire-compatible. See the `extension_roundtrip` test. #[serde_with::serde_as] @@ -27,6 +28,11 @@ pub struct Catalog { /// based on their preferences (codec, bitrate, language, etc). #[serde(default)] pub audio: Audio, + + /// Custom application-defined data tracks (e.g. a `meta.json` track) carried within the + /// broadcast. Omitted from the wire when empty, so existing catalogs are unaffected. + #[serde(default, skip_serializing_if = "Data::is_empty")] + pub data: Data, } impl Catalog { @@ -81,7 +87,7 @@ impl Catalog { mod test { use std::collections::BTreeMap; - use crate::catalog::{AudioCodec::Opus, AudioConfig, Container, H264, VideoConfig}; + use crate::catalog::{AudioCodec::Opus, AudioConfig, Container, Data, DataTrack, H264, VideoConfig}; use super::*; @@ -148,6 +154,7 @@ mod test { audio: Audio { renditions: audio_renditions, }, + data: Data::default(), }; let output = Catalog::from_str(&encoded).expect("failed to decode"); @@ -157,6 +164,46 @@ mod test { assert_eq!(encoded, output, "wrong encoded output"); } + #[test] + fn data_roundtrip() { + let mut encoded = r#"{ + "video": {"renditions": {}}, + "audio": {"renditions": {}}, + "data": { + "meta.json": {"mime": "application/json", "description": "metadata"} + } + }"# + .to_string(); + encoded.retain(|c| !c.is_whitespace()); + + let mut data = Data::default(); + data.insert( + "meta.json".to_string(), + DataTrack { + mime: Some("application/json".to_string()), + description: Some("metadata".to_string()), + }, + ); + + let decoded = Catalog { + data, + ..Default::default() + }; + + assert_eq!(decoded, Catalog::from_str(&encoded).expect("failed to decode")); + assert_eq!(encoded, decoded.to_string().expect("failed to encode")); + } + + #[test] + fn data_omitted_when_empty() { + // An empty data section must not appear on the wire, so existing catalogs are unaffected. + let encoded = Catalog::default().to_string().expect("failed to encode"); + assert!( + !encoded.contains("data"), + "empty data section should be omitted: {encoded}" + ); + } + #[test] fn extension_roundtrip() { // An application extends the catalog with its own root section by flattening Catalog. diff --git a/rs/moq-mux/src/catalog/hang/ext.rs b/rs/moq-mux/src/catalog/hang/ext.rs index 3de6a3c0f8..1c616a430c 100644 --- a/rs/moq-mux/src/catalog/hang/ext.rs +++ b/rs/moq-mux/src/catalog/hang/ext.rs @@ -57,6 +57,7 @@ impl Catalog { hang::Catalog { video: self.video.clone(), audio: self.audio.clone(), + data: Default::default(), } } } diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index d96ef84831..d0fd369904 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -274,6 +274,7 @@ mod test { audio: Audio { renditions: audio_renditions, }, + ..Default::default() }; let msf = to_msf(&catalog); @@ -420,6 +421,7 @@ mod test { audio: Audio { renditions: audio_renditions, }, + ..Default::default() }; let msf = to_msf(&catalog); From 610a8ab0114195a75824144ee972b15c326ab9a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 00:04:02 +0000 Subject: [PATCH 2/7] review: reserved-track guard, non_exhaustive DataTrack, doc + test coverage Address CodeRabbit review feedback on #1748: - publish Broadcast.publishTrack now rejects reserved built-in track names (catalog/audio/video) so a handler that could never run fails fast. - rs/hang DataTrack gains #[non_exhaustive] so future optional fields stay additive for external constructors. - js/hang Root type doc comment now mentions the data section. - Add a teardown assertion to the publish JsonProducer test and a resubscribe-on-broadcast-switch test to the watch JsonConsumer test. --- js/hang/src/catalog/root.ts | 5 ++++- js/publish/src/broadcast.ts | 15 ++++++++++++++- js/publish/src/json.test.ts | 4 ++++ js/watch/src/json.test.ts | 33 ++++++++++++++++++++++++++++++++- rs/hang/src/catalog/data.rs | 1 + 5 files changed, 55 insertions(+), 3 deletions(-) diff --git a/js/hang/src/catalog/root.ts b/js/hang/src/catalog/root.ts index 72ea18492b..ec2662ce4d 100644 --- a/js/hang/src/catalog/root.ts +++ b/js/hang/src/catalog/root.ts @@ -21,5 +21,8 @@ export const RootSchema = z.looseObject({ data: z.optional(DataSchema), }); -/** The root catalog object, with optional video and audio sections plus any app extensions. */ +/** + * The root catalog object, with optional `video`, `audio`, and `data` (custom application tracks) + * sections plus any app extensions. + */ export type Root = z.infer; diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 0edccb3b4b..11489f080d 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -45,6 +45,15 @@ export class Broadcast { // reconnects so a new `Moq.Broadcast` still serves them. #tracks = new Map(); + // Built-in track names handled before `#tracks`, so a custom handler registered under one of + // these would never run. `publishTrack` rejects them to fail fast. + static readonly #RESERVED_TRACKS: ReadonlySet = new Set([ + Broadcast.CATALOG_TRACK, + Audio.Encoder.TRACK, + Video.Root.TRACK_HD, + Video.Root.TRACK_SD, + ]); + signals = new Effect(); constructor(props?: BroadcastProps) { @@ -141,9 +150,13 @@ export class Broadcast { * {@link publishJson} for a JSON track that also advertises itself in the catalog. * * Returns a function that unregisters the handler. Note this does not close already-served - * subscriptions, nor touch the catalog. + * subscriptions, nor touch the catalog. Throws if `name` collides with a built-in track + * (catalog/audio/video), since those are served first and the handler would never run. */ publishTrack(name: string, serve: ServeTrack): () => void { + if (Broadcast.#RESERVED_TRACKS.has(name)) { + throw new Error(`Track name is reserved: ${name}`); + } this.#tracks.set(name, serve); return () => { if (this.#tracks.get(name) === serve) this.#tracks.delete(name); diff --git a/js/publish/src/json.test.ts b/js/publish/src/json.test.ts index 45ef46119c..a3f662f08e 100644 --- a/js/publish/src/json.test.ts +++ b/js/publish/src/json.test.ts @@ -23,7 +23,11 @@ test("seeds late subscribers with the current value and fans out updates", async producer.update({ title: "world" }); expect(await consumer.next()).toEqual({ title: "world" }); + // Closing the effect finishes the subscription: it's dropped from the fan-out and the track + // ends, so further updates never reach it. effect.close(); + producer.update({ title: "after close" }); + expect(await consumer.next()).toBeUndefined(); }); test("mutate composes from the last value", async () => { diff --git a/js/watch/src/json.test.ts b/js/watch/src/json.test.ts index ce064c5d91..449e9649ab 100644 --- a/js/watch/src/json.test.ts +++ b/js/watch/src/json.test.ts @@ -6,9 +6,14 @@ import { JsonConsumer } from "./json.ts"; // Resolve once the signal holds a defined value. function nextDefined(signal: Getter): Promise { + return waitFor(signal, (value): value is T => value !== undefined); +} + +// Resolve once the signal's value satisfies the predicate. +function waitFor(signal: Getter, predicate: (value: T) => value is U): Promise { return new Promise((resolve) => { const dispose = signal.subscribe((value) => { - if (value === undefined) return; + if (!predicate(value)) return; dispose(); resolve(value); }); @@ -58,3 +63,29 @@ test("clears the value when the broadcast goes away", async () => { consumer.close(); }); + +test("resubscribes when the active broadcast switches", async () => { + const first = new MoqBroadcast(); + const active = new Signal(first); + + const consumer = new JsonConsumer<{ title: string }>(active, "meta.json"); + + const req1 = await first.requested(); + if (!req1) throw new Error("expected a track request on the first broadcast"); + new Json.Producer<{ title: string }>(req1.track).update({ title: "first" }); + expect(await nextDefined(consumer.value)).toEqual({ title: "first" }); + + // Switching to a new broadcast resubscribes and reads from the new track. + const second = new MoqBroadcast(); + active.set(second); + + const req2 = await second.requested(); + if (!req2) throw new Error("expected a track request on the second broadcast"); + expect(req2.track.name).toBe("meta.json"); + new Json.Producer<{ title: string }>(req2.track).update({ title: "second" }); + + const value = await waitFor(consumer.value, (v): v is { title: string } => v?.title === "second"); + expect(value).toEqual({ title: "second" }); + + consumer.close(); +}); diff --git a/rs/hang/src/catalog/data.rs b/rs/hang/src/catalog/data.rs index f72b790eeb..87743422a5 100644 --- a/rs/hang/src/catalog/data.rs +++ b/rs/hang/src/catalog/data.rs @@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize}; #[serde_with::skip_serializing_none] #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] #[serde(rename_all = "camelCase")] +#[non_exhaustive] pub struct DataTrack { /// The MIME type of each frame's payload, e.g. `"application/json"`. Informational. pub mime: Option, From 32f8a289d95718d30492e433bb2f7ee1f4e1e546 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 00:26:48 +0000 Subject: [PATCH 3/7] Drop typed catalog data section; keep generic custom-track plumbing Reframe per review: applications add their own typed catalog sections (like scte35) through the existing loose catalog root, and publish/subscribe arbitrary tracks through generic helpers. MoQ stays generic with no per-application support and no catalog-format change. - Revert the typed `data` catalog section in js/hang and rs/hang (and the moq-mux construction sites). The loose RootSchema already lets an app add its own sections (e.g. scte35) that pass through to consumers untouched. - publishJson no longer writes a catalog entry; the app advertises the track by mutating its own section via broadcast.catalog.mutate(...). - Add watch Broadcast.subscribeTrack for arbitrary (non-JSON) tracks, mirroring publish Broadcast.publishTrack. - Update docs to show the app-driven pattern: a JS app supports scte35 with no explicit support in hang / @moq/publish / @moq/watch. --- doc/concept/layer/hang.md | 17 ++++----- doc/lib/js/@moq/publish.md | 23 ++++++++----- doc/lib/js/@moq/watch.md | 26 +++++++------- js/hang/src/catalog/data.ts | 31 ----------------- js/hang/src/catalog/index.ts | 1 - js/hang/src/catalog/root.test.ts | 7 ---- js/hang/src/catalog/root.ts | 10 +----- js/publish/src/broadcast.ts | 28 +++++---------- js/watch/src/broadcast.ts | 30 ++++++++++++++-- rs/hang/src/catalog/data.rs | 27 --------------- rs/hang/src/catalog/mod.rs | 2 -- rs/hang/src/catalog/root.rs | 55 +++--------------------------- rs/moq-mux/src/catalog/hang/ext.rs | 1 - rs/moq-mux/src/catalog/producer.rs | 2 -- 14 files changed, 78 insertions(+), 182 deletions(-) delete mode 100644 js/hang/src/catalog/data.ts delete mode 100644 rs/hang/src/catalog/data.rs diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index 7c928ea3ec..a86a9ef886 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -83,19 +83,16 @@ The catalog is a JSON document published through the merge-patch helper (`@moq/j This keeps application-specific sections in the application layer while the base catalog stays generic. -### Data tracks +### Custom tracks -Beyond `audio`/`video`, the catalog has an optional `data` section: a flat map of track name to a small descriptor (`mime`, `description`). It advertises arbitrary application-defined tracks (for example a `meta.json` metadata track) carried within the same broadcast, so a consumer can discover and subscribe to them without a separate broadcast. The catalog says nothing about how to decode a data track; the descriptor fields are hints for the consumer. +A custom catalog section can carry its payload inline (for low-rate metadata), or it can reference a separate track in the same broadcast (for a stream of data, e.g. a `meta.json` track or an SCTE-35 event track). The relay treats such a track like any other; only the publisher and consumer give it meaning. -```json -{ - "data": { - "meta.json": { "mime": "application/json", "description": "broadcast metadata" } - } -} -``` +The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support: + +- **Publish**: `broadcast.publishJson(name)` serves a JSON track (seeding late joiners with the latest value), and `broadcast.publishTrack(name, serve)` serves arbitrary bytes. Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`. +- **Watch**: `broadcast.subscribeJson(name)` follows the active broadcast and exposes the latest value as a signal, and `broadcast.subscribeTrack(name, priority, consume)` is the raw-bytes equivalent. Read your section back from `broadcast.catalog` (unknown sections pass through the loose schema). -The web components expose this directly: on the publish side, `broadcast.publishJson("meta.json")` serves a JSON track (seeding late joiners with the latest value) and adds the `data` entry; on the watch side, `broadcast.subscribeJson("meta.json")` follows the active broadcast and exposes the latest value as a signal. The section is omitted from the wire when empty, so existing catalogs are unaffected. +So an application supports something like SCTE-35 entirely in its own code: publish an `scte35` section (and optionally a track) on one side, read it on the other, without hang, `@moq/publish`, or `@moq/watch` knowing anything about SCTE-35. ## Container diff --git a/doc/lib/js/@moq/publish.md b/doc/lib/js/@moq/publish.md index b92da56f2f..73824e1e3e 100644 --- a/doc/lib/js/@moq/publish.md +++ b/doc/lib/js/@moq/publish.md @@ -127,21 +127,28 @@ broadcast.video.device.set("screen"); broadcast.name.set("bob.hang"); ``` -### Custom data tracks +### Custom tracks and catalog sections Beyond audio and video, you can publish arbitrary application tracks within the -same broadcast (no separate broadcast needed). `publishJson` serves a JSON track -(seeding late joiners with the latest value) and advertises it in the catalog -[`data` section](/concept/layer/hang#data-tracks): +same broadcast (no separate broadcast needed). `publishJson` serves a JSON track, +seeding late joiners with the latest value. It does not touch the catalog; you +advertise the track by writing your own section to `broadcast.catalog` (the +[catalog root](/concept/layer/hang#extensions) is a loose object, so any key +passes through). This lets an app support something like an `scte35` section with +no hang-specific support: ```typescript -const meta = broadcast.publishJson<{ title: string }>("meta.json"); -meta.update({ title: "Live from the moon" }); +const scte35 = broadcast.publishJson<{ splices: number[] }>("scte35.json"); +broadcast.catalog.mutate((c) => { + c.scte35 = { track: "scte35.json" }; +}); +scte35.update({ splices: [] }); ``` For non-JSON payloads, `publishTrack(name, serve)` is the low-level escape hatch: -`serve` runs whenever a subscriber requests the named track. The component -exposes both via its `broadcast` property (`el.broadcast.publishJson(...)`). +`serve` runs whenever a subscriber requests the named track. Both reject the +built-in track names (catalog/audio/video). The component exposes everything via +its `broadcast` property (`el.broadcast.publishJson(...)`). ## Related Packages diff --git a/doc/lib/js/@moq/watch.md b/doc/lib/js/@moq/watch.md index cab4af13f3..40fa1b8437 100644 --- a/doc/lib/js/@moq/watch.md +++ b/doc/lib/js/@moq/watch.md @@ -139,26 +139,28 @@ el.catalog = myCatalog; > `"msf"`) tears down the previous fetch loop, which clears `catalog`. Set the > catalog *after* switching to `"manual"`, not before. -### Custom data tracks +### Custom tracks and catalog sections A broadcast can carry arbitrary application tracks (for example a `meta.json` -metadata track) alongside the media, listed in the catalog -[`data` section](/concept/layer/hang#data-tracks). Use `subscribeJson` to read -one: it follows the active broadcast across reconnects and exposes the latest -value as a signal. Call `close()` when done (it's also closed when the broadcast -closes). +metadata track) alongside the media. An application advertises them in its own +catalog section (the [catalog root](/concept/layer/hang#extensions) is a loose +object, so unknown sections pass through to `broadcast.catalog`). Use +`subscribeJson` to read a JSON track: it follows the active broadcast across +reconnects and exposes the latest value as a signal. Call `close()` when done +(it's also closed when the broadcast closes). ```typescript +// The app's own catalog section, read back from the loose catalog. +const scte35 = broadcast.catalog.peek()?.scte35; + const meta = broadcast.subscribeJson<{ title: string }>("meta.json"); meta.value.subscribe((value) => console.log("meta", value)); ``` -The component exposes the same method via its `broadcast` property: - -```typescript -const el = document.querySelector("moq-watch")!; -const meta = el.broadcast.subscribeJson("meta.json"); -``` +For non-JSON payloads, `subscribeTrack(name, priority, consume)` is the +low-level equivalent: `consume` runs with the subscribed track each time a +broadcast becomes active. The component exposes everything via its `broadcast` +property (`el.broadcast.subscribeJson(...)`). ## UI Overlay diff --git a/js/hang/src/catalog/data.ts b/js/hang/src/catalog/data.ts deleted file mode 100644 index b66afc3bcf..0000000000 --- a/js/hang/src/catalog/data.ts +++ /dev/null @@ -1,31 +0,0 @@ -import * as z from "zod/mini"; - -/** - * A custom, application-defined data track in the catalog. - * - * Unlike `audio`/`video`, the catalog says nothing about how to decode a data track. It just - * advertises that the track exists (keyed by name in {@link DataSchema}) so a consumer can discover - * and subscribe to it. The optional fields are hints for the consumer, not instructions for hang. - */ -export const DataTrackSchema = z.object({ - /** The MIME type of each frame's payload, e.g. `"application/json"`. Informational. */ - mime: z.optional(z.string()), - - /** A free-form description of the track's contents, for humans. */ - description: z.optional(z.string()), -}); - -/** A custom, application-defined data track in the catalog. */ -export type DataTrack = z.infer; - -/** - * The `data` catalog section: a map of track name to {@link DataTrack}. - * - * Each key is the name of a track within the broadcast (e.g. `"meta.json"`). The tracks are - * independent of each other, so this is a flat map rather than the rendition groups used by - * `audio`/`video`. - */ -export const DataSchema = z.record(z.string(), DataTrackSchema); - -/** The `data` catalog section: a map of track name to {@link DataTrack}. */ -export type Data = z.infer; diff --git a/js/hang/src/catalog/index.ts b/js/hang/src/catalog/index.ts index a4429c8ab6..6a0c787b54 100644 --- a/js/hang/src/catalog/index.ts +++ b/js/hang/src/catalog/index.ts @@ -7,7 +7,6 @@ export * from "./audio"; export * from "./container"; -export * from "./data"; export * from "./format"; export * from "./integers"; export * from "./priority"; diff --git a/js/hang/src/catalog/root.test.ts b/js/hang/src/catalog/root.test.ts index 80bce585dc..5e2a2f3bc2 100644 --- a/js/hang/src/catalog/root.test.ts +++ b/js/hang/src/catalog/root.test.ts @@ -12,13 +12,6 @@ test("base catalog preserves unknown sections", () => { expect(parsed.scte35).toEqual({ spliceId: 42 }); }); -test("parses the data section of custom tracks", () => { - const parsed = RootSchema.parse({ - data: { "meta.json": { mime: "application/json", description: "broadcast metadata" } }, - }); - expect(parsed.data?.["meta.json"]).toEqual({ mime: "application/json", description: "broadcast metadata" }); -}); - test("extended schema validates app sections", () => { const Scte35Schema = z.object({ spliceId: z.number() }); const ExtendedSchema = z.extend(RootSchema, { scte35: z.optional(Scte35Schema) }); diff --git a/js/hang/src/catalog/root.ts b/js/hang/src/catalog/root.ts index ec2662ce4d..9f7be7f121 100644 --- a/js/hang/src/catalog/root.ts +++ b/js/hang/src/catalog/root.ts @@ -1,7 +1,6 @@ import * as z from "zod/mini"; import { AudioSchema } from "./audio"; -import { DataSchema } from "./data"; import { VideoSchema } from "./video"; /** @@ -11,18 +10,11 @@ import { VideoSchema } from "./video"; * application can add its own sections (e.g. `scte35`) without modifying hang. A base consumer * ignores the extra sections; an extended consumer validates them with its own schema, typically * built via `z.extend(RootSchema, { ... })`. - * - * The `data` section lists arbitrary application-defined tracks (e.g. a `meta.json` track) carried - * alongside the media within the same broadcast. */ export const RootSchema = z.looseObject({ video: z.optional(VideoSchema), audio: z.optional(AudioSchema), - data: z.optional(DataSchema), }); -/** - * The root catalog object, with optional `video`, `audio`, and `data` (custom application tracks) - * sections plus any app extensions. - */ +/** The root catalog object, with optional video and audio sections plus any app extensions. */ export type Root = z.infer; diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 11489f080d..68fc5e4150 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -18,14 +18,6 @@ export type BroadcastProps = { /** Serves a custom track when a subscriber requests it, scoped to the subscription's lifetime. */ export type ServeTrack = (track: Moq.Track, effect: Effect) => void; -/** Options for {@link Broadcast.publishJson}, extending the {@link Json.Config} for the value. */ -export type PublishJsonOptions = Json.Config & { - /** MIME type recorded in the catalog `data` entry. Defaults to `"application/json"`. */ - mime?: string; - /** Human-readable description recorded in the catalog `data` entry. */ - description?: string; -}; - export class Broadcast { static readonly CATALOG_TRACK = "catalog.json"; @@ -164,27 +156,25 @@ export class Broadcast { } /** - * Publish a custom JSON track within this broadcast and advertise it in the catalog `data` - * section. + * Publish a custom JSON track within this broadcast. * * Returns a {@link JsonProducer}: call `update`/`mutate` to set the value, which is served to * every subscriber (seeding late joiners with the latest value). The track lives for the * lifetime of the broadcast. * + * This does not touch the catalog. To advertise the track, write your own section to + * {@link catalog} (the root is a loose object, so any key passes through). For example, to support + * a custom `scte35` section with no hang-specific support: + * * ```ts - * const meta = broadcast.publishJson("meta.json"); - * meta.update({ title: "Live from the moon" }); + * const scte35 = broadcast.publishJson("scte35.json"); + * broadcast.catalog.mutate((c) => { c.scte35 = { track: "scte35.json" }; }); + * scte35.update({ splices: [] }); * ``` */ - publishJson(name: string, options?: PublishJsonOptions): JsonProducer { + publishJson(name: string, options?: Json.Config): JsonProducer { const producer = new JsonProducer(options); this.publishTrack(name, (track, effect) => producer.serve(track, effect)); - - this.catalog.mutate((catalog) => { - catalog.data ??= {}; - catalog.data[name] = { mime: options?.mime ?? "application/json", description: options?.description }; - }); - return producer; } diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index bae001a9b0..76ec147a7d 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -8,6 +8,9 @@ import { Effect, type Getter, Signal } from "@moq/signals"; import { JsonConsumer, type SubscribeJsonOptions } from "./json"; import { toHang } from "./msf"; +/** Consumes a custom track once subscribed, scoped to the subscription's lifetime. */ +export type ConsumeTrack = (track: Moq.Track, effect: Effect) => void; + // Watch supports the two on-the-wire catalog formats from @moq/hang plus a // "manual" mode where the user supplies the catalog directly without fetching. export const CATALOG_FORMATS = [...Catalog.FORMATS, "manual"] as const; @@ -192,8 +195,31 @@ export class Broadcast { } /** - * Subscribe to a custom JSON track within this broadcast (e.g. a `meta.json` track listed in the - * catalog `data` section). + * Subscribe to a custom track within this broadcast, following the active broadcast across + * reconnects. `consume` runs with a freshly-subscribed track and a subscription-scoped effect + * each time a broadcast becomes active (re-running on reconnect). This is the low-level escape + * hatch for arbitrary payloads; see {@link subscribeJson} for a JSON convenience. + * + * Returns a function to stop subscribing; also stopped when this broadcast closes. + */ + subscribeTrack(name: string, priority: number, consume: ConsumeTrack): () => void { + const signals = new Effect(); + signals.run((effect) => { + const active = effect.get(this.active); + if (!active) return; + + const track = active.subscribe(name, priority); + effect.cleanup(() => track.close()); + + consume(track, effect); + }); + this.signals.cleanup(() => signals.close()); + return () => signals.close(); + } + + /** + * Subscribe to a custom JSON track within this broadcast (e.g. a `meta.json` track that an + * application advertises in its own catalog section). * * Follows the active broadcast across reconnects and exposes the latest reconstructed value as a * Signal on the returned consumer. The consumer is closed when this broadcast closes; close it diff --git a/rs/hang/src/catalog/data.rs b/rs/hang/src/catalog/data.rs deleted file mode 100644 index 87743422a5..0000000000 --- a/rs/hang/src/catalog/data.rs +++ /dev/null @@ -1,27 +0,0 @@ -use std::collections::BTreeMap; - -use serde::{Deserialize, Serialize}; - -/// A custom, application-defined data track in the catalog. -/// -/// Unlike `audio`/`video`, the catalog says nothing about how to decode a data track. It just -/// advertises that the track exists (keyed by name in the [`Data`] map) so a consumer can discover -/// and subscribe to it. The optional fields are hints for the consumer, not instructions for hang. -#[serde_with::skip_serializing_none] -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] -#[serde(rename_all = "camelCase")] -#[non_exhaustive] -pub struct DataTrack { - /// The MIME type of each frame's payload, e.g. `"application/json"`. Informational. - pub mime: Option, - - /// A free-form description of the track's contents, for humans. - pub description: Option, -} - -/// The `data` catalog section: a map of track name to [`DataTrack`]. -/// -/// Each key is the name of a track within the broadcast (e.g. `"meta.json"`). The tracks are -/// independent of each other, so this is a flat map rather than the rendition groups used by -/// `audio`/`video`. -pub type Data = BTreeMap; diff --git a/rs/hang/src/catalog/mod.rs b/rs/hang/src/catalog/mod.rs index def874e573..bab4f5ffeb 100644 --- a/rs/hang/src/catalog/mod.rs +++ b/rs/hang/src/catalog/mod.rs @@ -6,12 +6,10 @@ mod audio; mod container; -mod data; mod root; mod video; pub use audio::*; pub use container::*; -pub use data::*; pub use root::*; pub use video::*; diff --git a/rs/hang/src/catalog/root.rs b/rs/hang/src/catalog/root.rs index 4162cfccb0..9c68c5c7a8 100644 --- a/rs/hang/src/catalog/root.rs +++ b/rs/hang/src/catalog/root.rs @@ -1,13 +1,12 @@ //! This module contains the structs and functions for the MoQ catalog format use crate::Result; -use crate::catalog::{Audio, Data, Video}; +use crate::catalog::{Audio, Video}; use serde::{Deserialize, Serialize}; /// A catalog track, created by a broadcaster to describe the tracks available in a broadcast. /// -/// The base catalog carries the media sections (`video`, `audio`) plus an optional `data` section -/// listing arbitrary application-defined tracks (e.g. a `meta.json` track). Applications extend it -/// with their own root sections (e.g. `scte35`) by flattening this struct into their own with +/// The base catalog carries only the media sections (`video`, `audio`). Applications extend it with +/// their own root sections (e.g. `scte35`) by flattening this struct into their own with /// `#[serde(flatten)]`. The catalog does not deny unknown fields, so a base consumer ignores the /// extra sections and an extended catalog stays wire-compatible. See the `extension_roundtrip` test. #[serde_with::serde_as] @@ -28,11 +27,6 @@ pub struct Catalog { /// based on their preferences (codec, bitrate, language, etc). #[serde(default)] pub audio: Audio, - - /// Custom application-defined data tracks (e.g. a `meta.json` track) carried within the - /// broadcast. Omitted from the wire when empty, so existing catalogs are unaffected. - #[serde(default, skip_serializing_if = "Data::is_empty")] - pub data: Data, } impl Catalog { @@ -87,7 +81,7 @@ impl Catalog { mod test { use std::collections::BTreeMap; - use crate::catalog::{AudioCodec::Opus, AudioConfig, Container, Data, DataTrack, H264, VideoConfig}; + use crate::catalog::{AudioCodec::Opus, AudioConfig, Container, H264, VideoConfig}; use super::*; @@ -154,7 +148,6 @@ mod test { audio: Audio { renditions: audio_renditions, }, - data: Data::default(), }; let output = Catalog::from_str(&encoded).expect("failed to decode"); @@ -164,46 +157,6 @@ mod test { assert_eq!(encoded, output, "wrong encoded output"); } - #[test] - fn data_roundtrip() { - let mut encoded = r#"{ - "video": {"renditions": {}}, - "audio": {"renditions": {}}, - "data": { - "meta.json": {"mime": "application/json", "description": "metadata"} - } - }"# - .to_string(); - encoded.retain(|c| !c.is_whitespace()); - - let mut data = Data::default(); - data.insert( - "meta.json".to_string(), - DataTrack { - mime: Some("application/json".to_string()), - description: Some("metadata".to_string()), - }, - ); - - let decoded = Catalog { - data, - ..Default::default() - }; - - assert_eq!(decoded, Catalog::from_str(&encoded).expect("failed to decode")); - assert_eq!(encoded, decoded.to_string().expect("failed to encode")); - } - - #[test] - fn data_omitted_when_empty() { - // An empty data section must not appear on the wire, so existing catalogs are unaffected. - let encoded = Catalog::default().to_string().expect("failed to encode"); - assert!( - !encoded.contains("data"), - "empty data section should be omitted: {encoded}" - ); - } - #[test] fn extension_roundtrip() { // An application extends the catalog with its own root section by flattening Catalog. diff --git a/rs/moq-mux/src/catalog/hang/ext.rs b/rs/moq-mux/src/catalog/hang/ext.rs index 1c616a430c..3de6a3c0f8 100644 --- a/rs/moq-mux/src/catalog/hang/ext.rs +++ b/rs/moq-mux/src/catalog/hang/ext.rs @@ -57,7 +57,6 @@ impl Catalog { hang::Catalog { video: self.video.clone(), audio: self.audio.clone(), - data: Default::default(), } } } diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index d0fd369904..d96ef84831 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -274,7 +274,6 @@ mod test { audio: Audio { renditions: audio_renditions, }, - ..Default::default() }; let msf = to_msf(&catalog); @@ -421,7 +420,6 @@ mod test { audio: Audio { renditions: audio_renditions, }, - ..Default::default() }; let msf = to_msf(&catalog); From 480378811ecfeccc4dc6041003277e010bb78927 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 00:32:27 +0000 Subject: [PATCH 4/7] docs: fix stale publishTrack JSDoc reference to catalog advertising publishJson no longer writes a catalog entry, so the cross-reference in publishTrack's doc was contradictory. --- js/publish/src/broadcast.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 68fc5e4150..9e4d9c229a 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -139,7 +139,7 @@ export class Broadcast { * When a subscriber requests a track with this name, `serve` runs with the track and an effect * scoped to that subscription (cleaned up when the subscriber goes away). The handler persists * across reconnects. This is the low-level escape hatch for arbitrary payloads; see - * {@link publishJson} for a JSON track that also advertises itself in the catalog. + * {@link publishJson} for a fan-out JSON-track helper. * * Returns a function that unregisters the handler. Note this does not close already-served * subscriptions, nor touch the catalog. Throws if `name` collides with a built-in track From 02619f7d10fe72858e8852e71efc0f08e0ce76c1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 01:06:22 +0000 Subject: [PATCH 5/7] Replace JSON wrappers with @moq/json Source; CatalogProducer is an alias Per review: drop the @moq/publish/@moq/watch JSON-specific wrappers and have applications use @moq/json directly on the generic track hooks. - Promote the fan-out producer into @moq/json as `Source` (holds a value, seeds late joiners, fans out to every subscriber). This is the primitive the catalog was already using. - `CatalogProducer` becomes a thin type alias for `Json.Source`; the broadcast constructs it with `initial: {}`. Its fan-out tests move to js/json/src/source.test.ts. - Remove publishJson/subscribeJson and the JsonProducer/JsonConsumer classes. publishTrack/subscribeTrack remain as the generic hooks; the app wraps the track with @moq/json Source (publish) or Consumer (watch). - Re-export @moq/json as `Json` from @moq/publish and @moq/watch so apps have a single import for the JSON track tooling. - Docs updated to the @moq/json-direct pattern. --- doc/concept/layer/hang.md | 6 +- doc/lib/js/@moq/publish.md | 29 +++--- doc/lib/js/@moq/watch.md | 34 ++++--- js/json/src/index.ts | 1 + js/json/src/source.test.ts | 60 ++++++++++++ .../src/json.ts => json/src/source.ts} | 23 ++--- js/publish/src/broadcast.ts | 48 ++++------ js/publish/src/catalog.test.ts | 56 ------------ js/publish/src/catalog.ts | 20 ++-- js/publish/src/index.ts | 2 +- js/publish/src/json.test.ts | 50 ---------- js/watch/src/broadcast.ts | 42 ++++----- js/watch/src/index.ts | 2 +- js/watch/src/json.test.ts | 91 ------------------- js/watch/src/json.ts | 61 ------------- 15 files changed, 161 insertions(+), 364 deletions(-) create mode 100644 js/json/src/source.test.ts rename js/{publish/src/json.ts => json/src/source.ts} (75%) delete mode 100644 js/publish/src/catalog.test.ts delete mode 100644 js/publish/src/json.test.ts delete mode 100644 js/watch/src/json.test.ts delete mode 100644 js/watch/src/json.ts diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index a86a9ef886..baefc7b514 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -87,10 +87,10 @@ This keeps application-specific sections in the application layer while the base A custom catalog section can carry its payload inline (for low-rate metadata), or it can reference a separate track in the same broadcast (for a stream of data, e.g. a `meta.json` track or an SCTE-35 event track). The relay treats such a track like any other; only the publisher and consumer give it meaning. -The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support: +The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support. Each exposes a low-level track hook and re-exports [`@moq/json`](/lib/js/@moq/hang) so the application encodes the payload itself: -- **Publish**: `broadcast.publishJson(name)` serves a JSON track (seeding late joiners with the latest value), and `broadcast.publishTrack(name, serve)` serves arbitrary bytes. Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`. -- **Watch**: `broadcast.subscribeJson(name)` follows the active broadcast and exposes the latest value as a signal, and `broadcast.subscribeTrack(name, priority, consume)` is the raw-bytes equivalent. Read your section back from `broadcast.catalog` (unknown sections pass through the loose schema). +- **Publish**: `broadcast.publishTrack(name, serve)` runs `serve(track, effect)` per subscriber. For JSON, serve each track from a shared `Json.Source` (the same fan-out producer the catalog uses, seeding late joiners with the latest value). Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`. +- **Watch**: `broadcast.subscribeTrack(name, priority, consume)` follows the active broadcast across reconnects. For JSON, wrap the track in a `Json.Consumer` inside `consume`. Read your section back from `broadcast.catalog` (unknown sections pass through the loose schema). So an application supports something like SCTE-35 entirely in its own code: publish an `scte35` section (and optionally a track) on one side, read it on the other, without hang, `@moq/publish`, or `@moq/watch` knowing anything about SCTE-35. diff --git a/doc/lib/js/@moq/publish.md b/doc/lib/js/@moq/publish.md index 73824e1e3e..da8c77ca68 100644 --- a/doc/lib/js/@moq/publish.md +++ b/doc/lib/js/@moq/publish.md @@ -130,25 +130,30 @@ broadcast.name.set("bob.hang"); ### Custom tracks and catalog sections Beyond audio and video, you can publish arbitrary application tracks within the -same broadcast (no separate broadcast needed). `publishJson` serves a JSON track, -seeding late joiners with the latest value. It does not touch the catalog; you -advertise the track by writing your own section to `broadcast.catalog` (the -[catalog root](/concept/layer/hang#extensions) is a loose object, so any key -passes through). This lets an app support something like an `scte35` section with -no hang-specific support: +same broadcast (no separate broadcast needed). `publishTrack(name, serve)` runs +`serve(track, effect)` for each subscriber; it rejects the built-in track names +(catalog/audio/video). Encode the payload yourself with the re-exported +[`@moq/json`](/lib/js/@moq/hang): a `Json.Source` is the same fan-out producer the +catalog uses, seeding late joiners with the latest value. + +`publishTrack` does not touch the catalog; advertise the track by writing your own +section to `broadcast.catalog` (the [catalog root](/concept/layer/hang#extensions) +is a loose object, so any key passes through). This lets an app support something +like an `scte35` section with no hang-specific support: ```typescript -const scte35 = broadcast.publishJson<{ splices: number[] }>("scte35.json"); +import { Json } from "@moq/publish"; + +const scte35 = new Json.Source<{ splices: number[] }>({ initial: { splices: [] } }); +broadcast.publishTrack("scte35.json", (track, effect) => scte35.serve(track, effect)); broadcast.catalog.mutate((c) => { c.scte35 = { track: "scte35.json" }; }); -scte35.update({ splices: [] }); +scte35.update({ splices: [42] }); ``` -For non-JSON payloads, `publishTrack(name, serve)` is the low-level escape hatch: -`serve` runs whenever a subscriber requests the named track. Both reject the -built-in track names (catalog/audio/video). The component exposes everything via -its `broadcast` property (`el.broadcast.publishJson(...)`). +The component exposes everything via its `broadcast` property +(`el.broadcast.publishTrack(...)`). ## Related Packages diff --git a/doc/lib/js/@moq/watch.md b/doc/lib/js/@moq/watch.md index 40fa1b8437..4b0a4356c7 100644 --- a/doc/lib/js/@moq/watch.md +++ b/doc/lib/js/@moq/watch.md @@ -144,23 +144,33 @@ el.catalog = myCatalog; A broadcast can carry arbitrary application tracks (for example a `meta.json` metadata track) alongside the media. An application advertises them in its own catalog section (the [catalog root](/concept/layer/hang#extensions) is a loose -object, so unknown sections pass through to `broadcast.catalog`). Use -`subscribeJson` to read a JSON track: it follows the active broadcast across -reconnects and exposes the latest value as a signal. Call `close()` when done -(it's also closed when the broadcast closes). +object, so unknown sections pass through to `broadcast.catalog`). +`subscribeTrack(name, priority, consume)` follows the active broadcast across +reconnects and runs `consume(track, effect)` each time it becomes active. Decode +the payload yourself with the re-exported [`@moq/json`](/lib/js/@moq/hang): ```typescript -// The app's own catalog section, read back from the loose catalog. -const scte35 = broadcast.catalog.peek()?.scte35; +import { Json } from "@moq/watch"; +import { Signals } from "@moq/watch"; -const meta = broadcast.subscribeJson<{ title: string }>("meta.json"); -meta.value.subscribe((value) => console.log("meta", value)); +// The app's own catalog section, read back from the loose catalog. +const section = broadcast.catalog.peek()?.scte35; + +const scte35 = new Signals.Signal<{ splices: number[] } | undefined>(undefined); +broadcast.subscribeTrack("scte35.json", 100, (track, effect) => { + const consumer = new Json.Consumer<{ splices: number[] }>(track); + effect.spawn(async () => { + for (;;) { + const next = await Promise.race([effect.cancel, consumer.next()]); + if (next === undefined) break; + scte35.set(next); + } + }); +}); ``` -For non-JSON payloads, `subscribeTrack(name, priority, consume)` is the -low-level equivalent: `consume` runs with the subscribed track each time a -broadcast becomes active. The component exposes everything via its `broadcast` -property (`el.broadcast.subscribeJson(...)`). +The component exposes everything via its `broadcast` property +(`el.broadcast.subscribeTrack(...)`). ## UI Overlay diff --git a/js/json/src/index.ts b/js/json/src/index.ts index ea6ff76d37..03418979e6 100644 --- a/js/json/src/index.ts +++ b/js/json/src/index.ts @@ -1,3 +1,4 @@ export { Consumer } from "./consumer.ts"; export { type Diff, deepEqual, diff, merge } from "./diff.ts"; export { type Config, Producer } from "./producer.ts"; +export { Source } from "./source.ts"; diff --git a/js/json/src/source.test.ts b/js/json/src/source.test.ts new file mode 100644 index 0000000000..4d08d9ffc6 --- /dev/null +++ b/js/json/src/source.test.ts @@ -0,0 +1,60 @@ +import { expect, test } from "bun:test"; +import { Track } from "@moq/net"; +import { Effect } from "@moq/signals"; +import { Consumer } from "./consumer.ts"; +import { Source } from "./source.ts"; + +test("seeds subscribers and fans out edits", async () => { + const source = new Source>({ initial: {} }); + + // Edit before anyone subscribes: the value is retained, not lost. + source.mutate((v) => { + v.video = { renditions: {} }; + }); + + const effect = new Effect(); + const track = new Track("catalog.json"); + source.serve(track, effect); + const consumer = new Consumer>(track); + + // A new subscriber is seeded with the current value. + expect((await consumer.next())?.video).toEqual({ renditions: {} }); + + // An independent owner edits its own key; the subscriber sees it, the other key untouched. + source.mutate((v) => { + v.scte35 = { splices: [] }; + }); + const update = await consumer.next(); + expect(update?.video).toEqual({ renditions: {} }); + expect(update?.scte35).toEqual({ splices: [] }); + + effect.close(); +}); + +test("a reconnecting subscriber is seeded with the full current value", async () => { + const source = new Source>({ initial: {} }); + source.mutate((v) => { + v.video = { renditions: {} }; + v.scte35 = { splices: [] }; + }); + + // The first subscription drains and ends... + const first = new Effect(); + source.serve(new Track("catalog.json"), first); + first.close(); + + // ...and a fresh subscription still gets the current value, not nothing. + const effect = new Effect(); + const track = new Track("catalog.json"); + source.serve(track, effect); + const seeded = await new Consumer>(track).next(); + expect(seeded?.video).toEqual({ renditions: {} }); + expect(seeded?.scte35).toEqual({ splices: [] }); + + effect.close(); +}); + +test("mutate without a value or initial throws", () => { + const source = new Source>(); + expect(() => source.mutate(() => {})).toThrow(); +}); diff --git a/js/publish/src/json.ts b/js/json/src/source.ts similarity index 75% rename from js/publish/src/json.ts rename to js/json/src/source.ts index 1bcd5f0fd9..9eebbf87d5 100644 --- a/js/publish/src/json.ts +++ b/js/json/src/source.ts @@ -1,26 +1,27 @@ -import * as Json from "@moq/json"; import type * as Moq from "@moq/net"; import type { Effect } from "@moq/signals"; +import { type Config, Producer } from "./producer.ts"; + /** * A stable JSON value that fans out to on-demand subscription tracks. * - * Unlike a raw {@link Json.Producer}, this exists independently of any subscription: set the value + * Unlike a per-track {@link Producer}, this exists independently of any subscription: set the value * at any time with {@link update} or {@link mutate}, and each subscriber (including a relay that * reconnects) is seeded with the current value before receiving updates. Multiple independent owners * can share one instance and each edit only their own keys via {@link mutate}, so their sections * compose instead of clobbering one another. * - * This backs both the broadcast catalog and arbitrary custom JSON tracks (see - * `Broadcast.publishJson`). + * This backs the hang catalog, and an application can use it for its own custom tracks (serve it + * from a publish `Broadcast.publishTrack` handler). */ -export class JsonProducer { +export class Source { #value: T | undefined; - #outputs = new Set>(); - #config: Json.Config; + #outputs = new Set>(); + #config: Config; - /** Create a producer, optionally seeding an initial value and per-track config. */ - constructor(config: Json.Config = {}) { + /** Create a source, optionally seeding an initial value and per-track {@link Config}. */ + constructor(config: Config = {}) { this.#config = config; this.#value = config.initial; } @@ -30,7 +31,7 @@ export class JsonProducer { return this.#value; } - /** Replace the value; the result is published to all current subscribers. No-op if unchanged. */ + /** Replace the value; the result is published to all current subscribers. */ update(value: T): void { this.#value = value; for (const output of this.#outputs) output.update(value); @@ -56,7 +57,7 @@ export class JsonProducer { /** Serve a subscription request: seed it with the current value, then forward updates. */ serve(track: Moq.Track, effect: Effect): void { - const output = new Json.Producer(track, this.#config); + const output = new Producer(track, this.#config); if (this.#value !== undefined) output.update(this.#value); this.#outputs.add(output); diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 9e4d9c229a..ad536972b5 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -1,10 +1,9 @@ import * as Catalog from "@moq/hang/catalog"; -import type * as Json from "@moq/json"; +import { Source } from "@moq/json"; import * as Moq from "@moq/net"; import { Effect, Signal } from "@moq/signals"; import * as Audio from "./audio"; -import { CatalogProducer } from "./catalog"; -import { JsonProducer } from "./json"; +import type { CatalogProducer } from "./catalog"; import * as Video from "./video"; export type BroadcastProps = { @@ -30,8 +29,8 @@ export class Broadcast { // The catalog, editable at any time regardless of whether anyone is subscribed. The base // `video`/`audio` sections are kept in sync from the encoders; an application adds its own root - // sections (e.g. `scte35`) by locking it too. - readonly catalog = new CatalogProducer(); + // sections (e.g. `scte35`) by mutating it too. + readonly catalog: CatalogProducer = new Source({ initial: {} }); // Handlers for custom tracks registered via `publishTrack`, keyed by track name. Persists across // reconnects so a new `Moq.Broadcast` still serves them. @@ -138,12 +137,24 @@ export class Broadcast { * * When a subscriber requests a track with this name, `serve` runs with the track and an effect * scoped to that subscription (cleaned up when the subscriber goes away). The handler persists - * across reconnects. This is the low-level escape hatch for arbitrary payloads; see - * {@link publishJson} for a fan-out JSON-track helper. + * across reconnects. This is the generic hook for arbitrary payloads; encode them yourself. * * Returns a function that unregisters the handler. Note this does not close already-served * subscriptions, nor touch the catalog. Throws if `name` collides with a built-in track * (catalog/audio/video), since those are served first and the handler would never run. + * + * For a JSON track, serve each track from a `@moq/json` `Source` (the same fan-out producer the + * catalog uses, seeding late joiners with the latest value). Advertise the track by writing your + * own section to {@link catalog}, e.g. to support a custom `scte35` section with no hang-specific + * support: + * + * ```ts + * import { Source } from "@moq/json"; + * const scte35 = new Source({ initial: { splices: [] } }); + * broadcast.publishTrack("scte35.json", (track, effect) => scte35.serve(track, effect)); + * broadcast.catalog.mutate((c) => { c.scte35 = { track: "scte35.json" }; }); + * scte35.update({ splices: [42] }); + * ``` */ publishTrack(name: string, serve: ServeTrack): () => void { if (Broadcast.#RESERVED_TRACKS.has(name)) { @@ -155,29 +166,6 @@ export class Broadcast { }; } - /** - * Publish a custom JSON track within this broadcast. - * - * Returns a {@link JsonProducer}: call `update`/`mutate` to set the value, which is served to - * every subscriber (seeding late joiners with the latest value). The track lives for the - * lifetime of the broadcast. - * - * This does not touch the catalog. To advertise the track, write your own section to - * {@link catalog} (the root is a loose object, so any key passes through). For example, to support - * a custom `scte35` section with no hang-specific support: - * - * ```ts - * const scte35 = broadcast.publishJson("scte35.json"); - * broadcast.catalog.mutate((c) => { c.scte35 = { track: "scte35.json" }; }); - * scte35.update({ splices: [] }); - * ``` - */ - publishJson(name: string, options?: Json.Config): JsonProducer { - const producer = new JsonProducer(options); - this.publishTrack(name, (track, effect) => producer.serve(track, effect)); - return producer; - } - close() { this.signals.close(); this.audio.close(); diff --git a/js/publish/src/catalog.test.ts b/js/publish/src/catalog.test.ts deleted file mode 100644 index 8bc695b84a..0000000000 --- a/js/publish/src/catalog.test.ts +++ /dev/null @@ -1,56 +0,0 @@ -import { expect, test } from "bun:test"; -import type * as Catalog from "@moq/hang/catalog"; -import * as Json from "@moq/json"; -import { Track } from "@moq/net"; -import { Effect } from "@moq/signals"; -import { CatalogProducer } from "./catalog.ts"; - -test("catalog producer seeds subscribers and fans out edits", async () => { - const catalog = new CatalogProducer(); - - // Edit before anyone subscribes: the value is retained, not lost. - catalog.mutate((c) => { - c.video = { renditions: {} }; - }); - - const effect = new Effect(); - const track = new Track("catalog.json"); - catalog.serve(track, effect); - const consumer = new Json.Consumer(track); - - // A new subscriber is seeded with the current catalog. - expect((await consumer.next())?.video).toEqual({ renditions: {} }); - - // An extension owner adds its own section; the subscriber sees the update, video untouched. - catalog.mutate((c) => { - c.scte35 = { splices: [] }; - }); - const update = await consumer.next(); - expect(update?.video).toEqual({ renditions: {} }); - expect(update?.scte35).toEqual({ splices: [] }); - - effect.close(); -}); - -test("a reconnecting subscriber is seeded with the full current catalog", async () => { - const catalog = new CatalogProducer(); - catalog.mutate((c) => { - c.video = { renditions: {} }; - c.scte35 = { splices: [] }; - }); - - // The first subscription drains and ends... - const first = new Effect(); - catalog.serve(new Track("catalog.json"), first); - first.close(); - - // ...and a fresh subscription still gets the current catalog, not nothing. - const effect = new Effect(); - const track = new Track("catalog.json"); - catalog.serve(track, effect); - const seeded = await new Json.Consumer(track).next(); - expect(seeded?.video).toEqual({ renditions: {} }); - expect(seeded?.scte35).toEqual({ splices: [] }); - - effect.close(); -}); diff --git a/js/publish/src/catalog.ts b/js/publish/src/catalog.ts index 7bce178f1c..c4485ca3cd 100644 --- a/js/publish/src/catalog.ts +++ b/js/publish/src/catalog.ts @@ -1,19 +1,11 @@ import type * as Catalog from "@moq/hang/catalog"; - -import { JsonProducer } from "./json"; +import type { Source } from "@moq/json"; /** - * A stable catalog producer that fans out to on-demand subscription tracks. + * The broadcast catalog: a {@link Source} of the catalog {@link Catalog.Root}, seeded empty. * - * Unlike a raw track producer, this exists independently of any subscription: edit it at any time - * with `mutate`, and each subscriber (including a relay that reconnects) is seeded with the current - * catalog before receiving updates. Independent owners (the base `video`/`audio` and an - * application's own sections, e.g. `scte35`) each edit only their own keys, so their sections - * compose instead of clobbering one another. + * Edit it at any time with `mutate` (the base `video`/`audio` sections are kept in sync by the + * encoders; an application adds its own root sections, e.g. `scte35`, the same way). Each + * subscriber, including a relay that reconnects, is seeded with the current catalog before updates. */ -export class CatalogProducer extends JsonProducer { - /** Create a catalog producer seeded with an empty catalog. */ - constructor() { - super({ initial: {} }); - } -} +export type CatalogProducer = Source; diff --git a/js/publish/src/index.ts b/js/publish/src/index.ts index f81cec91f7..03ffe56892 100644 --- a/js/publish/src/index.ts +++ b/js/publish/src/index.ts @@ -1,4 +1,5 @@ export * as Hang from "@moq/hang"; +export * as Json from "@moq/json"; export * as Net from "@moq/net"; /** @deprecated Use `Net` instead. */ export * as Lite from "@moq/net"; @@ -6,7 +7,6 @@ export * as Signals from "@moq/signals"; export * as Audio from "./audio"; export * from "./broadcast"; export * from "./catalog"; -export * from "./json"; export * as Preview from "./preview"; export * as Source from "./source"; export * as Video from "./video"; diff --git a/js/publish/src/json.test.ts b/js/publish/src/json.test.ts deleted file mode 100644 index a3f662f08e..0000000000 --- a/js/publish/src/json.test.ts +++ /dev/null @@ -1,50 +0,0 @@ -import { expect, test } from "bun:test"; -import * as Json from "@moq/json"; -import { Track } from "@moq/net"; -import { Effect } from "@moq/signals"; -import { JsonProducer } from "./json.ts"; - -test("seeds late subscribers with the current value and fans out updates", async () => { - const producer = new JsonProducer<{ title?: string; count?: number }>(); - - // Set a value before anyone subscribes: it is retained, not lost. - producer.update({ title: "hello" }); - expect(producer.value).toEqual({ title: "hello" }); - - const effect = new Effect(); - const track = new Track("meta.json"); - producer.serve(track, effect); - const consumer = new Json.Consumer<{ title?: string; count?: number }>(track); - - // A new subscriber is seeded with the current value. - expect(await consumer.next()).toEqual({ title: "hello" }); - - // Subsequent updates fan out to the subscriber. - producer.update({ title: "world" }); - expect(await consumer.next()).toEqual({ title: "world" }); - - // Closing the effect finishes the subscription: it's dropped from the fan-out and the track - // ends, so further updates never reach it. - effect.close(); - producer.update({ title: "after close" }); - expect(await consumer.next()).toBeUndefined(); -}); - -test("mutate composes from the last value", async () => { - const producer = new JsonProducer>({ initial: {} }); - - // mutate works before any update because of the configured initial value. - producer.mutate((v) => { - v.a = 1; - }); - producer.mutate((v) => { - v.b = 2; - }); - - expect(producer.value).toEqual({ a: 1, b: 2 }); -}); - -test("mutate without a value or initial throws", () => { - const producer = new JsonProducer>(); - expect(() => producer.mutate(() => {})).toThrow(); -}); diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 76ec147a7d..9acf21ae50 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -5,7 +5,6 @@ import type * as Moq from "@moq/net"; import { Path } from "@moq/net"; import { Effect, type Getter, Signal } from "@moq/signals"; -import { JsonConsumer, type SubscribeJsonOptions } from "./json"; import { toHang } from "./msf"; /** Consumes a custom track once subscribed, scoped to the subscription's lifetime. */ @@ -197,8 +196,26 @@ export class Broadcast { /** * Subscribe to a custom track within this broadcast, following the active broadcast across * reconnects. `consume` runs with a freshly-subscribed track and a subscription-scoped effect - * each time a broadcast becomes active (re-running on reconnect). This is the low-level escape - * hatch for arbitrary payloads; see {@link subscribeJson} for a JSON convenience. + * each time a broadcast becomes active (re-running on reconnect). + * + * For a JSON track, wrap the track with a `@moq/json` `Consumer` and read it in a spawned loop + * (e.g. into a Signal). An application advertises the track in its own catalog section, which it + * reads back from {@link catalog} (unknown sections pass through the loose schema): + * + * ```ts + * import * as Json from "@moq/json"; + * const scte35 = new Signal<{ splices: number[] } | undefined>(undefined); + * broadcast.subscribeTrack("scte35.json", Catalog.PRIORITY.catalog, (track, effect) => { + * const consumer = new Json.Consumer<{ splices: number[] }>(track); + * effect.spawn(async () => { + * for (;;) { + * const next = await Promise.race([effect.cancel, consumer.next()]); + * if (next === undefined) break; + * scte35.set(next); + * } + * }); + * }); + * ``` * * Returns a function to stop subscribing; also stopped when this broadcast closes. */ @@ -217,25 +234,6 @@ export class Broadcast { return () => signals.close(); } - /** - * Subscribe to a custom JSON track within this broadcast (e.g. a `meta.json` track that an - * application advertises in its own catalog section). - * - * Follows the active broadcast across reconnects and exposes the latest reconstructed value as a - * Signal on the returned consumer. The consumer is closed when this broadcast closes; close it - * sooner via its `close()` to stop subscribing. - * - * ```ts - * const meta = broadcast.subscribeJson<{ title: string }>("meta.json"); - * meta.value.subscribe((v) => console.log("meta", v)); - * ``` - */ - subscribeJson(name: string, options?: SubscribeJsonOptions): JsonConsumer { - const consumer = new JsonConsumer(this.active, name, options); - this.signals.cleanup(() => consumer.close()); - return consumer; - } - close() { this.signals.close(); } diff --git a/js/watch/src/index.ts b/js/watch/src/index.ts index 2fa90e3b19..a3d7c46a01 100644 --- a/js/watch/src/index.ts +++ b/js/watch/src/index.ts @@ -1,4 +1,5 @@ export * as Hang from "@moq/hang"; +export * as Json from "@moq/json"; export * as Net from "@moq/net"; /** @deprecated Use `Net` instead. */ export * as Lite from "@moq/net"; @@ -6,7 +7,6 @@ export * as Signals from "@moq/signals"; export * as Audio from "./audio"; export * from "./backend"; export * from "./broadcast"; -export * from "./json"; export * as Mse from "./mse"; export * from "./sync"; export * as Video from "./video"; diff --git a/js/watch/src/json.test.ts b/js/watch/src/json.test.ts deleted file mode 100644 index 449e9649ab..0000000000 --- a/js/watch/src/json.test.ts +++ /dev/null @@ -1,91 +0,0 @@ -import { expect, test } from "bun:test"; -import * as Json from "@moq/json"; -import { Broadcast as MoqBroadcast } from "@moq/net"; -import { type Getter, Signal } from "@moq/signals"; -import { JsonConsumer } from "./json.ts"; - -// Resolve once the signal holds a defined value. -function nextDefined(signal: Getter): Promise { - return waitFor(signal, (value): value is T => value !== undefined); -} - -// Resolve once the signal's value satisfies the predicate. -function waitFor(signal: Getter, predicate: (value: T) => value is U): Promise { - return new Promise((resolve) => { - const dispose = signal.subscribe((value) => { - if (!predicate(value)) return; - dispose(); - resolve(value); - }); - }); -} - -test("reads a custom JSON track from the active broadcast", async () => { - const broadcast = new MoqBroadcast(); - const active = new Signal(broadcast); - - const consumer = new JsonConsumer<{ title: string }>(active, "meta.json"); - - // The consumer subscribes; serve the request from the producer side. - const request = await broadcast.requested(); - if (!request) throw new Error("expected a track request"); - expect(request.track.name).toBe("meta.json"); - const producer = new Json.Producer<{ title: string }>(request.track); - producer.update({ title: "hello" }); - - expect(await nextDefined(consumer.value)).toEqual({ title: "hello" }); - - consumer.close(); -}); - -test("clears the value when the broadcast goes away", async () => { - const broadcast = new MoqBroadcast(); - const active = new Signal(broadcast); - - const consumer = new JsonConsumer<{ title: string }>(active, "meta.json"); - - const request = await broadcast.requested(); - if (!request) throw new Error("expected a track request"); - const producer = new Json.Producer<{ title: string }>(request.track); - producer.update({ title: "hello" }); - await nextDefined(consumer.value); - - // Dropping the active broadcast tears down the subscription and clears the value. - await new Promise((resolve) => { - const dispose = consumer.value.subscribe((value) => { - if (value !== undefined) return; - dispose(); - resolve(); - }); - active.set(undefined); - }); - expect(consumer.value.peek()).toBeUndefined(); - - consumer.close(); -}); - -test("resubscribes when the active broadcast switches", async () => { - const first = new MoqBroadcast(); - const active = new Signal(first); - - const consumer = new JsonConsumer<{ title: string }>(active, "meta.json"); - - const req1 = await first.requested(); - if (!req1) throw new Error("expected a track request on the first broadcast"); - new Json.Producer<{ title: string }>(req1.track).update({ title: "first" }); - expect(await nextDefined(consumer.value)).toEqual({ title: "first" }); - - // Switching to a new broadcast resubscribes and reads from the new track. - const second = new MoqBroadcast(); - active.set(second); - - const req2 = await second.requested(); - if (!req2) throw new Error("expected a track request on the second broadcast"); - expect(req2.track.name).toBe("meta.json"); - new Json.Producer<{ title: string }>(req2.track).update({ title: "second" }); - - const value = await waitFor(consumer.value, (v): v is { title: string } => v?.title === "second"); - expect(value).toEqual({ title: "second" }); - - consumer.close(); -}); diff --git a/js/watch/src/json.ts b/js/watch/src/json.ts deleted file mode 100644 index b4b0cbc2d2..0000000000 --- a/js/watch/src/json.ts +++ /dev/null @@ -1,61 +0,0 @@ -import * as Catalog from "@moq/hang/catalog"; -import * as Json from "@moq/json"; -import type * as Moq from "@moq/net"; -import { Effect, type Getter, Signal } from "@moq/signals"; - -/** Options for {@link Broadcast.subscribeJson}, extending the {@link Json.Config} for the value. */ -export type SubscribeJsonOptions = Json.Config & { - /** - * Subscription priority. Defaults to {@link Catalog.PRIORITY.catalog} so metadata arrives ahead - * of media. - */ - priority?: number; -}; - -/** - * Consumes a custom JSON track from a broadcast, following the active broadcast across reconnects. - * - * The latest reconstructed value is exposed as a {@link Signal} via {@link value}. Call - * {@link close} when done (or let the owning watch {@link Broadcast} close it). - */ -export class JsonConsumer { - /** The latest reconstructed value, or `undefined` before the first frame or while offline. */ - readonly value = new Signal(undefined); - - #signals = new Effect(); - - /** Subscribe to `name` on whatever broadcast `active` currently holds. */ - constructor(active: Getter, name: string, options?: SubscribeJsonOptions) { - const priority = options?.priority ?? Catalog.PRIORITY.catalog; - - this.#signals.run((effect) => { - const broadcast = effect.get(active); - if (!broadcast) return; - - const track = broadcast.subscribe(name, priority); - effect.cleanup(() => track.close()); - - // Clear the value when this subscription tears down so a stale value from a previous - // broadcast doesn't linger after a reconnect. - effect.cleanup(() => this.value.set(undefined)); - - const consumer = new Json.Consumer(track, options); - effect.spawn(async () => { - try { - for (;;) { - const next = await Promise.race([effect.cancel, consumer.next()]); - if (next === undefined) break; - this.value.set(next); - } - } catch (err) { - console.warn("error fetching json track", name, err); - } - }); - }); - } - - /** Stop subscribing and release resources. */ - close(): void { - this.#signals.close(); - } -} From c840e0f38845c8239992f2474830e7605bf15523 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 01:15:40 +0000 Subject: [PATCH 6/7] Fix publish test CI; isolate fan-out failures; doc fixes - CI: the publish package has no headless-testable units left (Broadcast transitively imports an AudioWorklet bun can't load), so `bun test` exited 1 on "no tests found". Use `--pass-with-no-tests`. - @moq/json Source.update now isolates per-subscriber failures during fan-out so one broken track doesn't stop updates to the others. - Add a module-level doc to the @moq/json entrypoint. - Drop dead `@moq/json` doc links (there is no @moq/json doc page); keep it as inline code. --- doc/concept/layer/hang.md | 2 +- doc/lib/js/@moq/publish.md | 4 ++-- doc/lib/js/@moq/watch.md | 2 +- js/json/src/index.ts | 8 ++++++++ js/json/src/source.ts | 16 +++++++++++++++- js/publish/package.json | 2 +- 6 files changed, 28 insertions(+), 6 deletions(-) diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index baefc7b514..3d27dbbfd0 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -87,7 +87,7 @@ This keeps application-specific sections in the application layer while the base A custom catalog section can carry its payload inline (for low-rate metadata), or it can reference a separate track in the same broadcast (for a stream of data, e.g. a `meta.json` track or an SCTE-35 event track). The relay treats such a track like any other; only the publisher and consumer give it meaning. -The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support. Each exposes a low-level track hook and re-exports [`@moq/json`](/lib/js/@moq/hang) so the application encodes the payload itself: +The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support. Each exposes a low-level track hook and re-exports `@moq/json` so the application encodes the payload itself: - **Publish**: `broadcast.publishTrack(name, serve)` runs `serve(track, effect)` per subscriber. For JSON, serve each track from a shared `Json.Source` (the same fan-out producer the catalog uses, seeding late joiners with the latest value). Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`. - **Watch**: `broadcast.subscribeTrack(name, priority, consume)` follows the active broadcast across reconnects. For JSON, wrap the track in a `Json.Consumer` inside `consume`. Read your section back from `broadcast.catalog` (unknown sections pass through the loose schema). diff --git a/doc/lib/js/@moq/publish.md b/doc/lib/js/@moq/publish.md index da8c77ca68..765394ab48 100644 --- a/doc/lib/js/@moq/publish.md +++ b/doc/lib/js/@moq/publish.md @@ -133,8 +133,8 @@ Beyond audio and video, you can publish arbitrary application tracks within the same broadcast (no separate broadcast needed). `publishTrack(name, serve)` runs `serve(track, effect)` for each subscriber; it rejects the built-in track names (catalog/audio/video). Encode the payload yourself with the re-exported -[`@moq/json`](/lib/js/@moq/hang): a `Json.Source` is the same fan-out producer the -catalog uses, seeding late joiners with the latest value. +`@moq/json`: a `Json.Source` is the same fan-out producer the catalog uses, +seeding late joiners with the latest value. `publishTrack` does not touch the catalog; advertise the track by writing your own section to `broadcast.catalog` (the [catalog root](/concept/layer/hang#extensions) diff --git a/doc/lib/js/@moq/watch.md b/doc/lib/js/@moq/watch.md index 4b0a4356c7..dac976bc55 100644 --- a/doc/lib/js/@moq/watch.md +++ b/doc/lib/js/@moq/watch.md @@ -147,7 +147,7 @@ catalog section (the [catalog root](/concept/layer/hang#extensions) is a loose object, so unknown sections pass through to `broadcast.catalog`). `subscribeTrack(name, priority, consume)` follows the active broadcast across reconnects and runs `consume(track, effect)` each time it becomes active. Decode -the payload yourself with the re-exported [`@moq/json`](/lib/js/@moq/hang): +the payload yourself with the re-exported `@moq/json`: ```typescript import { Json } from "@moq/watch"; diff --git a/js/json/src/index.ts b/js/json/src/index.ts index 03418979e6..2ed9bf6ebf 100644 --- a/js/json/src/index.ts +++ b/js/json/src/index.ts @@ -1,3 +1,11 @@ +/** + * Snapshot/delta JSON publishing over MoQ tracks using RFC 7396 JSON Merge Patch: a per-track + * {@link Producer}/{@link Consumer} pair and a {@link Source} that fans one value out to many + * subscribers. + * + * @module + */ + export { Consumer } from "./consumer.ts"; export { type Diff, deepEqual, diff, merge } from "./diff.ts"; export { type Config, Producer } from "./producer.ts"; diff --git a/js/json/src/source.ts b/js/json/src/source.ts index 9eebbf87d5..f7e9494506 100644 --- a/js/json/src/source.ts +++ b/js/json/src/source.ts @@ -34,7 +34,21 @@ export class Source { /** Replace the value; the result is published to all current subscribers. */ update(value: T): void { this.#value = value; - for (const output of this.#outputs) output.update(value); + for (const output of this.#outputs) { + // Isolate per-subscriber failures: a bad track (e.g. closed mid-update) must not stop the + // fan-out to the others. Drop it and keep going. + try { + output.update(value); + } catch (err) { + this.#outputs.delete(output); + try { + output.finish(); + } catch { + // Already broken; nothing more to do. + } + console.warn("dropping failed json subscriber during fan-out", err); + } + } } /** diff --git a/js/publish/package.json b/js/publish/package.json index a6eaaa8ba9..8f4570a073 100644 --- a/js/publish/package.json +++ b/js/publish/package.json @@ -21,7 +21,7 @@ "scripts": { "build": "rimraf dist && vite build && tsc && bun ../common/package.ts", "check": "tsc --noEmit", - "test": "bun test", + "test": "bun test --pass-with-no-tests", "release": "bun ../common/release.ts" }, "dependencies": { From 7d9746499c2a5858975575eb5c8635d3297f45f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 03:16:37 +0000 Subject: [PATCH 7/7] Merge json.Source into json.Producer A Producer constructed with a track is unchanged (the leaf encoder, byte-for-byte identical, so the wire-format vectors still pass). A Producer constructed without a track is the fan-out source: it retains the value, seeds late joiners, and fans updates out to subscribers attached with the new `serve(track, effect)`. The constructor overloads on `instanceof Track`, so `new Producer(track, config)` (existing) and `new Producer(config)` (fan-out) both read cleanly. - Remove the separate `Source` class; `CatalogProducer` aliases `Producer` and the broadcast constructs `new Producer({ initial: {} })`. - Move the fan-out tests to producer.test.ts and add a serve()-on-leaf guard test. - Docs use a track-less `Json.Producer` instead of `Json.Source`. --- doc/concept/layer/hang.md | 2 +- doc/lib/js/@moq/publish.md | 6 +- js/json/src/index.ts | 7 +- .../src/{source.test.ts => producer.test.ts} | 17 +++- js/json/src/producer.ts | 99 ++++++++++++++++--- js/json/src/source.ts | 83 ---------------- js/publish/src/broadcast.ts | 16 +-- js/publish/src/catalog.ts | 7 +- 8 files changed, 119 insertions(+), 118 deletions(-) rename js/json/src/{source.test.ts => producer.test.ts} (74%) delete mode 100644 js/json/src/source.ts diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index 3d27dbbfd0..648766c181 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -89,7 +89,7 @@ A custom catalog section can carry its payload inline (for low-rate metadata), o The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support. Each exposes a low-level track hook and re-exports `@moq/json` so the application encodes the payload itself: -- **Publish**: `broadcast.publishTrack(name, serve)` runs `serve(track, effect)` per subscriber. For JSON, serve each track from a shared `Json.Source` (the same fan-out producer the catalog uses, seeding late joiners with the latest value). Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`. +- **Publish**: `broadcast.publishTrack(name, serve)` runs `serve(track, effect)` per subscriber. For JSON, serve each track from a shared track-less `Json.Producer` (the same fan-out producer the catalog uses, seeding late joiners with the latest value). Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`. - **Watch**: `broadcast.subscribeTrack(name, priority, consume)` follows the active broadcast across reconnects. For JSON, wrap the track in a `Json.Consumer` inside `consume`. Read your section back from `broadcast.catalog` (unknown sections pass through the loose schema). So an application supports something like SCTE-35 entirely in its own code: publish an `scte35` section (and optionally a track) on one side, read it on the other, without hang, `@moq/publish`, or `@moq/watch` knowing anything about SCTE-35. diff --git a/doc/lib/js/@moq/publish.md b/doc/lib/js/@moq/publish.md index 765394ab48..0b1a2ef30e 100644 --- a/doc/lib/js/@moq/publish.md +++ b/doc/lib/js/@moq/publish.md @@ -133,8 +133,8 @@ Beyond audio and video, you can publish arbitrary application tracks within the same broadcast (no separate broadcast needed). `publishTrack(name, serve)` runs `serve(track, effect)` for each subscriber; it rejects the built-in track names (catalog/audio/video). Encode the payload yourself with the re-exported -`@moq/json`: a `Json.Source` is the same fan-out producer the catalog uses, -seeding late joiners with the latest value. +`@moq/json`: a track-less `Json.Producer` is the same fan-out producer the catalog +uses, seeding late joiners with the latest value. `publishTrack` does not touch the catalog; advertise the track by writing your own section to `broadcast.catalog` (the [catalog root](/concept/layer/hang#extensions) @@ -144,7 +144,7 @@ like an `scte35` section with no hang-specific support: ```typescript import { Json } from "@moq/publish"; -const scte35 = new Json.Source<{ splices: number[] }>({ initial: { splices: [] } }); +const scte35 = new Json.Producer<{ splices: number[] }>({ initial: { splices: [] } }); broadcast.publishTrack("scte35.json", (track, effect) => scte35.serve(track, effect)); broadcast.catalog.mutate((c) => { c.scte35 = { track: "scte35.json" }; diff --git a/js/json/src/index.ts b/js/json/src/index.ts index 2ed9bf6ebf..3d3fba6710 100644 --- a/js/json/src/index.ts +++ b/js/json/src/index.ts @@ -1,7 +1,7 @@ /** - * Snapshot/delta JSON publishing over MoQ tracks using RFC 7396 JSON Merge Patch: a per-track - * {@link Producer}/{@link Consumer} pair and a {@link Source} that fans one value out to many - * subscribers. + * Snapshot/delta JSON publishing over MoQ tracks using RFC 7396 JSON Merge Patch. A + * {@link Producer} writes a JSON value to one track or fans it out to many; a {@link Consumer} + * reconstructs the value on the other side. * * @module */ @@ -9,4 +9,3 @@ export { Consumer } from "./consumer.ts"; export { type Diff, deepEqual, diff, merge } from "./diff.ts"; export { type Config, Producer } from "./producer.ts"; -export { Source } from "./source.ts"; diff --git a/js/json/src/source.test.ts b/js/json/src/producer.test.ts similarity index 74% rename from js/json/src/source.test.ts rename to js/json/src/producer.test.ts index 4d08d9ffc6..b6e346d7ac 100644 --- a/js/json/src/source.test.ts +++ b/js/json/src/producer.test.ts @@ -2,10 +2,10 @@ import { expect, test } from "bun:test"; import { Track } from "@moq/net"; import { Effect } from "@moq/signals"; import { Consumer } from "./consumer.ts"; -import { Source } from "./source.ts"; +import { Producer } from "./producer.ts"; -test("seeds subscribers and fans out edits", async () => { - const source = new Source>({ initial: {} }); +test("a track-less producer seeds subscribers and fans out edits", async () => { + const source = new Producer>({ initial: {} }); // Edit before anyone subscribes: the value is retained, not lost. source.mutate((v) => { @@ -32,7 +32,7 @@ test("seeds subscribers and fans out edits", async () => { }); test("a reconnecting subscriber is seeded with the full current value", async () => { - const source = new Source>({ initial: {} }); + const source = new Producer>({ initial: {} }); source.mutate((v) => { v.video = { renditions: {} }; v.scte35 = { splices: [] }; @@ -55,6 +55,13 @@ test("a reconnecting subscriber is seeded with the full current value", async () }); test("mutate without a value or initial throws", () => { - const source = new Source>(); + const source = new Producer>(); expect(() => source.mutate(() => {})).toThrow(); }); + +test("serve() throws on a track-bound producer", () => { + const producer = new Producer>(new Track("meta.json")); + const effect = new Effect(); + expect(() => producer.serve(new Track("other"), effect)).toThrow(); + effect.close(); +}); diff --git a/js/json/src/producer.ts b/js/json/src/producer.ts index c014f3256e..d4c0e794ef 100644 --- a/js/json/src/producer.ts +++ b/js/json/src/producer.ts @@ -1,4 +1,5 @@ -import type * as Moq from "@moq/net"; +import * as Moq from "@moq/net"; +import type { Effect } from "@moq/signals"; import type * as z from "zod/mini"; import { deepEqual, diff } from "./diff.ts"; @@ -25,23 +26,72 @@ export interface Config { initial?: T; } -/** Publishes a JSON value over a track, choosing snapshots and deltas automatically. */ +/** + * Publishes a JSON value as snapshots and deltas, chosen automatically. + * + * Construct it two ways: + * + * - **With a track** (`new Producer(track, config)`): writes directly to that one track. + * - **Without a track** (`new Producer(config)`): retains the value and fans it out to any number of + * subscription tracks attached with {@link serve}, seeding late joiners with the current value. + * This backs the hang catalog and is how an application publishes its own custom tracks. + */ export class Producer { - #track: Moq.Track; #config: Config; + // Leaf mode: writes snapshots/deltas straight to a single track. + #track?: Moq.Track; #group?: Moq.Group; #last?: unknown; #groupBytes = 0; #groupFrames = 0; - constructor(track: Moq.Track, config: Config = {}) { - this.#track = track; - this.#config = config; + // Fan-out mode: retains the value and serves a child (leaf) Producer per subscriber. + #outputs?: Set>; + #value?: T; + + /** Create a track-less, fan-out producer; attach subscribers with {@link serve}. */ + constructor(config?: Config); + /** Create a producer that writes directly to `track`. */ + constructor(track: Moq.Track, config?: Config); + constructor(trackOrConfig?: Moq.Track | Config, config: Config = {}) { + if (trackOrConfig instanceof Moq.Track) { + this.#track = trackOrConfig; + this.#config = config; + } else { + this.#config = trackOrConfig ?? {}; + this.#outputs = new Set(); + this.#value = this.#config.initial; + } + } + + /** The current value, or `undefined` if nothing has been published yet. */ + get value(): T | undefined { + return this.#track ? (this.#last as T | undefined) : this.#value; } /** Publish a new value, emitting a snapshot or delta automatically. No-op if unchanged. */ update(value: T): void { + if (!this.#track) { + // Fan-out: retain the value and forward it to every subscriber. Isolate per-subscriber + // failures so one broken track (e.g. closed mid-update) doesn't stop the others. + this.#value = value; + for (const output of this.#outputs ?? []) { + try { + output.update(value); + } catch (err) { + this.#outputs?.delete(output); + try { + output.finish(); + } catch { + // Already broken; nothing more to do. + } + console.warn("dropping failed json subscriber during fan-out", err); + } + } + return; + } + const valid = this.#config.schema ? this.#config.schema.parse(value) : value; // Serialize once; parse it back to a normalized JSON value for diffing and comparison @@ -57,7 +107,7 @@ export class Producer { this.#groupBytes += delta.length; this.#groupFrames += 1; } else { - this.#snapshot(snapshot); + this.#snapshot(this.#track, snapshot); } this.#last = json; @@ -83,7 +133,7 @@ export class Producer { mutate(fn: (value: T) => void): void { // Start from the last-published value, falling back to the configured initial value. We // don't invent an empty object: mutating with nothing to start from is a usage error. - const base = this.#last ?? this.#config.initial; + const base = (this.#track ? this.#last : this.#value) ?? this.#config.initial; if (base === undefined) { throw new Error("mutate() requires a prior update() or `initial` in the config"); } @@ -93,8 +143,35 @@ export class Producer { this.update(value); } - /** Finish the track, closing any open group. */ + /** + * Serve a subscription request: seed the track with the current value, then forward updates. + * + * Only available on a track-less (fan-out) producer. The subscriber is removed and finished when + * `effect` is cleaned up. + */ + serve(track: Moq.Track, effect: Effect): void { + if (!this.#outputs) { + throw new Error("serve() is only available on a track-less Producer"); + } + + const output = new Producer(track, this.#config); + if (this.#value !== undefined) output.update(this.#value); + + this.#outputs.add(output); + effect.cleanup(() => { + this.#outputs?.delete(output); + output.finish(); + }); + } + + /** Finish: close the track (leaf) or finish every subscriber (fan-out). */ finish(): void { + if (!this.#track) { + for (const output of this.#outputs ?? []) output.finish(); + this.#outputs?.clear(); + return; + } + this.#group?.close(); this.#group = undefined; this.#track.close(); @@ -117,11 +194,11 @@ export class Producer { return delta; } - #snapshot(snapshot: Uint8Array): void { + #snapshot(track: Moq.Track, snapshot: Uint8Array): void { // The previous group is complete; no more frames will be appended to it. this.#group?.close(); - const group = this.#track.appendGroup(); + const group = track.appendGroup(); group.writeFrame(snapshot); this.#groupBytes = snapshot.length; this.#groupFrames = 1; diff --git a/js/json/src/source.ts b/js/json/src/source.ts deleted file mode 100644 index f7e9494506..0000000000 --- a/js/json/src/source.ts +++ /dev/null @@ -1,83 +0,0 @@ -import type * as Moq from "@moq/net"; -import type { Effect } from "@moq/signals"; - -import { type Config, Producer } from "./producer.ts"; - -/** - * A stable JSON value that fans out to on-demand subscription tracks. - * - * Unlike a per-track {@link Producer}, this exists independently of any subscription: set the value - * at any time with {@link update} or {@link mutate}, and each subscriber (including a relay that - * reconnects) is seeded with the current value before receiving updates. Multiple independent owners - * can share one instance and each edit only their own keys via {@link mutate}, so their sections - * compose instead of clobbering one another. - * - * This backs the hang catalog, and an application can use it for its own custom tracks (serve it - * from a publish `Broadcast.publishTrack` handler). - */ -export class Source { - #value: T | undefined; - #outputs = new Set>(); - #config: Config; - - /** Create a source, optionally seeding an initial value and per-track {@link Config}. */ - constructor(config: Config = {}) { - this.#config = config; - this.#value = config.initial; - } - - /** The current value, or `undefined` if nothing has been published yet. */ - get value(): T | undefined { - return this.#value; - } - - /** Replace the value; the result is published to all current subscribers. */ - update(value: T): void { - this.#value = value; - for (const output of this.#outputs) { - // Isolate per-subscriber failures: a bad track (e.g. closed mid-update) must not stop the - // fan-out to the others. Drop it and keep going. - try { - output.update(value); - } catch (err) { - this.#outputs.delete(output); - try { - output.finish(); - } catch { - // Already broken; nothing more to do. - } - console.warn("dropping failed json subscriber during fan-out", err); - } - } - } - - /** - * Mutate the current value in place and publish the result. - * - * The callback receives a deep clone of the last value (falling back to the configured `initial`, - * throwing if neither exists). Edit it in place; on return the result is published via - * {@link update}. - */ - mutate(fn: (value: T) => void): void { - const base = this.#value ?? this.#config.initial; - if (base === undefined) { - throw new Error("mutate() requires a prior update() or `initial` in the config"); - } - - const value = structuredClone(base) as T; - fn(value); - this.update(value); - } - - /** Serve a subscription request: seed it with the current value, then forward updates. */ - serve(track: Moq.Track, effect: Effect): void { - const output = new Producer(track, this.#config); - if (this.#value !== undefined) output.update(this.#value); - - this.#outputs.add(output); - effect.cleanup(() => { - this.#outputs.delete(output); - output.finish(); - }); - } -} diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index ad536972b5..a773ebf234 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -1,5 +1,5 @@ import * as Catalog from "@moq/hang/catalog"; -import { Source } from "@moq/json"; +import { Producer } from "@moq/json"; import * as Moq from "@moq/net"; import { Effect, Signal } from "@moq/signals"; import * as Audio from "./audio"; @@ -30,7 +30,7 @@ export class Broadcast { // The catalog, editable at any time regardless of whether anyone is subscribed. The base // `video`/`audio` sections are kept in sync from the encoders; an application adds its own root // sections (e.g. `scte35`) by mutating it too. - readonly catalog: CatalogProducer = new Source({ initial: {} }); + readonly catalog: CatalogProducer = new Producer({ initial: {} }); // Handlers for custom tracks registered via `publishTrack`, keyed by track name. Persists across // reconnects so a new `Moq.Broadcast` still serves them. @@ -143,14 +143,14 @@ export class Broadcast { * subscriptions, nor touch the catalog. Throws if `name` collides with a built-in track * (catalog/audio/video), since those are served first and the handler would never run. * - * For a JSON track, serve each track from a `@moq/json` `Source` (the same fan-out producer the - * catalog uses, seeding late joiners with the latest value). Advertise the track by writing your - * own section to {@link catalog}, e.g. to support a custom `scte35` section with no hang-specific - * support: + * For a JSON track, serve each track from a track-less `@moq/json` `Producer` (the same fan-out + * producer the catalog uses, seeding late joiners with the latest value). Advertise the track by + * writing your own section to {@link catalog}, e.g. to support a custom `scte35` section with no + * hang-specific support: * * ```ts - * import { Source } from "@moq/json"; - * const scte35 = new Source({ initial: { splices: [] } }); + * import { Producer } from "@moq/json"; + * const scte35 = new Producer({ initial: { splices: [] } }); * broadcast.publishTrack("scte35.json", (track, effect) => scte35.serve(track, effect)); * broadcast.catalog.mutate((c) => { c.scte35 = { track: "scte35.json" }; }); * scte35.update({ splices: [42] }); diff --git a/js/publish/src/catalog.ts b/js/publish/src/catalog.ts index c4485ca3cd..cee515cd65 100644 --- a/js/publish/src/catalog.ts +++ b/js/publish/src/catalog.ts @@ -1,11 +1,12 @@ import type * as Catalog from "@moq/hang/catalog"; -import type { Source } from "@moq/json"; +import type { Producer } from "@moq/json"; /** - * The broadcast catalog: a {@link Source} of the catalog {@link Catalog.Root}, seeded empty. + * The broadcast catalog: a track-less {@link Producer} of the catalog {@link Catalog.Root}, seeded + * empty. * * Edit it at any time with `mutate` (the base `video`/`audio` sections are kept in sync by the * encoders; an application adds its own root sections, e.g. `scte35`, the same way). Each * subscriber, including a relay that reconnects, is seeded with the current catalog before updates. */ -export type CatalogProducer = Source; +export type CatalogProducer = Producer;