diff --git a/doc/concept/layer/hang.md b/doc/concept/layer/hang.md index 0753b749b6..648766c181 100644 --- a/doc/concept/layer/hang.md +++ b/doc/concept/layer/hang.md @@ -83,6 +83,17 @@ 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. +### Custom tracks + +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` 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 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. + ## 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..0b1a2ef30e 100644 --- a/doc/lib/js/@moq/publish.md +++ b/doc/lib/js/@moq/publish.md @@ -127,6 +127,34 @@ broadcast.video.device.set("screen"); 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). `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 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) +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 +import { Json } from "@moq/publish"; + +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" }; +}); +scte35.update({ splices: [42] }); +``` + +The component exposes everything via its `broadcast` property +(`el.broadcast.publishTrack(...)`). + ## 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..dac976bc55 100644 --- a/doc/lib/js/@moq/watch.md +++ b/doc/lib/js/@moq/watch.md @@ -139,6 +139,39 @@ el.catalog = myCatalog; > `"msf"`) tears down the previous fetch loop, which clears `catalog`. Set the > catalog *after* switching to `"manual"`, not before. +### Custom tracks and catalog sections + +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`). +`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`: + +```typescript +import { Json } from "@moq/watch"; +import { Signals } from "@moq/watch"; + +// 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); + } + }); +}); +``` + +The component exposes everything via its `broadcast` property +(`el.broadcast.subscribeTrack(...)`). + ## UI Overlay Import `@moq/watch/ui` for a Web Component overlay with buffering indicator, stats panel, and playback controls: diff --git a/js/json/src/index.ts b/js/json/src/index.ts index ea6ff76d37..3d3fba6710 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 + * {@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 + */ + 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/producer.test.ts b/js/json/src/producer.test.ts new file mode 100644 index 0000000000..b6e346d7ac --- /dev/null +++ b/js/json/src/producer.test.ts @@ -0,0 +1,67 @@ +import { expect, test } from "bun:test"; +import { Track } from "@moq/net"; +import { Effect } from "@moq/signals"; +import { Consumer } from "./consumer.ts"; +import { Producer } from "./producer.ts"; + +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) => { + 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 Producer>({ 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 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/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": { diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index ae2638d578..a773ebf234 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -1,8 +1,9 @@ import * as Catalog from "@moq/hang/catalog"; +import { Producer } 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 type { CatalogProducer } from "./catalog"; import * as Video from "./video"; export type BroadcastProps = { @@ -13,6 +14,9 @@ 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; + export class Broadcast { static readonly CATALOG_TRACK = "catalog.json"; @@ -25,8 +29,21 @@ 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 Producer({ initial: {} }); + + // 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(); + + // 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(); @@ -100,15 +117,55 @@ 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 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 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 { 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] }); + * ``` + */ + 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); + }; + } + 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 7cad29cb90..cee515cd65 100644 --- a/js/publish/src/catalog.ts +++ b/js/publish/src/catalog.ts @@ -1,38 +1,12 @@ 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 type { Producer } from "@moq/json"; /** - * A stable catalog producer that fans out to on-demand subscription tracks. + * The broadcast catalog: a track-less {@link Producer} 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 {@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 - * 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 { - #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 type CatalogProducer = Producer; diff --git a/js/publish/src/index.ts b/js/publish/src/index.ts index 082e2f005f..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"; diff --git a/js/watch/src/broadcast.ts b/js/watch/src/broadcast.ts index 2b05cc6f4a..9acf21ae50 100644 --- a/js/watch/src/broadcast.ts +++ b/js/watch/src/broadcast.ts @@ -7,6 +7,9 @@ import { Effect, type Getter, Signal } from "@moq/signals"; 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; @@ -190,6 +193,47 @@ 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). + * + * 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. + */ + 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(); + } + close() { this.signals.close(); } diff --git a/js/watch/src/index.ts b/js/watch/src/index.ts index 6103f721f7..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";