diff --git a/bench/browser/page/transport.ts b/bench/browser/page/transport.ts index 53f30ffea1a..ca6cc0c1dbc 100644 --- a/bench/browser/page/transport.ts +++ b/bench/browser/page/transport.ts @@ -32,11 +32,12 @@ async function publish(config: Transport): Promise { // Fixture generation stays outside every timed interval. const fixtures = Array.from({ length: config.frames }, (_, frame) => payload(config.size, frame)); const origin = new Moq.Origin.Producer(); - const producer = origin.createBroadcast(Moq.Path.from(config.path)); + const publication = origin.createBroadcast(Moq.Path.from(config.path)); + const producer = publication.broadcast; // Retention covers the whole run, so a slow consumer cannot lose a group to eviction. const track = producer.createTrack("data", { timescale: Moq.Time.Timescale.MILLI, maxAge: Milli(600_000) }); const demand = track.demand().wanted; - producer.announce(); + publication.announce(); const connection = await Moq.Connection.connect({ url: new URL(config.url), publish: origin.consume(), diff --git a/demo/carla/web/src/truck.ts b/demo/carla/web/src/truck.ts index 01d28e25715..d3fc13a4a82 100644 --- a/demo/carla/web/src/truck.ts +++ b/demo/carla/web/src/truck.ts @@ -416,7 +416,8 @@ export class Truck { const driverId = crypto.randomUUID().replaceAll("-", "").slice(0, 12); this.driverId.set(driverId); - const driverBroadcast = conn.createBroadcast(driverPath(this.name, driverId)); + const driverBroadcastPublication = conn.createBroadcast(driverPath(this.name, driverId)); + const driverBroadcast = driverBroadcastPublication.broadcast; effect.cleanup(() => { driverBroadcast.close(); this.driverId.set(undefined); @@ -429,7 +430,7 @@ export class Truck { effect.cleanup(() => producer.finish()); effect.run(this.#runControlTrack.bind(this, track, producer)); // A fresh epoch per lease: the truck binds its single driver to this path and epoch. - driverBroadcast.announce({ epoch: Moq.Epoch.mint() }); + driverBroadcastPublication.announce({ epoch: Moq.Epoch.mint() }); } #runSafetyControl(connection: Moq.Connection, effect: Moq.Signals.Effect) { @@ -438,9 +439,10 @@ export class Truck { if (!conn || !command) return; const driverId = `safety-${crypto.randomUUID().replaceAll("-", "").slice(0, 12)}`; - const broadcast = conn.createBroadcast(driverPath(this.name, driverId)); + const broadcastPublication = conn.createBroadcast(driverPath(this.name, driverId)); + const broadcast = broadcastPublication.broadcast; const safety = new SafetyCommandBroadcast(command, broadcast); - broadcast.announce({ epoch: Moq.Epoch.mint() }); + broadcastPublication.announce({ epoch: Moq.Epoch.mint() }); effect.cleanup(() => safety.close()); } diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index c6050cf7a59..ea1ccadb65d 100644 --- a/doc/lib/js/net.md +++ b/doc/lib/js/net.md @@ -25,16 +25,17 @@ const connection = await Moq.Connection.connect({ consume: origin, }); -const broadcast = origin.createBroadcast(Moq.Path.from("chat.room")); +const publication = origin.createBroadcast(Moq.Path.from("chat.room")); +const broadcast = publication.broadcast; const track = broadcast.createTrack("messages", { timescale: Moq.Time.Timescale.MILLI }); const group = track.appendGroup(); group.writeFrame({ payload: new TextEncoder().encode("hello"), timestamp: Moq.Time.Timestamp.now() }); group.close(); -broadcast.announce(); +publication.announce(); ``` -- **The origin holds the broadcasts, not the connection.** Closing a session unannounces them and leaves them created for the next one. A broadcast is invisible, locally and remotely, until `announce()`. -- **Answer claimed paths with `using`.** `dynamic(prefix)` claims a prefix and yields each requested path. Use `using prepared = request.prepare(epoch)`, fill its broadcast, then `prepared.publish()` to answer with a new publisher instance. Leaving the scope without publishing, including on a throw, refuses the request as unroutable. Explicit `prepared.close()` or closing its unpublished broadcast does the same; disposal after publishing leaves the returned producer live. Reject before preparing with `request.reject(err)`. A retained, unclosed prepared answer keeps the request pending; JavaScript garbage collection provides no prompt cleanup guarantee. +- **The origin holds the broadcasts, not the connection.** Closing a session unannounces them and leaves them created for the next one. A publication controls announcements; its `broadcast` fills tracks. The broadcast is invisible, locally and remotely, until the publication calls `announce()`. +- **Answer claimed paths with `using`.** `dynamic(prefix)` claims a prefix and yields each requested path. Use `using prepared = request.prepare(epoch)`, fill its broadcast, then `prepared.publish()` to answer with a new publisher instance and keep the `Publication` it returns. Leaving the scope without publishing, including on a throw, refuses the request as unroutable. Explicit `prepared.close()` or closing its unpublished broadcast does the same; disposal after publishing leaves the returned publication and its fill broadcast live. Reject before preparing with `request.reject(err)`. A retained, unclosed prepared answer keeps the request pending; JavaScript garbage collection provides no prompt cleanup guarantee. - **End content by closing tracks.** A broadcast has no finished state. When the input ends, close each track cleanly; the broadcast stays announced and keeps serving them until you `close()` it. How long to keep it up for late readers is yours to decide. - **Requests can pin an epoch.** `consumer.request(path, { epoch })` resolves only through a route announcing that [publisher epoch](/concept/moq-lite#publisher-epochs). The resolved broadcast's `epoch` names the route it came through, pinned or not. - **Follow restarts.** `origin.request(path, { announced: true })` follows whichever publisher instance serves the path: its `active` changes to a fresh broadcast on a restart, clears on an end, and resolves again on the next start, so read it reactively and close the request when done. Instances are never stitched, so subscribe again to each new broadcast. Another route of the same epoch taking over (epochs travel only on `moq-lite-07-wip`) keeps `active` as is, and each track subscription resumes there from the first group it lacks. A refusal ends the request. Add `epoch` to pin one instance; a plain `request(path)` keeps its accepted instance, can resolve blindly, and ends once that instance stops serving. To list what is live, iterate `announced(scope)`. @@ -47,7 +48,7 @@ broadcast.announce(); - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` and `@moq/flate` producers. - **Subscriber staleness is `maxDelay`.** It counts from a group's successor, by media time and by wall clock, whichever reaches it first. Passing the old `maxAge` key throws a `TypeError` naming `maxDelay`. Publisher retention is the separate `Track.Info.maxAge`. - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts). -- **Authorization is in band.** On moq-lite 07 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription or track listing the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Auth` it returns (`broadcast.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. A `Connection`'s `auth` handle (a getter on a pooled connection, following its URL) outlives its sessions: a token added there is presented on every reconnect and keeps riding the announcements that name it, with no re-add or re-announce. A one-shot session's `auth` ends with it, unless `Connection.connect` was given a `new Auth.Handle()` as `auth`. +- **Authorization is in band.** On moq-lite 07 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription or track listing the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Auth` it returns (`publication.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. A `Connection`'s `auth` handle (a getter on a pooled connection, following its URL) outlives its sessions: a token added there is presented on every reconnect and keeps riding the announcements that name it, with no re-add or re-announce. A one-shot session's `auth` ends with it, unless `Connection.connect` was given a `new Auth.Handle()` as `auth`. - **A graceful close waits.** `await connection.close()` withdraws announcements and drains for up to one second. Application delivery requires explicit `moq-lite-07-wip`; older Lite versions wait only for transport ACKs. IETF media streams are not drained. `abort()` ends immediately. Examples: diff --git a/js/e2ee/README.md b/js/e2ee/README.md index 6c5dff5166f..0c6471ed653 100644 --- a/js/e2ee/README.md +++ b/js/e2ee/README.md @@ -32,11 +32,11 @@ import { Time } from "@moq/net"; const credential = new Credential({ context, kid, secret }); const generation = credential.produce(); -const broadcast = origin.createBroadcast(await credential.path("meeting.hang")); -const video = await new Broadcast.Producer(broadcast, generation).createTrack("video", { +const publication = origin.createBroadcast(await credential.path("meeting.hang")); +const video = await new Broadcast.Producer(publication.broadcast, generation).createTrack("video", { timescale: Time.Timescale.MILLI, }); -broadcast.announce({ epoch: generation.epoch }); +publication.announce({ epoch: generation.epoch }); const group = video.appendGroup(); await group.writeFrame(Time.Timestamp.now(), frame); diff --git a/js/e2ee/src/index.ts b/js/e2ee/src/index.ts index e69235513e3..b2f4d0085d8 100644 --- a/js/e2ee/src/index.ts +++ b/js/e2ee/src/index.ts @@ -15,9 +15,9 @@ * ```ts * const credential = new Credential({ context, kid, secret }); * const generation = credential.produce(); - * const broadcast = origin.createBroadcast(await credential.path("meeting.hang")); - * const video = await new Broadcast.Producer(broadcast, generation).createTrack("video", { timescale }); - * broadcast.announce({ epoch: generation.epoch }); + * const publication = origin.createBroadcast(await credential.path("meeting.hang")); + * const video = await new Broadcast.Producer(publication.broadcast, generation).createTrack("video", { timescale }); + * publication.announce({ epoch: generation.epoch }); * * const group = video.appendGroup(); * await group.writeFrame(timestamp, frame); diff --git a/js/e2ee/src/transport.test.ts b/js/e2ee/src/transport.test.ts index 84588ae7242..a62643fd9ac 100644 --- a/js/e2ee/src/transport.test.ts +++ b/js/e2ee/src/transport.test.ts @@ -62,11 +62,12 @@ for (const [name, protocol, routed, timed] of transports) { const cred = credential(); const generation = cred.produce(); const path = await cred.path("meeting.hang"); - const broadcast = publisher.createBroadcast(path); + const broadcastPublication = publisher.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; const video = await new Protected.Producer(broadcast, generation).createTrack("video", { timescale: Time.Timescale.MILLI, }); - broadcast.announce({ epoch: generation.epoch }); + broadcastPublication.announce({ epoch: generation.epoch }); const group = video.appendGroup(); await group.writeFrame(ms(1), encoder.encode("keyframe")); @@ -117,11 +118,12 @@ for (const [name, protocol] of [ const cred = credential(); const generation = cred.produce(); const path = await cred.path("meeting.hang"); - const broadcast = publisher.createBroadcast(path); + const broadcastPublication = publisher.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; const audio = await new Protected.Producer(broadcast, generation).createTrack("audio", { timescale: Time.Timescale.MILLI, }); - broadcast.announce({ epoch: generation.epoch }); + broadcastPublication.announce({ epoch: generation.epoch }); // A group first, so the subscription is serving before datagrams, which are never cached. const group = audio.appendGroup(); diff --git a/js/hang/src/catalog/consumer.test.ts b/js/hang/src/catalog/consumer.test.ts index 8de5cca9a96..4e727477238 100644 --- a/js/hang/src/catalog/consumer.test.ts +++ b/js/hang/src/catalog/consumer.test.ts @@ -90,10 +90,11 @@ function referencing(broadcast: string): Root { // Watch the catalog of `room/alice`, published through an origin and requested from it. async function watchRequested(root: Root) { const origin = new Moq.Origin.Producer(); - const broadcast = origin.createBroadcast(Moq.Path.from("room/alice")); + const broadcastPublication = origin.createBroadcast(Moq.Path.from("room/alice")); + const broadcast = broadcastPublication.broadcast; const track = broadcast.createTrack(TRACK, { timescale: Moq.Time.Timescale.MILLI }); const producer = new Json.Snapshot.Producer({ track, deltas: false }); - broadcast.announce(); + broadcastPublication.announce(); const request = origin.request(Moq.Path.from("room/alice")); const active = request.active.peek(); if (!active) throw new Error("request did not resolve"); diff --git a/js/moq-boy/src/game.ts b/js/moq-boy/src/game.ts index 660b49cc72c..a70a0896ed4 100644 --- a/js/moq-boy/src/game.ts +++ b/js/moq-boy/src/game.ts @@ -317,10 +317,11 @@ export class Game { const viewerId = Math.random().toString(36).slice(2, 8); this.viewerId.set(viewerId); - const viewerBroadcast = origin.createBroadcast( + const viewerBroadcastPublication = origin.createBroadcast( Moq.Path.from(`${this.#viewerPrefix}/${this.sessionId}/${viewerId}`), ); - viewerBroadcast.announce({ epoch: Moq.Epoch.mint() }); + const viewerBroadcast = viewerBroadcastPublication.broadcast; + viewerBroadcastPublication.announce({ epoch: Moq.Epoch.mint() }); effect.cleanup(() => { viewerBroadcast.close(); this.viewerId.set(undefined); diff --git a/js/net/MIGRATE.md b/js/net/MIGRATE.md index 15ff507eb2d..868c8362119 100644 --- a/js/net/MIGRATE.md +++ b/js/net/MIGRATE.md @@ -4,6 +4,20 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Publication handles own announcement control + +`Origin.Producer.createBroadcast(path)` and `Origin.Prepared.publish()` return an `Origin.Publication`. +Keep that handle for `announce(route)`, `unannounce()`, and `route`; fill tracks through its readonly `broadcast` producer. +`Broadcast.Producer` no longer exposes announcement methods or route metadata, including standalone producers and `prepared.broadcast` before admission. +Closing `publication.broadcast` still retracts and ends the broadcast. Keeping only that fill handle leaves the broadcast and its announcement live; releasing a publication reference does not retract it. + +```ts +const publication = origin.createBroadcast(path); +const broadcast = publication.broadcast; +broadcast.createTrack("data"); +publication.announce({ epoch: Epoch.mint() }); +``` + ### `Varint.sizeLeadingOnes` is gone Take the length of `Varint.encodeLeadingOnes(v)`. @@ -51,9 +65,9 @@ Treat a rejected group read as the loss of that group and keep reading the track ### A claim answers by publishing (#5678) `Origin.Producer.dynamic(prefix)` takes no route: an application claim has no epoch, cost, or hops. -Answer each `Origin.Request` with `using prepared = request.prepare(epoch)` (a fresh `Epoch.mint()` per output run), fill `prepared.broadcast`, then `prepared.publish()`, which announces the broadcast at the exact path under that epoch and returns its producer. -Only `publish` announces it: `prepared.broadcast.announce()` throws. -`Prepared.reject(err)` is replaced by idempotent `close()` and `[Symbol.dispose]()`; either refuses the unpublished answer as `Unroutable` and closes its broadcast. Closing that broadcast also refuses the request. `using` handles normal scope exit and exceptions; after `publish()`, disposal leaves the returned producer live. +Answer each `Origin.Request` with `using prepared = request.prepare(epoch)` (a fresh `Epoch.mint()` per output run), fill `prepared.broadcast`, then `prepared.publish()`, which announces the broadcast at the exact path under that epoch and returns its publication handle. +Only `publish` announces it: `prepared.broadcast` has no announcement methods. +`Prepared.reject(err)` is replaced by idempotent `close()` and `[Symbol.dispose]()`; either refuses the unpublished answer as `Unroutable` and closes its broadcast. Closing that broadcast also refuses the request. `using` handles normal scope exit and exceptions; after `publish()`, disposal leaves the returned publication and its fill producer live. Use `request.reject(err)` to refuse before preparing. A retained, unclosed `Prepared` keeps its request pending; garbage collection provides no prompt cleanup guarantee. Construct a `Prepared` only with `request.prepare()`. Its constructor and the `Request.publisher`, `publishPrepared`, and `rejectPrepared` completion hooks are no longer public. `Dynamic.update` is no longer part of the public API, and `Request` has no `accept` (#6496): only the built-in sessions answer with a broadcast they already hold. @@ -216,7 +230,7 @@ Call `close()` with no argument. ### Origins hold broadcasts (#2705, #3225) -`connection.publish(path, broadcast)` is `origin.createBroadcast(path)` then `broadcast.announce()`. +`connection.publish(path, broadcast)` is `origin.createBroadcast(path)` then `publication.announce()` on the returned handle. `connection.announced(prefix)` is `origin.announced(scope)`, yielding `{ prefix, kind, route }`. `ignoreSelf` is gone: reflected announces are always dropped. diff --git a/js/net/api/index.api.md b/js/net/api/index.api.md index 1afd5d29b42..4a95b0a4740 100644 --- a/js/net/api/index.api.md +++ b/js/net/api/index.api.md @@ -526,6 +526,7 @@ declare namespace Origin_2 { Consumer_5 as Consumer, Dynamic, Request_3 as Request, + Publication, Prepared } } @@ -543,6 +544,7 @@ declare namespace Origin_3 { Consumer_5 as Consumer, Dynamic, Request_3 as Request, + Publication, Prepared } } @@ -612,7 +614,7 @@ class Prepared { [Symbol.dispose](): void; readonly broadcast: Broadcast_2.Producer; close(): void; - publish(): Broadcast_2.Producer; + publish(): Publication; } interface Probe { @@ -653,20 +655,12 @@ class Producer_2 { class Producer_3 { constructor(); - announce(route?: Route | { - epoch?: Route["epoch"]; - hops?: Route["hops"]; - cost?: Route["cost"]; - auth?: Route["auth"]; - }): void; close(): void; get closed(): GetPromise; consume(): Consumer_4; createTrack(name: string, info?: Partial): track.Producer; removeTrack(name: string): void; - get route(): Route | undefined; track(name: string): track.Consumer; - unannounce(): void; } class Producer_4 { @@ -682,7 +676,7 @@ class Producer_5 implements Table { close(abort?: Error): void; get closed(): GetPromise; consume(): Consumer_5; - createBroadcast(path: Path_2.Valid): Broadcast_2.Producer; + createBroadcast(path: Path_2.Valid): Publication; get discovery(): Getter; dynamic(prefix: Path_2.Valid): Dynamic; request(path: Path_2.Valid, options?: RequestOptions): Requesting; @@ -697,6 +691,18 @@ class ProtocolViolation extends Error { readonly code: SessionCode; } +class Publication { + announce(route?: Route | { + epoch?: Route["epoch"]; + hops?: Route["hops"]; + cost?: Route["cost"]; + auth?: Route["auth"]; + }): void; + readonly broadcast: Broadcast_2.Producer; + get route(): Route | undefined; + unannounce(): void; +} + interface ReadOptions { from?: number; } @@ -899,7 +905,7 @@ interface Subscription { interface Table { announced(scope?: Path_2.Pattern, options?: Announce_3.Options): Announce_3.Consumer; readonly closed: GetPromise; - createBroadcast(path: Path_2.Valid): Broadcast_2.Producer; + createBroadcast(path: Path_2.Valid): Publication; readonly discovery: Getter; dynamic(prefix: Path_2.Valid): Dynamic; request(path: Path_2.Valid, options?: RequestOptions): Requesting; diff --git a/js/net/bench/broadcasts.ts b/js/net/bench/broadcasts.ts index e500038b24d..14e3a711e7e 100644 --- a/js/net/bench/broadcasts.ts +++ b/js/net/bench/broadcasts.ts @@ -146,9 +146,9 @@ for (const pinned of [false, true]) { const reader = origin.consume(); const options = pinned ? { epoch } : {}; const publishers = Array.from({ length: routeCount }, (_, index) => { - const source = origin.createBroadcast(Path.from(`room/${index}`)); - source.announce({ epoch }); - return source; + const sourcePublication = origin.createBroadcast(Path.from(`room/${index}`)); + sourcePublication.announce({ epoch }); + return sourcePublication; }); const background = publishers.map((_, index) => reader.request(Path.from(`room/${index}`), options)); const requests = Array.from({ length: readerCount }, () => reader.request(Path.from("room/0"), options)); @@ -416,8 +416,9 @@ for (const routeCount of routeCounts) { for (const observerCount of observerCounts) { const origin = new Producer(); const routes = Array.from({ length: routeCount }, (_, index) => { - const broadcast = origin.createBroadcast(Path.from(`z/${index}`)); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from(`z/${index}`)); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; }); const senders = Array.from({ length: observerCount }, () => { @@ -441,8 +442,9 @@ for (const routeCount of routeCounts) { await Promise.all(senders.map((sender) => sender.entered.promise)); const start = performance.now(); for (let index = 0; index < updates; index++) { - const broadcast = origin.createBroadcast(Path.from(`a/${index}`)); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from(`a/${index}`)); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); await new Promise((resolve) => setImmediate(resolve)); broadcast.close(); await new Promise((resolve) => setImmediate(resolve)); diff --git a/js/net/bench/refresh.ts b/js/net/bench/refresh.ts index 106f3f3b89d..98fd852491c 100644 --- a/js/net/bench/refresh.ts +++ b/js/net/bench/refresh.ts @@ -16,8 +16,9 @@ for (const retained of [1, 64, 1024]) const installed = clock.install(); const source = new Producer(); const destination = new Producer(); - const broadcast = source.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = source.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producers = Array.from({ length: tracks }, (_, i) => broadcast.createTrack(`data${i}`, { refresh: 1 }), ); diff --git a/js/net/bench/requests.ts b/js/net/bench/requests.ts index b57a20aa424..d37c8f69f68 100644 --- a/js/net/bench/requests.ts +++ b/js/net/bench/requests.ts @@ -29,8 +29,9 @@ for (const touched of ["watched", "unrelated"]) const origin = new Producer(); const path = Path.from("room/0"); const broadcasts = Array.from({ length: routes }, (_, i) => { - const broadcast = origin.createBroadcast(Path.from(`room/${i}`)); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from(`room/${i}`)); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; }); const requests = Array.from({ length: followers }, () => origin.request(path, { announced: true })); @@ -41,8 +42,9 @@ for (const touched of ["watched", "unrelated"]) for (let round = 0; round < 6; round++) { const start = performance.now(); for (let i = 0; i < 128; i++) { - const next = origin.createBroadcast(touched === "watched" ? path : Path.from("other")); - next.announce(); + const nextPublication = origin.createBroadcast(touched === "watched" ? path : Path.from("other")); + const next = nextPublication.broadcast; + nextPublication.announce(); if (touched === "watched") { current.close(); current = next; @@ -279,8 +281,9 @@ for (const [version, alpn] of [ const sources = []; for (const prefix of ["a", "b"]) for (let i = 0; i < namespaces; i++) { - const source = origin.createBroadcast(Path.from(`${prefix}/${i}`)); - source.announce(); + const sourcePublication = origin.createBroadcast(Path.from(`${prefix}/${i}`)); + const source = sourcePublication.broadcast; + sourcePublication.announce(); sources.push(source); } const peers = await Promise.all( diff --git a/js/net/bench/serve-browser-entry.ts b/js/net/bench/serve-browser-entry.ts index b511b53453f..64c0cfa9cd4 100644 --- a/js/net/bench/serve-browser-entry.ts +++ b/js/net/bench/serve-browser-entry.ts @@ -17,8 +17,9 @@ import { Timescale, Timestamp } from "../src/time.ts"; async function measure(protocol: string, groups: number, viewers: number) { const origin = new Origin(); - const broadcast = origin.createBroadcast(Path.from("bench")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("bench")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("data", { maxAge: Milli(60_000), timescale: Timescale.MICRO }); const pair = createMockTransportPair(protocol === "lite" ? ALPN_05 : ALPN.DRAFT_22); const version = protocol === "lite" ? LiteVersion.DRAFT_05 : IetfVersion.DRAFT_22; diff --git a/js/net/bench/subscribe-drop.ts b/js/net/bench/subscribe-drop.ts index b7bf27d3294..1eef8afded2 100644 --- a/js/net/bench/subscribe-drop.ts +++ b/js/net/bench/subscribe-drop.ts @@ -24,8 +24,9 @@ try { for (let rep = 0; rep < reps; rep++) { const origins = Array.from({ length: publishers }, () => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("bench")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("bench")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video", { timescale: Timescale.MILLI }); for (let sequence = 0; sequence < groups; sequence++) { const group = new GroupProducer(sequence); diff --git a/js/net/bench/subscribe-tracks.ts b/js/net/bench/subscribe-tracks.ts index 6a7e3e2680c..e51fc49e578 100644 --- a/js/net/bench/subscribe-tracks.ts +++ b/js/net/bench/subscribe-tracks.ts @@ -16,8 +16,9 @@ for (const broadcasts of [1, 16, 128]) { for (const subscribers of [1, 4, 16]) { const origin = new Producer(); const sources = Array.from({ length: broadcasts }, (_, i) => { - const source = origin.createBroadcast(Path.from(`room/${i}`)); - source.announce(); + const sourcePublication = origin.createBroadcast(Path.from(`room/${i}`)); + const source = sourcePublication.broadcast; + sourcePublication.announce(); source.createTrack("initial"); return source; }); diff --git a/js/net/examples/publish.ts b/js/net/examples/publish.ts index e9c38f6d262..98217910d69 100644 --- a/js/net/examples/publish.ts +++ b/js/net/examples/publish.ts @@ -8,8 +8,9 @@ async function main() { const connection = await Moq.Connection.connect({ url, publish: origin.consume() }); // Create a broadcast (a collection of tracks) at a path on the origin - const broadcast = origin.createBroadcast(Moq.Path.from("my-broadcast")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Moq.Path.from("my-broadcast")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); // Create the "chat" track up front. A subscriber is served directly from this // track, no requested() round-trip needed. Mirrors Rust's create_track. diff --git a/js/net/src/announced.test.ts b/js/net/src/announced.test.ts index 70569c6f12d..63d619b840a 100644 --- a/js/net/src/announced.test.ts +++ b/js/net/src/announced.test.ts @@ -7,8 +7,9 @@ import * as Path from "./path.ts"; import { wireOf } from "./wire.ts"; function publish(origin: OriginProducer, path: Path.Valid): BroadcastProducer { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } diff --git a/js/net/src/auth.test.ts b/js/net/src/auth.test.ts index c947339655a..15f5df5e87f 100644 --- a/js/net/src/auth.test.ts +++ b/js/net/src/auth.test.ts @@ -462,8 +462,9 @@ describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", const installed = clock.install(); const logged = spyOn(console, "debug"); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const published = broadcast.createTrack("video"); published.appendGroup().writeFrame({ payload: new Uint8Array([1]) }); const { client, server } = await connect({ serverPublish: origin, protocol }); @@ -529,7 +530,7 @@ describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", try { const unanswered = await requests.next(); if (!unanswered) throw new Error("missing token"); - const track = new OriginProducer().createBroadcast(Path.from("room/x")).createTrack("video"); + const track = new OriginProducer().createBroadcast(Path.from("room/x")).broadcast.createTrack("video"); const reader = track.subscribe(); const ready = wireOf(client).auth.ready(Path.from("room/x"), "subscribe", { until: () => abandoned(track.demand()), @@ -554,8 +555,9 @@ describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", test("a subscribe abandoned while its token waits never reaches the wire", async () => { const logged = spyOn(console, "debug"); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const { client, server, transport } = await connect({ serverPublish: origin, protocol }); const requests = server.auth.requests(); @@ -592,8 +594,9 @@ describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", "a subscribe that regains a reader while its token waits keeps waiting (%d)", async (gap) => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const published = broadcast.createTrack("video"); published.appendGroup().writeFrame({ payload: new Uint8Array([7]) }); const { client, server } = await connect({ serverPublish: origin, protocol }); @@ -628,8 +631,9 @@ describe.each([Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_17, Ietf.ALPN.DRAFT_22])("%s", test("a session's close is not a revocation of its subscriptions", async () => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast .createTrack("video") .appendGroup() @@ -1017,8 +1021,9 @@ test("lite-07 ignores a draining session's late refusal while its successor retr test("a fetch abandoned while its token waits never reaches the wire", async () => { const logged = spyOn(console, "debug"); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast .createTrack("video") .appendGroup() @@ -1060,8 +1065,9 @@ test("a fetch abandoned while its token waits never reaches the wire", async () // The peer refuses a listing its token has not yet authorized, so the request waits for it. test("a listing waits for its token before reaching the wire", async () => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const { client, server, transport } = await connect({ serverPublish: origin, protocol: Lite.ALPN_07_WIP }); const requests = server.auth.requests(); @@ -1097,8 +1103,9 @@ test.each(Array.from({ length: 17 }, (_, gap) => gap))( "a fetch that regains a caller while its token waits keeps waiting (%d)", async (gap) => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room/x")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room/x")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast .createTrack("video") .appendGroup() @@ -1269,13 +1276,15 @@ test.each([ // moq-transport has no stream code for it, so only moq-lite resets with UNAUTHORIZED. test("a revoked grant resets its subscriptions with UNAUTHORIZED", async () => { const clientOrigin = new OriginProducer(); - const up = clientOrigin.createBroadcast(Path.from("up/y")); - up.announce(); + const upPublication = clientOrigin.createBroadcast(Path.from("up/y")); + const up = upPublication.broadcast; + upPublication.announce(); const upTrack = up.createTrack("video"); const serverOrigin = new OriginProducer(); - const down = serverOrigin.createBroadcast(Path.from("room/x")); - down.announce(); + const downPublication = serverOrigin.createBroadcast(Path.from("room/x")); + const down = downPublication.broadcast; + downPublication.announce(); const downTrack = down.createTrack("video"); const { client, server } = await connect({ diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index 07a452d170e..6ece76f5543 100644 --- a/js/net/src/broadcast.ts +++ b/js/net/src/broadcast.ts @@ -7,7 +7,7 @@ import { type Dispose, type GetPromise, type Getter, Once, Signal } from "@moq/s import type * as Epoch from "./epoch.ts"; import { NotFound } from "./error.ts"; import type { Consumer as GroupConsumer } from "./group.ts"; -import { Route } from "./hop.ts"; +import type { Route } from "./hop.ts"; import { hooks } from "./internal.ts"; import { Lister, Listing, listTracks, type Tracks } from "./listing.ts"; import * as Path from "./path.ts"; @@ -382,11 +382,7 @@ export class Producer { producer.#announcer = announcer; }; hooks.attachAnnouncer = attachAnnouncer; - hooks.withholdAnnouncer = (producer) => { - const announcer = producer.#announcer; - producer.#announcer = undefined; - return announcer; - }; + hooks.onClosed = (producer, fn) => { const closers = producer.#state.closers; closers.add(fn); @@ -473,49 +469,7 @@ export class Producer { return trackOf(name, this); } - /** The route this broadcast is announced with, or undefined while it is not announced. */ - get route(): Route | undefined { - return this.#announcer?.route(); - } - - /** - * Advertise this broadcast's exact path, or replace the standing advertisement's route. - * - * The route is taken as given, epoch included: a route with another epoch (or none) - * announces a new broadcast, so re-price from the current one, - * `announce({ ...broadcast.route, cost })`, to keep the instance. - * - * Call it once the tracks a subscriber needs first (a catalog) exist. Until then the - * broadcast exists for nobody, on its own origin or at a peer. Retracts on - * {@link unannounce} or {@link close}. Throws if this producer was not created through an - * origin, is a claim answer still being prepared, or if the broadcast is already closed. - */ - announce( - route: - | Route - | { - epoch?: Route["epoch"]; - hops?: Route["hops"]; - cost?: Route["cost"]; - auth?: Route["auth"]; - } = Route.default, - ): void { - if (this.#state.closed.peek() !== undefined) { - throw new Error("broadcast is closed"); - } - if (!this.#announcer) throw new Error("broadcast is not attached to an origin"); - this.#announcer.announce(Route.normalize(route)); - } - - /** - * Retract the advertisement of this broadcast's path, if any, from local consumers and - * peers alike. {@link announce} brings it back. - */ - unannounce(): void { - this.#announcer?.unannounce(); - } - - /** End the broadcast for good: retract it, serve no new tracks, and refuse a later {@link announce}. Idempotent. */ + /** End the broadcast for good: retract its route and serve no new tracks. Idempotent. */ close(): void { this.#announcer?.unannounce(); this.#announcer = undefined; diff --git a/js/net/src/connection/close.test.ts b/js/net/src/connection/close.test.ts index 38238268f32..fc4dbb53761 100644 --- a/js/net/src/connection/close.test.ts +++ b/js/net/src/connection/close.test.ts @@ -57,8 +57,9 @@ for (const protocol of [Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_18]) { const fin = holdFin(pair.server); const source = new Producer(); const destination = new Producer(); - const broadcast = source.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = source.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const url = new URL("https://localhost/test"); const [client, server] = await Promise.all([ connect({ url, transport: pair.client, consume: destination }), diff --git a/js/net/src/connection/migrate.test.ts b/js/net/src/connection/migrate.test.ts index a6fb2e0988b..9dcd75e1064 100644 --- a/js/net/src/connection/migrate.test.ts +++ b/js/net/src/connection/migrate.test.ts @@ -80,8 +80,9 @@ test("an empty GOAWAY migrates the announced request to the replacement session" const clock = new Manual(); const installed = clock.install(); const { dials, origin } = fleet(); - const broadcast = origin.createBroadcast(Path.from("cam")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("cam")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const consume = new OriginProducer(); const reload = new Reload({ diff --git a/js/net/src/connection/pool.test.ts b/js/net/src/connection/pool.test.ts index 4c0df15b5db..e971e1d6af6 100644 --- a/js/net/src/connection/pool.test.ts +++ b/js/net/src/connection/pool.test.ts @@ -1,11 +1,10 @@ import { afterEach, beforeEach, expect, jest, test } from "bun:test"; import { Milli } from "@moq/clock"; import { Signal } from "@moq/signals"; -import type { Producer as BroadcastProducer } from "../broadcast.ts"; import { SessionCode, SessionError } from "../error.ts"; import * as Lite from "../lite/index.ts"; import { createMockTransportPair } from "../mock.ts"; -import { Producer as OriginProducer } from "../origin.ts"; +import { Producer as OriginProducer, type Table } from "../origin.ts"; import * as Path from "../path.ts"; import { Stream } from "../stream.ts"; import * as Time from "../time.ts"; @@ -14,9 +13,10 @@ import { type AcceptProps, accept as acceptSession } from "./index.ts"; import { Connection, resetShared } from "./pool.ts"; import { Reload } from "./reload.ts"; -function publish(origin: { createBroadcast(path: Path.Valid): BroadcastProducer }, path: Path.Valid) { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); +function publish(origin: Pick, path: Path.Valid) { + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } diff --git a/js/net/src/connection/reload.test.ts b/js/net/src/connection/reload.test.ts index febd5649673..1d308f4dfaf 100644 --- a/js/net/src/connection/reload.test.ts +++ b/js/net/src/connection/reload.test.ts @@ -24,8 +24,9 @@ function accept(transport: WebTransport, url: URL, props: Omit handle.origin.peek() !== undefined); const origin = handle.origin.peek(); if (!origin) throw new Error("no origin"); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); await waitUntil(() => dials.length > 0); const first = dials[0]; @@ -748,8 +750,9 @@ for (const [name, drainEnd] of Object.entries(drainEnds)) { globalThis.WebTransport = stub as unknown as typeof WebTransport; const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const reload = new Reload({ url: new URL("https://example.com/a"), @@ -807,8 +810,9 @@ test("a retired GOAWAY predecessor still closes by its handover cap", async () = globalThis.WebTransport = stub as unknown as typeof WebTransport; const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const reload = new Reload({ url: new URL("https://example.com/"), @@ -889,8 +893,9 @@ test("closing the loop keeps the tokens until the session drains", async () => { const origin = new OriginProducer(); const reload = new Reload({ url, websocket: { enabled: false }, publish: origin.consume() }); await reload.auth.add("token"); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const first = dials[0]; if (!first) throw new Error("no first dial"); diff --git a/js/net/src/connection/scope.test.ts b/js/net/src/connection/scope.test.ts index f3858c2635b..c4d8e4fe784 100644 --- a/js/net/src/connection/scope.test.ts +++ b/js/net/src/connection/scope.test.ts @@ -22,8 +22,9 @@ for (const protocol of [Lite.ALPN_07_WIP, Ietf.ALPN.DRAFT_18]) { new Path.Patterns(["room/**", "other/**"].map(Path.Pattern.parse)), ); const broadcasts = ["room/live", "room/.hidden", "other/live", "outside"].map((path) => { - const broadcast = publish.createBroadcast(Path.from(path)); - broadcast.announce(); + const broadcastPublication = publish.createBroadcast(Path.from(path)); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; }); const [client, server] = await Promise.all([ diff --git a/js/net/src/goaway-requests.test.ts b/js/net/src/goaway-requests.test.ts index 328d3a44d27..3589df5bab0 100644 --- a/js/net/src/goaway-requests.test.ts +++ b/js/net/src/goaway-requests.test.ts @@ -107,8 +107,9 @@ async function handover(kind: "lite" | "ietf"): Promise { const consume = new OriginProducer(); const publish = new OriginProducer(); - const broadcast = publish.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = publish.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const served: TrackProducer[] = []; let armVideo = true; @@ -302,8 +303,9 @@ async function routeDeath(kind: "lite" | "ietf", after: boolean): Promise const consume = new OriginProducer(); const publish = new OriginProducer(); - const broadcast = publish.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = publish.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const served: TrackProducer[] = []; const serving = (async () => { diff --git a/js/net/src/ietf/connection.test.ts b/js/net/src/ietf/connection.test.ts index 47a26930646..c615877ff1a 100644 --- a/js/net/src/ietf/connection.test.ts +++ b/js/net/src/ietf/connection.test.ts @@ -612,8 +612,9 @@ for (const [version, alpn] of [ const pair = createMockTransportPair(alpn); const control = await Stream.open(pair.server, { version }); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video", { timescale: Timescale.MILLI }); const connection = new Connection({ url: new URL("https://example.com"), @@ -664,8 +665,9 @@ async function serving(version: IetfVersion, alpn: string) { const pair = createMockTransportPair(alpn); const control = await Stream.open(pair.server, { version }); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video", { timescale: Timescale.MILLI }); const connection = new Connection({ url: new URL("https://example.com"), diff --git a/js/net/src/ietf/malformed.test.ts b/js/net/src/ietf/malformed.test.ts index 89d34ecd6d5..50cad8252ea 100644 --- a/js/net/src/ietf/malformed.test.ts +++ b/js/net/src/ietf/malformed.test.ts @@ -57,8 +57,9 @@ async function serving(version: IetfVersion, alpn: string) { const pair = createMockTransportPair(alpn); const control = await Stream.open(pair.server, { version }); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video", { timescale: Timescale.MILLI }); const connection = new Connection({ url: new URL("https://example.com"), diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 034b0ed0c55..11141894966 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -45,8 +45,9 @@ import { TRACK_STATUS_ERROR_14, TrackStatusOk, TrackStatusRequest } from "./trac import { ALPN, type IetfVersion, Version } from "./version.ts"; function publish(origin: OriginProducer, path: Path.Valid) { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } @@ -1644,8 +1645,9 @@ test("a request pinned to another epoch is refused DOES_NOT_EXIST", async () => const session = new NativeSession(pair.server, VERSION, true); const { pub, origin } = publisher(pair.server, { session, epoch: true }); const epoch = Epoch.mint(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce({ epoch }); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce({ epoch }); broadcast.createTrack("video"); origin.createBroadcast(Path.from("epochless")).announce(); @@ -2346,8 +2348,9 @@ test("closing a route's token makes its request again without it", async () => { test("a republish withdraws the PUBLISH_NAMESPACE and advertises again", async () => { const { pub, origin, streams, close } = clustered(); try { - const broadcast = origin.createBroadcast(Path.from("mine")); - broadcast.announce({ hops: [VIA], cost: 4n }); + const broadcastPublication = origin.createBroadcast(Path.from("mine")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce({ hops: [VIA], cost: 4n }); void pub.runPublishNamespaces(); const old = await take(streams); @@ -2367,8 +2370,9 @@ test("a republish withdraws the PUBLISH_NAMESPACE and advertises again", async ( test("a republish withdraws the solicited NAMESPACE and sends it again", async () => { const { pair, pub, origin, close } = clustered(true); try { - const broadcast = origin.createBroadcast(Path.from("mine")); - broadcast.announce({ hops: [VIA], cost: 4n }); + const broadcastPublication = origin.createBroadcast(Path.from("mine")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce({ hops: [VIA], cost: 4n }); const subscription = await Stream.open(pair.client, { version: VERSION }); const stream = await Stream.accept(pair.server, VERSION); diff --git a/js/net/src/ietf/subscribe_tracks.test.ts b/js/net/src/ietf/subscribe_tracks.test.ts index f807acf0ff7..550503605b6 100644 --- a/js/net/src/ietf/subscribe_tracks.test.ts +++ b/js/net/src/ietf/subscribe_tracks.test.ts @@ -145,8 +145,9 @@ for (const version of drafts) { const installed = clock.install(); using _installed = { [Symbol.dispose]: () => installed.close() }; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video"); const group = track.appendGroup(); group.writeFrame({ payload: new Uint8Array([1]) }); @@ -191,8 +192,9 @@ for (const version of drafts) { for (const finished of [true, false]) { test(`an unlisted track's PUBLISH ends ${finished ? "with TRACK_ENDED" : "in error"} (${version})`, async () => { const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video"); const listing = await fx.list(); await ok(listing, version); @@ -212,8 +214,9 @@ for (const version of drafts) { const installed = clock.install(); using _installed = { [Symbol.dispose]: () => installed.close() }; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video"); const listing = await fx.list(); await ok(listing, version); @@ -247,8 +250,9 @@ for (const version of drafts) { }); using _held = { [Symbol.dispose]: () => held.mockRestore() }; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video"); const listing = await fx.list(); await ok(listing, version); @@ -275,8 +279,9 @@ for (const version of drafts) { const installed = clock.install(); using _installed = { [Symbol.dispose]: () => installed.close() }; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video"); broadcast.createTrack("audio"); const listing = await fx.list(); @@ -313,8 +318,9 @@ for (const version of drafts) { test(`track listing rejects overlapping prefixes (${version})`, async () => { const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const listing = await fx.list(); await ok(listing, version); @@ -328,8 +334,9 @@ for (const version of drafts) { test(`forwarding listings are skipped or refused according to the draft (${version})`, async () => { const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const listing = await fx.list("room", true); if (version === Version.DRAFT_18) { @@ -348,8 +355,9 @@ for (const version of drafts) { test("draft-18 refuses SUBSCRIBE to a PUBLISHed track", async () => { const version = Version.DRAFT_18; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const listing = await fx.list(); await ok(listing, version); @@ -379,8 +387,9 @@ test("a cancelled listing cannot release its replacement's prefix", async () => const installed = clock.install(); using _installed = { [Symbol.dispose]: () => installed.close() }; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const first = await fx.list(); await ok(first, version); @@ -407,8 +416,9 @@ test("grants hide tracks, revoke PUBLISHes and relist when broadened", async () const fx = fixture(version, { grant }); const listing = await fx.list(); await ok(listing, version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); let delivered = false; const offered = fx.offered(); @@ -436,8 +446,9 @@ test("PUBLISH carries listing defaults and honors INCLUDE_PROPERTIES", async () parameters.includeProperties = false; const listing = await fx.list("room", false, parameters); await ok(listing, version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("later"); const { publish } = await fx.offered(); expect(publish.properties).toEqual({}); @@ -452,8 +463,9 @@ test("PUBLISH restates an omitted listing priority as 128", async () => { const fx = fixture(version); const listing = await fx.list("room", false, new Parameters()); await ok(listing, version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("later"); const { publish } = await fx.offered(); expect(publish.parameters?.subscriberPriority).toBe(128); @@ -479,8 +491,9 @@ test("an update ends only its listing and frees its prefix", async () => { test("session shutdown releases listing tasks while the local origin stays open", async () => { const version = Version.DRAFT_22; const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("video"); const listing = await fx.list(); await ok(listing, version); @@ -493,8 +506,9 @@ test("session shutdown releases listing tasks while the local origin stays open" for (const version of [Version.DRAFT_22]) { test(`a paused filter applies on resume, but an active pause/filter update is refused atomically (${version})`, async () => { const fx = fixture(version); - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video"); for (let i = 0; i < 2; i++) { const group = track.appendGroup(); @@ -543,8 +557,9 @@ const CATALOGS = [4, 64, 512]; /** A broadcast under `room` listing `catalog` tracks. */ function catalogOf(fx: ReturnType, catalog: number) { - const broadcast = fx.origin.createBroadcast(Path.from("room/live")); - broadcast.announce(); + const broadcastPublication = fx.origin.createBroadcast(Path.from("room/live")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); for (let i = 0; i < catalog; i++) broadcast.createTrack(`t${i}`); } diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index ac7c85af870..fd724f18035 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -33,8 +33,9 @@ import { withTimeout } from "./util/timeout.ts"; import { wireOf } from "./wire.ts"; function publish(origin: OriginProducer, path: Path.Valid) { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } @@ -959,8 +960,9 @@ test("integration: a peer is served the cheaper route it was offered, not the lo const path = Path.from("test"); // One publisher instance, so the two compete on cost alone. const epoch = Epoch.mint(); - const local = origin.createBroadcast(path); - local.announce({ epoch, cost: 5n }); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; + localPublication.announce({ epoch, cost: 5n }); const dynamic = origin.dynamicRoute(path, { epoch, cost: 1n }); const [client, server] = await Promise.all([ @@ -2705,11 +2707,12 @@ test("create then announce is discoverable on the wire", async () => { ]); const announced = client.announced(); - const broadcast = origin.createBroadcast(Path.from("later")); + const broadcastPublication = origin.createBroadcast(Path.from("later")); + const broadcast = broadcastPublication.broadcast; broadcast.createTrack("chat", { timescale: Timescale.MILLI }); const pending = announced.next(); - broadcast.announce(); + broadcastPublication.announce(); const entry = await pending; expect(entry?.prefix).toBe("later" as Path.Valid); expect(entry?.kind).toBe("start"); diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index c30337c1b88..37994f7a00b 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -500,10 +500,7 @@ export const hooks: { producer: BroadcastProducer, announcer: { announce(route: Route): void; unannounce(): void; route(): Route | undefined }, ) => void; - /** Take a created broadcast's advertisement away, so it cannot announce until attached again. */ - withholdAnnouncer: ( - producer: BroadcastProducer, - ) => { announce(route: Route): void; unannounce(): void; route(): Route | undefined } | undefined; + /** Call `fn` synchronously when an open broadcast closes, from any handle. */ onClosed: (producer: BroadcastProducer, fn: () => void) => Dispose; /** Name a broadcast handle by the path an origin created or resolved it at. */ @@ -683,9 +680,7 @@ export const hooks: { attachAnnouncer: () => { throw new Error("broadcast.ts not loaded"); }, - withholdAnnouncer: () => { - throw new Error("broadcast.ts not loaded"); - }, + onClosed: () => { throw new Error("broadcast.ts not loaded"); }, diff --git a/js/net/src/lite/compression.test.ts b/js/net/src/lite/compression.test.ts index e24a9f0978a..a023997c355 100644 --- a/js/net/src/lite/compression.test.ts +++ b/js/net/src/lite/compression.test.ts @@ -13,9 +13,10 @@ for (const downstream of [ALPN_06, ALPN_07_WIP]) { const installed = clock.install(); const source = new Producer(); const destination = new Producer(); - const broadcast = source.createBroadcast(Path.from("compressed")); + const broadcastPublication = source.createBroadcast(Path.from("compressed")); + const broadcast = broadcastPublication.broadcast; const track = broadcast.createTrack("data", { compress: true }); - broadcast.announce(); + broadcastPublication.announce(); const peers: { abort(): void }[] = []; async function link(publish: Producer, consume: Producer, alpn: string) { const pair = createMockTransportPair(alpn); diff --git a/js/net/src/lite/connection.test.ts b/js/net/src/lite/connection.test.ts index 700d86adf71..1012a28a98f 100644 --- a/js/net/src/lite/connection.test.ts +++ b/js/net/src/lite/connection.test.ts @@ -223,8 +223,9 @@ for (const alpn of [ALPN_05, ALPN_06, ALPN_07_WIP]) { const pair = createMockTransportPair(alpn); const fin = holdFins(pair.server, "uni"); const origin = new Producer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const url = new URL("https://localhost/test"); const [client, server] = await Promise.all([ @@ -273,8 +274,9 @@ async function servedRawSubscription() { const version = Version.DRAFT_07; const pair = createMockTransportPair(ALPN_07_WIP); const origin = new Producer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const server = new Connection({ url: new URL("https://relay.example/"), @@ -381,8 +383,9 @@ test("close waits for a served FETCH to be acknowledged", async () => { const pair = createMockTransportPair(ALPN_05); const fin = holdFins(pair.server, "incoming-bidi"); const origin = new Producer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const group = producer.appendGroup(); group.writeFrame(textFrame("last")); @@ -433,8 +436,9 @@ test("close waits for a request served while withdrawals are in flight", async ( const fin = holdFins(pair.server, "incoming-bidi"); const source = new Producer(); const destination = new Producer(); - const broadcast = source.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = source.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const group = producer.appendGroup(); group.writeFrame(textFrame("last")); diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index f2de5ce23c7..ba4a0675e58 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -33,8 +33,9 @@ import { Track as TrackMessage } from "./track.ts"; import { ALPN_05, ALPN_06, ALPN_07_WIP, Version } from "./version.ts"; function publish(origin: OriginProducer, path: Path.Valid) { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } @@ -106,8 +107,9 @@ test.each([ const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("cam")); - broadcast.announce({ cost: 7n }); + const broadcastPublication = origin.createBroadcast(Path.from("cam")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce({ cost: 7n }); const written: Uint8Array[] = []; const stream = new Stream({ @@ -125,7 +127,7 @@ test.each([ const initial = written.length; expect(initial).toBeGreaterThan(0); - broadcast.announce({ cost: 9n }); + broadcastPublication.announce({ cost: 9n }); await settle(); expect(written.length > initial).toBe(sent); @@ -2443,8 +2445,9 @@ test("lite draft-06: announces wait for the setup grant", async () => { answered = resolve; }); const publisher = new Publisher(pair.server, Version.DRAFT_06, randomHop(), origin.consume(), { grant, ready }); - const broadcast = origin.createBroadcast(Path.from("foo/bar")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("foo/bar")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const written: Uint8Array[] = []; const stream = new Stream({ diff --git a/js/net/src/lite/refresh.test.ts b/js/net/src/lite/refresh.test.ts index 64091a24c69..cdde14e64d8 100644 --- a/js/net/src/lite/refresh.test.ts +++ b/js/net/src/lite/refresh.test.ts @@ -13,8 +13,9 @@ test.each([ALPN_06, ALPN_07_WIP])("refresh joins reach a browser publisher on %s const installed = clock.install(); const source = new Producer(); const b = new Producer(); - const broadcast = source.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = source.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("data", { refresh: 1 }); const pair = createMockTransportPair(alpn); const url = new URL("https://localhost/refresh"); diff --git a/js/net/src/lite/track_stream.test.ts b/js/net/src/lite/track_stream.test.ts index 903d7f7fc13..97bdb20ac3f 100644 --- a/js/net/src/lite/track_stream.test.ts +++ b/js/net/src/lite/track_stream.test.ts @@ -118,8 +118,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); const serving = (async () => { for (;;) { @@ -155,8 +156,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); const serving = (async () => { for (;;) { @@ -201,8 +203,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); let requests = 0; const serving = (async () => { @@ -243,8 +246,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); let requests = 0; const serving = (async () => { @@ -302,8 +306,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); // Answered by hand, after every requester left. const requested = wireOf(broadcast).requested(); @@ -334,8 +339,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); const requested = wireOf(broadcast).requested(); @@ -357,8 +363,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const demand = broadcast.demand(); // Answered by hand, after the first TRACK stream has left. const requested = wireOf(broadcast).requested(); @@ -392,8 +399,9 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver const pair = createMockTransportPair(ALPN_05); const origin = new OriginProducer(); const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const serving = (async () => { for (;;) { const request = await wireOf(broadcast).requested(); diff --git a/js/net/src/lite/tracks.test.ts b/js/net/src/lite/tracks.test.ts index 5c7a9185085..9574c45028e 100644 --- a/js/net/src/lite/tracks.test.ts +++ b/js/net/src/lite/tracks.test.ts @@ -353,8 +353,9 @@ function publisher(origin: OriginProducer) { test("data after a TRACKS request closes the session", async () => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("a"); const { list, closing } = publisher(origin); const listing = await list("room"); @@ -368,8 +369,9 @@ test("data after a TRACKS request closes the session", async () => { test("a publisher lists its broadcast's own tracks, never one served on request", async () => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("a", { priority: 5 }); // Served on request: answered when asked, never listed. void wireOf(broadcast) @@ -390,8 +392,9 @@ test("a publisher lists its broadcast's own tracks, never one served on request" test("a newer listing of a broadcast on one session replaces the older", async () => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("a"); const { list } = publisher(origin); const older = await list("room"); @@ -461,8 +464,9 @@ test("a listing over a session carries each track's info, end to end", async () connect({ url, transport: pair.client }), accept({ transport: pair.server, url, publish: origin.consume() }), ]); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("a", { timescale: Timescale.MILLI, maxAge: Milli(3000), @@ -496,9 +500,10 @@ test("an epoch-pinned front lists its serving route's tracks", async () => { connect({ url, transport: pair.client, consume: clientOrigin }), accept({ transport: pair.server, url, publish: serverOrigin.consume() }), ]); - const broadcast = serverOrigin.createBroadcast(Path.from("room")); + const broadcastPublication = serverOrigin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; broadcast.createTrack("a"); - broadcast.announce({ epoch: Epoch.parse("01900000-0000-7000-8000-000000000001") }); + broadcastPublication.announce({ epoch: Epoch.parse("01900000-0000-7000-8000-000000000001") }); const reader = clientOrigin.consume(); const announced = reader.announced(); @@ -551,8 +556,9 @@ test("lite-06 cannot list over a session", async () => { connect({ url, transport: pair.client }), accept({ transport: pair.server, url, publish: origin.consume() }), ]); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const tracks = wireOf(client).consume(Path.from("room")).tracks(); expect(await next(tracks)).toMatch(/requires moq-lite-07/); broadcast.close(); diff --git a/js/net/src/order.test.ts b/js/net/src/order.test.ts index b29247db139..0cb8b6433b3 100644 --- a/js/net/src/order.test.ts +++ b/js/net/src/order.test.ts @@ -85,8 +85,9 @@ async function backlog(protocol: string, order: Order, update?: Order) { const origin = new OriginProducer(); const { client, server, transport } = await session(protocol, origin); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const groups = [0, 1, 2].map(() => writeGroup(video)); @@ -138,8 +139,9 @@ test.each(["ascending", "descending"] as const)("ietf serves GROUP_ORDER %s", as const origin = new OriginProducer(); const { client, server, transport } = await session(Ietf.ALPN.DRAFT_17, origin); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const remote = wireOf(client).consume(Path.from("test")); @@ -202,8 +204,9 @@ test("ietf keeps a closing group stream ranked until it settles", async () => { const { client, server, transport } = await session(Ietf.ALPN.DRAFT_17, origin); const fins = holdFins(transport); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const remote = wireOf(client).consume(Path.from("test")); @@ -242,8 +245,9 @@ test("a shared track aggregates descending and serves each subscriber in its own session(Lite.ALPN_07_WIP, origin), ]); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const ascendingRemote = wireOf(ascending.client).consume(Path.from("test")); diff --git a/js/net/src/origin-request.test.ts b/js/net/src/origin-request.test.ts index 47a9e024dfa..fb6f1315128 100644 --- a/js/net/src/origin-request.test.ts +++ b/js/net/src/origin-request.test.ts @@ -4,7 +4,7 @@ import * as Epoch from "./epoch"; import { StreamCode, StreamError } from "./error"; import { Route } from "./hop"; import { hooks } from "./internal.ts"; -import { Producer, type Request } from "./origin"; +import { Producer, type Publication, type Request } from "./origin"; import * as Path from "./path"; import { wireOf } from "./wire"; @@ -13,20 +13,23 @@ const settle = () => new Promise((resolve) => setImmediate(resolve)); test("an announced request follows restart, end and a later start", async () => { const origin = new Producer(); const path = Path.from("room/audio"); - const first = origin.createBroadcast(path); - first.announce(); + const firstPublication = origin.createBroadcast(path); + const first = firstPublication.broadcast; + firstPublication.announce(); const request = origin.request(path, { announced: true }); expect(request.active.peek()?.closed).toBe(first.closed); - const second = origin.createBroadcast(path); - second.announce(); + const secondPublication = origin.createBroadcast(path); + const second = secondPublication.broadcast; + secondPublication.announce(); await settle(); expect(request.active.peek()?.closed).toBe(second.closed); second.close(); await settle(); expect(request.active.peek()).toBeUndefined(); expect(request.closed.peek()).toBeUndefined(); - const third = origin.createBroadcast(path); - third.announce(); + const thirdPublication = origin.createBroadcast(path); + const third = thirdPublication.broadcast; + thirdPublication.announce(); await settle(); expect(request.active.peek()?.closed).toBe(third.closed); request.close(); @@ -39,11 +42,13 @@ test("an epoch-pinned announced request refuses another instance", async () => { const origin = new Producer(); const path = Path.from("room/audio"); const epoch = Epoch.mint(); - const first = origin.createBroadcast(path); - first.announce({ epoch }); + const firstPublication = origin.createBroadcast(path); + const first = firstPublication.broadcast; + firstPublication.announce({ epoch }); const request = origin.request(path, { announced: true, epoch }); - const second = origin.createBroadcast(path); - second.announce({ epoch: Epoch.mint() }); + const secondPublication = origin.createBroadcast(path); + const second = secondPublication.broadcast; + secondPublication.announce({ epoch: Epoch.mint() }); await settle(); expect(request.active.peek()?.closed).not.toBe(second.closed); request.close(); @@ -55,13 +60,14 @@ test("an epoch-pinned announced request refuses another instance", async () => { test("an announced request changes its handle on an in-place epoch restart", async () => { const origin = new Producer(); const path = Path.from("room/audio"); - const broadcast = origin.createBroadcast(path); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; const before = Epoch.mint(); - broadcast.announce({ epoch: before }); + broadcastPublication.announce({ epoch: before }); const request = origin.request(path, { announced: true }); const old = request.active.peek(); const after = Epoch.mint(); - broadcast.announce({ epoch: after }); + broadcastPublication.announce({ epoch: after }); await settle(); expect(request.active.peek()?.epoch).toBe(after); expect(request.active.peek()).not.toBe(old); @@ -75,8 +81,9 @@ test("a closed announced request never acquires another publisher", async () => const path = Path.from("room/audio"); const request = origin.request(path, { announced: true }); request.close(); - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); await settle(); expect(request.active.peek()).toBeUndefined(); expect(request.closed.peek()).toBeNull(); @@ -98,10 +105,11 @@ test("unrelated routes never notify announced request followers", async () => { await settle(); let changes = 0; const stops = requests.map((request) => request.active.subscribe(() => changes++)); - const other = origin.createBroadcast(Path.from("unrelated")); - other.announce(); + const otherPublication = origin.createBroadcast(Path.from("unrelated")); + const other = otherPublication.broadcast; + otherPublication.announce(); await settle(); - other.announce({ cost: 5n }); + otherPublication.announce({ cost: 5n }); await settle(); other.close(); await settle(); @@ -149,12 +157,14 @@ test("announced requests follow the serving prefix, ignoring descendants and re- const request = origin.request(Path.from("pool/job"), { announced: true }); await settle(); expect(request.active.peek()?.closed).toBe(source.closed); - const below = origin.createBroadcast(Path.from("pool/job/thumb")); - below.announce(); + const belowPublication = origin.createBroadcast(Path.from("pool/job/thumb")); + const below = belowPublication.broadcast; + belowPublication.announce(); await settle(); expect(request.active.peek()?.closed).toBe(source.closed); - const exact = origin.createBroadcast(Path.from("pool/job")); - exact.announce(); + const exactPublication = origin.createBroadcast(Path.from("pool/job")); + const exact = exactPublication.broadcast; + exactPublication.announce(); await settle(); expect(request.active.peek()?.closed).toBe(exact.closed); prefix.updateRoute({ cost: 5n }); @@ -179,8 +189,9 @@ test("an announced request ignores a covering route whose scope excludes its pat const origin = new Producer(); const scoped = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.parse("room/*/chat")])); const prefix = scoped.dynamicRoute(Path.from("room/alice")); - const exact = origin.createBroadcast(Path.from("room/alice")); - exact.announce({ cost: 5n }); + const exactPublication = origin.createBroadcast(Path.from("room/alice")); + const exact = exactPublication.broadcast; + exactPublication.announce({ cost: 5n }); const request = origin.request(Path.from("room/alice"), { announced: true }); expect(request.active.peek()?.closed).toBe(exact.closed); exact.close(); @@ -314,7 +325,7 @@ test("a claim answer is hidden until published under its own epoch", async () => await settle(); expect(request.active.peek()?.epoch).toBe(epoch); request.close(); - published.close(); + published.broadcast.close(); claim.close(); origin.close(); }); @@ -329,8 +340,8 @@ test("a prepared answer cannot announce itself before publish", async () => { if (!pending) throw new Error("expected a request"); const epoch = Epoch.mint(); const prepared = pending.prepare(epoch); - expect(() => prepared.broadcast.announce({ epoch: Epoch.mint() })).toThrow("not attached to an origin"); - expect(prepared.broadcast.route).toBeUndefined(); + expect(prepared.broadcast).not.toHaveProperty("announce"); + expect(prepared.broadcast).not.toHaveProperty("route"); await settle(); expect(request.active.peek()).toBeUndefined(); const published = prepared.publish(); @@ -338,7 +349,7 @@ test("a prepared answer cannot announce itself before publish", async () => { await settle(); expect(request.active.peek()?.epoch).toBe(epoch); request.close(); - published.close(); + published.broadcast.close(); claim.close(); origin.close(); }); @@ -468,7 +479,7 @@ test("using transfers a published answer without closing it on disposal", async const request = origin.request(path); const pending = (await requests.next()).value; if (!pending) throw new Error("expected a request"); - let published: BroadcastProducer; + let published: Publication; { using prepared = pending.prepare(Epoch.mint()); prepared.broadcast.createTrack("data").writeFrame({ payload: new TextEncoder().encode("published") }); @@ -477,13 +488,13 @@ test("using transfers a published answer without closing it on disposal", async prepared.close(); } await settle(); - expect(published.closed.peek()).toBeUndefined(); + expect(published.broadcast.closed.peek()).toBeUndefined(); expect(request.active.peek()?.epoch).toBe(published.route?.epoch); const reader = request.active.peek()?.track("data").subscribe().ordered(); expect(await reader?.readString()).toBe("published"); reader?.close(); request.close(); - published.close(); + published.broadcast.close(); claim.close(); origin.close(); }); @@ -519,7 +530,7 @@ for (const cleanup of ["close", "dispose", "broadcast", "publish"] as const) { reader?.close(); old.close(); later.close(); - published.close(); + published.broadcast.close(); claim.close(); origin.close(); }); diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index 2851a8df8e4..57261d5b64e 100644 --- a/js/net/src/origin.test.ts +++ b/js/net/src/origin.test.ts @@ -15,8 +15,9 @@ import { Timescale } from "./time.ts"; import { overrideBroadcastWire, wireOf } from "./wire.ts"; function publish(origin: Producer, path: Path.Valid) { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } @@ -77,13 +78,14 @@ test("a borrowed table exposes dynamic serving and a live scoped broadcast map", const other = publish(origin, Path.from("other")); const localPath = Path.from("room/alice"); - const local = table.createBroadcast(localPath); + const localPublication = table.createBroadcast(localPath); + const local = localPublication.broadcast; await new Promise((resolve) => setImmediate(resolve)); expect(live.peek().has(localPath)).toBe(false); await new Promise((resolve) => setImmediate(resolve)); expect(live.peek().has(Path.from("other"))).toBe(false); - local.announce({ epoch: EPOCH, cost: 4n }); + localPublication.announce({ epoch: EPOCH, cost: 4n }); await settle(); await new Promise((resolve) => setImmediate(resolve)); expect(live.peek().get(localPath)).toEqual(Route.normalize({ epoch: EPOCH, cost: 4n })); @@ -119,15 +121,16 @@ test("an announced local path competes with a received route on cost", async () expect(live.peek().get(path)).toEqual(Route.normalize({ epoch: EPOCH, hops: [PEER], cost: 5n })); // Unannounced, it competes for nothing. - const local = origin.createBroadcast(path); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; await new Promise((resolve) => setImmediate(resolve)); expect(live.peek().get(path)).toEqual(Route.normalize({ epoch: EPOCH, hops: [PEER], cost: 5n })); // Cheaper wins; a tie falls to the local broadcast. - local.announce({ epoch: EPOCH, cost: 9n }); + localPublication.announce({ epoch: EPOCH, cost: 9n }); await new Promise((resolve) => setImmediate(resolve)); expect(live.peek().get(path)).toEqual(Route.normalize({ epoch: EPOCH, hops: [PEER], cost: 5n })); - local.announce({ epoch: EPOCH, cost: 5n }); + localPublication.announce({ epoch: EPOCH, cost: 5n }); await new Promise((resolve) => setImmediate(resolve)); expect(live.peek().get(path)).toEqual(Route.normalize({ epoch: EPOCH, cost: 5n })); @@ -244,15 +247,24 @@ test("a consumer clone keeps a superseded broadcast alive", async () => { const consumer = origin.consume(); const path = Path.from("room"); - const first = publish(origin, path); + const publication = origin.createBroadcast(path); + const first = publication.broadcast; + publication.announce(); const mine = await routed(consumer, path); expect(mine).toBeDefined(); - publish(origin, path); + const replacement = origin.createBroadcast(path); + replacement.announce(); await settle(); // The application's clone holds the old broadcast open even though it is unpublished. expect(first.closed.peek()).toBeUndefined(); + // Holding the fill alive never lets its stale publication displace or retract the replacement. + expect(() => publication.announce({ epoch: EPOCH })).toThrow("broadcast is closed"); + publication.unannounce(); + const current = await routed(consumer, path); + expect(readsFrom(current, replacement.broadcast)).toBe(true); + current?.close(); mine?.close(); await settle(); @@ -951,14 +963,15 @@ test("createBroadcast is invisible to everyone until announce", async () => { const consumer = origin.consume(); const path = Path.from("room"); - const broadcast = origin.createBroadcast(path); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; expect(wireOf(consumer).routes(path)).toBe(false); expect(wireOf(consumer).advertised.peek()?.has(path)).toBe(false); const request = consumer.request(path); expect(request.active.peek()).toBeUndefined(); const announced = consumer.announced(); - broadcast.announce({ cost: 4n }); + broadcastPublication.announce({ cost: 4n }); expect(wireOf(consumer).routes(path)).toBe(true); expect(wireOf(consumer).advertised.peek()?.has(path)).toBe(true); expect(await announced.next()).toMatchObject({ @@ -970,7 +983,7 @@ test("createBroadcast is invisible to everyone until announce", async () => { expect(first).toBeDefined(); // Off the air for local consumers and peers alike, which ends the request on it. - broadcast.unannounce(); + broadcastPublication.unannounce(); expect(wireOf(consumer).routes(path)).toBe(false); expect(wireOf(consumer).advertised.peek()?.has(path)).toBe(false); expect(await announced.next()).toMatchObject({ prefix: path, kind: "end" }); @@ -978,7 +991,7 @@ test("createBroadcast is invisible to everyone until announce", async () => { expect(request.closed.peek()).toBeInstanceOf(Error); // Back on the air for a fresh request, through a fresh handle. - broadcast.announce(); + broadcastPublication.announce(); expect(await announced.next()).toMatchObject({ prefix: path, kind: "start" }); const fresh = consumer.request(path); const again = fresh.active.peek(); @@ -999,8 +1012,9 @@ test("a request prefers a cheaper received route over an announced local broadca const path = Path.from("room"); const upstream = new BroadcastProducer(); const dispose = serve(origin, path, provider(upstream), Route.normalize({ epoch: EPOCH, hops: [PEER], cost: 1n })); - const local = origin.createBroadcast(path); - local.announce({ epoch: EPOCH, cost: 3n }); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH, cost: 3n }); // The cheaper route is served on demand, so it resolves only once its handler answers. const request = consumer.request(path); @@ -1013,7 +1027,7 @@ test("a request prefers a cheaper received route over an announced local broadca // Re-priced below it, the local broadcast of the same instance wins at once: the request's // broadcast stays, and its tracks read the local one. - local.announce({ epoch: EPOCH, cost: 0n }); + localPublication.announce({ epoch: EPOCH, cost: 0n }); expect(await announced.next()).toMatchObject({ prefix: path, kind: "update", route: { epoch: EPOCH, hops: [] } }); expect(request.active.peek()).toBe(remote); expect(readsFrom(remote, local)).toBe(true); @@ -1038,8 +1052,9 @@ test("the cheapest received route competes with the local broadcast, not the new }; const cheap = serve(origin, path, via("cheap"), Route.normalize({ epoch: EPOCH, hops: [PEER], cost: 1n })); const pricey = serve(origin, path, via("pricey"), Route.normalize({ epoch: EPOCH, hops: [PEER], cost: 10n })); - const local = origin.createBroadcast(path); - local.announce({ epoch: EPOCH, cost: 5n }); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH, cost: 5n }); await new Promise((resolve) => setImmediate(resolve)); @@ -1209,12 +1224,13 @@ test("a peer is offered and served the cheapest originated route", async () => { expect(await pending).toBeDefined(); // An exact-path local broadcast competes with them on cost too. - const local = origin.createBroadcast(prefix); - local.announce({ epoch: EPOCH, cost: 5n }); + const localPublication = origin.createBroadcast(prefix); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH, cost: 5n }); expect(wireOf(consumer).advertised.peek()?.get(prefix)?.[0]?.route).toEqual( Route.normalize({ epoch: EPOCH, cost: 1n }), ); - local.announce({ epoch: EPOCH, cost: 0n }); + localPublication.announce({ epoch: EPOCH, cost: 0n }); expect(wireOf(consumer).advertised.peek()?.get(prefix)?.[0]?.route).toEqual( Route.normalize({ epoch: EPOCH, cost: 0n }), ); @@ -1502,8 +1518,9 @@ test("a request on a replaced epoch fails over to a local replica of it", async const primary = new BroadcastProducer(); const keepPrimary = primary.consume(); const disposePrimary = serve(origin, path, () => keepPrimary.clone(), Route.normalize({ epoch: EPOCH })); - const local = origin.createBroadcast(path); - local.announce({ epoch: EPOCH, cost: 5n }); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH, cost: 5n }); const request = consumer.request(path); await settle(); const first = request.active.peek(); @@ -1543,8 +1560,9 @@ test("a stale takeover's refusal leaves a request whose source came back", async const path = Path.from("resumed"); const newer = Epoch.parse("01900000-0000-7000-8000-000000000002"); - const local = origin.createBroadcast(path); - local.announce({ epoch: EPOCH }); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH }); const standby = wireOf(origin).receive(path, Route.normalize({ epoch: EPOCH, cost: 5n })); const asked = standby.requested(); const request = consumer.request(path); @@ -1558,9 +1576,9 @@ test("a stale takeover's refusal leaves a request whose source came back", async expect(request.active.peek()).toBe(first); // The local broadcast steps away, so the standby is asked to take over; it comes back first. - local.unannounce(); + localPublication.unannounce(); const takeover = await asked.next(); - local.announce({ epoch: EPOCH }); + localPublication.announce({ epoch: EPOCH }); await settle(); takeover.value?.reject(new Error("too late")); await settle(); @@ -1884,7 +1902,7 @@ test("a rooted producer shares the table and enforces its pattern union", async await new Promise((resolve) => setImmediate(resolve)); expect(request.active.peek()).toBeDefined(); await new Promise((resolve) => setImmediate(resolve)); - expect(() => scoped.createBroadcast(Path.from("other"))).toThrow("outside the origin scope"); + expect(() => scoped.createBroadcast(Path.from("other")).broadcast).toThrow("outside the origin scope"); await new Promise((resolve) => setImmediate(resolve)); expect(() => scoped.request(Path.from("room/alice/deep"))).toThrow("outside the origin scope"); await new Promise((resolve) => setImmediate(resolve)); @@ -1900,7 +1918,7 @@ test("a rooted producer shares the table and enforces its pattern union", async await new Promise((resolve) => setImmediate(resolve)); expect([...(await announcedView(nested)).peek().keys()]).toEqual([Path.from("alice")]); await new Promise((resolve) => setImmediate(resolve)); - expect(() => nested.createBroadcast(Path.from("alice/deep"))).toThrow("outside the origin scope"); + expect(() => nested.createBroadcast(Path.from("alice/deep")).broadcast).toThrow("outside the origin scope"); const announced = nested.announced(); await new Promise((resolve) => setImmediate(resolve)); expect(await announced.next()).toMatchObject({ kind: "start", prefix: Path.from("alice") }); @@ -1930,7 +1948,7 @@ test("closing a derived handle releases only what it registered", async () => { expect([...routes.keys()]).toEqual([Path.from("lobby")]); // A released handle refuses new work, so a borrower notices instead of publishing into nothing. - expect(() => scoped.createBroadcast(Path.from("room/late"))).toThrow("origin handle is closed"); + expect(() => scoped.createBroadcast(Path.from("room/late")).broadcast).toThrow("origin handle is closed"); scoped.dynamic(Path.from("room/late")).close(); expect((await announcedView(origin)).peek().has(Path.from("room/late"))).toBe(false); @@ -1953,14 +1971,16 @@ test("a broadcast a derived handle released cannot announce again", async () => const consumer = origin.consume(); const scoped = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.all()])); const path = Path.from("room"); - const broadcast = publish(scoped, path); + const publication = scoped.createBroadcast(path); + const broadcast = publication.broadcast; + publication.announce(); const held = await routed(consumer, path); scoped.close(); // The application's clone keeps the broadcast open, but the released handle no longer routes it. expect(broadcast.closed.peek()).toBeUndefined(); - expect(() => broadcast.announce()).toThrow("broadcast is closed"); + expect(() => publication.announce()).toThrow("broadcast is closed"); expect((await announcedView(origin)).peek().has(path)).toBe(false); held?.close(); @@ -2136,8 +2156,9 @@ test("a scoped reader sees a local broadcast that only loses to a route outside const chat = origin .scope(Path.empty(), new Path.Patterns([Path.Pattern.parse("room/*/chat")])) .dynamicRoute(Path.from("room/alice"), { epoch: EPOCH, cost: 0n }); - const local = origin.createBroadcast(Path.from("room/alice")); - local.announce({ epoch: EPOCH, cost: 5n }); + const localPublication = origin.createBroadcast(Path.from("room/alice")); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH, cost: 5n }); const reader = origin.scope(Path.empty(), new Path.Patterns([Path.Pattern.parse("room/*")])); await new Promise((resolve) => setImmediate(resolve)); @@ -2192,8 +2213,9 @@ test("the newest epoch wins the path over a cheaper one and arrives as a new bro const path = Path.from("room/alice"); const announced = consumer.announced(); - const old = origin.createBroadcast(path); - old.announce({ epoch: Epoch.mint(), cost: 1n }); + const oldPublication = origin.createBroadcast(path); + const old = oldPublication.broadcast; + oldPublication.announce({ epoch: Epoch.mint(), cost: 1n }); expect(await announced.next()).toMatchObject({ prefix: path, kind: "start" }); // Minted after the local broadcast's, so it is the newer publisher despite the cost. @@ -2291,13 +2313,14 @@ test("re-pricing from the current route keeps the epoch", () => { handle.updateRoute({ ...handle.route, cost: 1n }); expect(handle.route).toEqual(Route.normalize({ epoch: EPOCH, cost: 1n })); - const broadcast = origin.createBroadcast(Path.from("room")); - expect(broadcast.route).toBeUndefined(); - broadcast.announce({ epoch: EPOCH, cost: 5n }); - broadcast.announce({ ...broadcast.route, cost: 1n }); - expect(broadcast.route).toEqual(Route.normalize({ epoch: EPOCH, cost: 1n })); - broadcast.unannounce(); - expect(broadcast.route).toBeUndefined(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + expect(broadcastPublication.route).toBeUndefined(); + broadcastPublication.announce({ epoch: EPOCH, cost: 5n }); + broadcastPublication.announce({ ...broadcastPublication.route, cost: 1n }); + expect(broadcastPublication.route).toEqual(Route.normalize({ epoch: EPOCH, cost: 1n })); + broadcastPublication.unannounce(); + expect(broadcastPublication.route).toBeUndefined(); broadcast.close(); handle.close(); @@ -2308,8 +2331,9 @@ test("a peer naming another epoch is refused rather than served a different inst const origin = new Producer(); const consumer = origin.consume(); const path = Path.from("room"); - const local = origin.createBroadcast(path); - local.announce({ epoch: EPOCH }); + const localPublication = origin.createBroadcast(path); + const local = localPublication.broadcast; + localPublication.announce({ epoch: EPOCH }); expect(wireOf(consumer).local(path, EPOCH)).toBeDefined(); const other = Epoch.parse("01900000-0000-7000-8000-000000000002"); @@ -2333,6 +2357,8 @@ test("a newer epoch ends pinned and unpinned requests on the replaced publisher const newerEpoch = Epoch.parse("01900000-0000-7000-8000-000000000002"); const second = origin.createBroadcast(path); second.announce({ epoch: newerEpoch }); + expect(() => first.announce({ epoch: EPOCH })).toThrow("broadcast is closed"); + first.unannounce(); expect(pinned.active.peek()).toBeUndefined(); // The replaced publisher closed: the unpinned request ends rather than move to the newer one. expect(unpinned.active.peek()).toBeUndefined(); @@ -2599,16 +2625,19 @@ test("a route without an epoch resolves a handle without one", () => { test("public announcements saturate static costs and refuse invalid cost shapes", async () => { const origin = new Producer(); const path = Path.from("priced"); - const broadcast = origin.createBroadcast(path); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; const view = await announcedView(origin); - broadcast.announce({ cost: 2n ** 64n - 1n }); + broadcastPublication.announce({ cost: 2n ** 64n - 1n }); await settle(); await new Promise((resolve) => setImmediate(resolve)); expect(view.peek().get(path)?.cost).toBe(2n ** 62n - 1n); await new Promise((resolve) => setImmediate(resolve)); - expect(() => broadcast.announce({ cost: -1n })).toThrow(RangeError); + expect(() => broadcastPublication.announce({ cost: -1n })).toThrow(RangeError); await new Promise((resolve) => setImmediate(resolve)); - expect(() => broadcast.announce({ cost: { warm: 1n, cold: 2n } } as unknown as Route)).toThrow(RangeError); + expect(() => broadcastPublication.announce({ cost: { warm: 1n, cold: 2n } } as unknown as Route)).toThrow( + RangeError, + ); broadcast.close(); origin.close(); }); @@ -2618,14 +2647,14 @@ test("published names refuse navigation segments", () => { try { for (const name of ["room/./cam", "room/../cam", ".", ".."]) { const path = Path.from(name); - expect(() => origin.createBroadcast(path)).toThrow(); + expect(() => origin.createBroadcast(path).broadcast).toThrow(); expect(() => origin.dynamic(path)).toThrow(); expect(() => wireOf(origin).receive(path, Route.default)).toThrow(); } const scoped = origin.scope(Path.from("room/.."), new Path.Patterns([Path.Pattern.all()])); - expect(() => scoped.createBroadcast(Path.from("cam"))).toThrow(); + expect(() => scoped.createBroadcast(Path.from("cam")).broadcast).toThrow(); for (const path of ["room/.pro/cam", "room/a..b", "room/cam"]) { - origin.createBroadcast(Path.from(path)).close(); + origin.createBroadcast(Path.from(path)).broadcast.close(); } } finally { origin.close(); diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 1d331596015..086ffcdad98 100644 --- a/js/net/src/origin.ts +++ b/js/net/src/origin.ts @@ -1057,7 +1057,7 @@ export interface Table { readonly discovery: Getter; /** Create an unannounced broadcast at `path`; see {@link Producer.createBroadcast}. */ - createBroadcast(path: Path.Valid): broadcast.Producer; + createBroadcast(path: Path.Valid): Publication; /** Resolve `path`, optionally waiting for an announcement; see {@link Consumer.request}. */ request(path: Path.Valid, options?: RequestOptions): Requesting; @@ -1095,7 +1095,7 @@ export interface RequestOptions { * the origin down. A reconnecting session re-announces the table on each attach, so * advertisements made while offline surface on the next connection. * - * Create a broadcast, populate its tracks, then {@link broadcast.Producer.announce}: an + * Create a broadcast, populate its tracks, then {@link Publication.announce}: an * exact-path subscribe before the tracks exist is refused, and nobody can see or reach a * broadcast until it announces. To publish broadcasts on demand instead, claim a prefix with * {@link dynamic} and answer the requests beneath it. @@ -1180,11 +1180,11 @@ export class Producer implements Table { } /** - * Create a broadcast at `path`, returning its producer. + * Create a broadcast at `path`, returning its publication handle. * - * The broadcast exists for nobody until {@link broadcast.Producer.announce}: announce + * The broadcast exists for nobody until {@link Publication.announce}: announce * streams skip it and requests for its path find nothing, on this origin exactly as at - * a peer. Announce once its tracks exist; {@link broadcast.Producer.unannounce} + * a peer. Announce once its tracks exist; {@link Publication.unannounce} * withdraws it from everyone again. * * Close the producer to drop it. Creating a path again supersedes the previous @@ -1193,7 +1193,7 @@ export class Producer implements Table { * (`Epoch.mint()` per run) so a restart replaces the old broadcast rather than resuming * into it; at the same epoch, a local broadcast competes on cost and wins ties. */ - createBroadcast(path: Path.Valid): broadcast.Producer { + createBroadcast(path: Path.Valid): Publication { path = this.#scope.path(path); if (this.#borrowed?.closed) throw new Error("origin handle is closed"); publishPath(path); @@ -1204,14 +1204,15 @@ export class Producer implements Table { hooks.stampPath(producer, path); const front = producer.consume(); - hooks.attachAnnouncer(producer, { + const announcer: broadcast.Announcer = { announce: (route) => this.#advertiseExact(path, front, route), unannounce: () => this.#retractExact(path, front), route: () => this.#state.local.peek()?.get(path) === front ? this.#state.advertisedLocal.peek()?.get(path) : undefined, - }); + }; + hooks.attachAnnouncer(producer, announcer); const previous = created.get(path); created.set(path, front); @@ -1230,7 +1231,7 @@ export class Producer implements Table { }); void front.closed.then(release); - return producer; + return makePublication(producer, announcer); } #advertiseExact(path: Path.Valid, front: broadcast.Consumer, route: Route): void { @@ -1547,18 +1548,8 @@ let makeDynamic: ( let makeRequest: (path: Path.Valid, server: ServeState) => Request; let finishRequest: (request: Request, err: Error) => void; let setRequestPublisher: (request: Request, publisher: Producer | undefined) => void; -let makePrepared: ( - request: Request, - answer: broadcast.Producer, - announcer: broadcast.Announcer, - epoch: Epoch.Valid, -) => Prepared; -let publishPrepared: ( - request: Request, - answer: broadcast.Producer, - announcer: broadcast.Announcer, - epoch: Epoch.Valid, -) => broadcast.Producer; +let makePrepared: (request: Request, publication: Publication, epoch: Epoch.Valid) => Prepared; +let publishPrepared: (request: Request, publication: Publication, epoch: Epoch.Valid) => Publication; let rejectPrepared: (request: Request, err: Error) => void; /** @@ -2323,14 +2314,13 @@ export class Request { setRequestPublisher = (request, publisher) => { request.#publisher = publisher; }; - publishPrepared = (request, answer, announcer, epoch) => { + publishPrepared = (request, publication, epoch) => { if (!request.#server.has(request)) throw new Error("request is closed"); - hooks.attachAnnouncer(answer, announcer); // The exact route usually settles the request on its own, resolving its waiters // through the new epoch; answer it directly if it is still pending. - answer.announce({ epoch }); - if (request.#server.has(request)) request.#server.accept(request, answer.consume()); - return answer; + publication.announce({ epoch }); + if (request.#server.has(request)) request.#server.accept(request, publication.broadcast.consume()); + return publication; }; rejectPrepared = (request, err) => request.#server.reject(request, err); finishRequest = (request, err) => { @@ -2345,16 +2335,11 @@ export class Request { prepare(epoch: Epoch.Valid): Prepared { if (this.#done) throw new Error("request is closed"); if (!this.#publisher) throw new Error("forwarded requests cannot be prepared"); - const answer = this.#publisher.createBroadcast(this.path); - // Withheld until publish, so the answer cannot announce itself under another - // epoch or skip the request's admission. - const announcer = hooks.withholdAnnouncer(answer); - if (!announcer) { - answer.close(); - throw new Error("origin is closed"); - } + const publication = this.#publisher.createBroadcast(this.path); + // Only the fill handle escapes until publish, so the answer cannot announce itself + // under another epoch or skip the request's admission. this.#done = true; - return makePrepared(this, answer, announcer, epoch); + return makePrepared(this, publication, epoch); } // Resolve a session-forwarded request with its peer's routed broadcast, through @@ -2386,6 +2371,52 @@ export class Request { } } +let makePublication: (producer: broadcast.Producer, announcer: broadcast.Announcer) => Publication; + +/** An origin-owned broadcast and the capability to announce its route. */ +export class Publication { + /** The broadcast to fill with tracks before announcing it. */ + readonly broadcast: broadcast.Producer; + readonly #announcer: broadcast.Announcer; + + private constructor(producer: broadcast.Producer, announcer: broadcast.Announcer) { + this.broadcast = producer; + this.#announcer = announcer; + } + + static { + makePublication = (producer, announcer) => new Publication(producer, announcer); + } + + /** The route this broadcast is announced with, if any. */ + get route(): Route | undefined { + return this.#announcer.route(); + } + + /** + * Announce this broadcast, or replace its route; retain its epoch when changing only cost. + * Announce once its initial tracks exist. Throws after the broadcast closes or is replaced. + */ + announce( + route: + | Route + | { + epoch?: Route["epoch"]; + hops?: Route["hops"]; + cost?: Route["cost"]; + auth?: Route["auth"]; + } = Route.default, + ): void { + if (this.broadcast.closed.peek() !== undefined) throw new Error("broadcast is closed"); + this.#announcer.announce(Route.normalize(route)); + } + + /** Retract this broadcast's route until announced again, without closing its fill handle. */ + unannounce(): void { + this.#announcer.unannounce(); + } +} + /** * A claim answer being filled before its exact route is announced. * @@ -2397,37 +2428,32 @@ export class Prepared { /** The unannounced broadcast to fill before publishing; only {@link publish} announces it. */ readonly broadcast: broadcast.Producer; readonly #request: Request; - readonly #announcer: broadcast.Announcer; + readonly #publication: Publication; readonly #epoch: Epoch.Valid; readonly #stop: Dispose; #done = false; - private constructor( - request: Request, - answer: broadcast.Producer, - announcer: broadcast.Announcer, - epoch: Epoch.Valid, - ) { + private constructor(request: Request, publication: Publication, epoch: Epoch.Valid) { this.#request = request; - this.broadcast = answer; - this.#announcer = announcer; + this.broadcast = publication.broadcast; + this.#publication = publication; this.#epoch = epoch; // Synchronous, so a request made right after the broadcast closes reaches the claim again. - this.#stop = hooks.onClosed(answer, () => this.close()); + this.#stop = hooks.onClosed(this.broadcast, () => this.close()); } static { - makePrepared = (request, answer, announcer, epoch) => new Prepared(request, answer, announcer, epoch); + makePrepared = (request, publication, epoch) => new Prepared(request, publication, epoch); } /** Announce the filled broadcast and resolve every waiting requester, refusing them if it cannot be announced. */ - publish(): broadcast.Producer { + publish(): Publication { // Closing the broadcast closes this answer at once, so `#done` covers it too. if (this.#done) throw new Error("prepared answer is closed"); this.#done = true; this.#stop(); try { - return publishPrepared(this.#request, this.broadcast, this.#announcer, this.#epoch); + return publishPrepared(this.#request, this.#publication, this.#epoch); } catch (err) { this.broadcast.close(); rejectPrepared(this.#request, err instanceof Error ? err : new Error(String(err))); diff --git a/js/net/src/parked-expiry.test.ts b/js/net/src/parked-expiry.test.ts index f39d4329adb..8ad77a26948 100644 --- a/js/net/src/parked-expiry.test.ts +++ b/js/net/src/parked-expiry.test.ts @@ -31,8 +31,9 @@ for (const protocol of protocols) { const installed = clock.install(); const pair = createMockTransportPair(protocol.alpn); const origin = new Origin(); - const broadcast = origin.createBroadcast(Path.from("parked")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("parked")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const write = (sequence: number) => { const group = new Group(sequence); @@ -154,8 +155,9 @@ for (const protocol of protocols) { const installed = clock.install(); const pair = createMockTransportPair(protocol.alpn); const origin = new Origin(); - const broadcast = origin.createBroadcast(Path.from("parked")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("parked")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const write = (sequence: number, frames: number) => { const group = new Group(sequence); diff --git a/js/net/src/resume.test.ts b/js/net/src/resume.test.ts index 9218d0e658f..021f2509f9b 100644 --- a/js/net/src/resume.test.ts +++ b/js/net/src/resume.test.ts @@ -331,8 +331,9 @@ test("a resumed fork shares an on-demand track request", async () => { test("a lagging same-epoch replica resumes from one publisher's bytes exactly once", async () => { const { clock, installed } = manual(); const publisher = new Producer(); - const original = publisher.createBroadcast(path); - original.announce({ epoch: EPOCH }); + const originalPublication = publisher.createBroadcast(path); + const original = originalPublication.broadcast; + originalPublication.announce({ epoch: EPOCH }); const source = original.createTrack("seq"); const origin = new Producer(); diff --git a/js/net/src/retention.test.ts b/js/net/src/retention.test.ts index 3258f7cf165..24f73d49bc4 100644 --- a/js/net/src/retention.test.ts +++ b/js/net/src/retention.test.ts @@ -81,8 +81,9 @@ async function session(alpn: string) { accept({ transport: pair.server, url, publish: origin.consume() }), ]); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const remote = wireOf(client).consume(Path.from("test")); diff --git a/js/net/src/subscribe_priority.test.ts b/js/net/src/subscribe_priority.test.ts index 51c6d2617da..c2b4ac207c3 100644 --- a/js/net/src/subscribe_priority.test.ts +++ b/js/net/src/subscribe_priority.test.ts @@ -69,8 +69,9 @@ test.each([Lite.ALPN_07_WIP, ...Object.values(Ietf.ALPN)])( "%s follows the aggregate priority upstream", async (protocol) => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const [client, server] = await session(protocol, origin); @@ -108,8 +109,9 @@ test.each(Object.values(Ietf.ALPN))("%s ends a subscription mid-update without a await nextTask(); }); const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const [client, server] = await session(protocol, origin); let closed: Error | null | undefined; @@ -144,8 +146,9 @@ test.each(Object.values(Ietf.ALPN))("%s ends a subscription mid-update without a // update of defaults first, as on lite and in Rust. test.each(Object.values(Ietf.ALPN))("%s sends no update when the last subscriber leaves", async (protocol) => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const [client, server] = await session(protocol, origin); try { @@ -171,8 +174,9 @@ test.each(Object.values(Ietf.ALPN))("%s sends no update when the last subscriber // Drafts 14 and 15 require the next request ID, so the one the update spent ends the session. test.each(Object.values(Ietf.ALPN))("%s ends the track when an update fails to encode", async (protocol) => { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const [client, server] = await session(protocol, origin); let closed: Error | null | undefined; diff --git a/js/net/src/tail.test.ts b/js/net/src/tail.test.ts index 7c1bce00145..dfa55351175 100644 --- a/js/net/src/tail.test.ts +++ b/js/net/src/tail.test.ts @@ -26,8 +26,9 @@ async function session(protocol: string) { connect({ url, transport: pair.client }), accept({ transport: pair.server, url, publish: origin.consume() }), ]); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const video = broadcast.createTrack("video", { timescale: Timescale.MILLI }); const remote = wireOf(client).consume(Path.from("test")); const reader = remote.track("video").subscribe({ maxDelay: MAX_DELAY }).ordered(); diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index c78ac5e650c..a4a38b3b766 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -45,8 +45,9 @@ async function roundTrip(wire: Wire, timescale: Timescale | undefined, timestamp accept({ transport: pair.server, url, version: wire.version, publish: origin.consume() }), ]); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); // Every request is answered, including the TRACK lookup lite-05+ makes before subscribing. const serving = (async () => { for (;;) { @@ -114,8 +115,9 @@ test("a forwarded untimed track stays untimed", async () => { connect({ url, transport: downstream.client }), ]); - const broadcast = source.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = source.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const serving = (async () => { for (;;) { const request = await wireOf(broadcast).requested(); @@ -189,8 +191,9 @@ for (const wire of WIRES.filter((wire) => wire.name === "lite-06" || wire.name = accept({ transport: pair.server, url, publish: origin.consume() }), ]); - const broadcast = origin.createBroadcast(Path.from("test")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("test")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const producer = broadcast.createTrack("video", {}); const subscriber = wireOf(client).consume(Path.from("test")).track("video").subscribe(); diff --git a/js/publish/src/audio/encoder.test.ts b/js/publish/src/audio/encoder.test.ts index 2e040d5b01b..dd6cd153f61 100644 --- a/js/publish/src/audio/encoder.test.ts +++ b/js/publish/src/audio/encoder.test.ts @@ -701,12 +701,14 @@ test("audio failure stays local through demand changes and capture format re-arm } await feed.drain(); }; + let announced: Moq.Origin.Requesting | undefined; try { await configuring; await settle(); const net = broadcast.net.peek(); if (!net) throw new Error("expected network broadcast"); - const epoch = net.route?.epoch; + announced = broadcast.in.origin.peek()?.request(broadcast.in.name.peek(), { announced: true }); + const epoch = announced?.active.peek()?.epoch; const sibling = broadcast.video("video"); sibling.config.set({ codec: "avc1.640028", container: { kind: "legacy" } }); await settle(); @@ -744,7 +746,7 @@ test("audio failure stays local through demand changes and capture format re-arm expect(encoder.out.catalog.peek()?.enabled).toBe(false); expect(LaggingAudioEncoder.instances).toHaveLength(instances); expect(broadcast.net.peek()).toBe(net); - expect(net.route?.epoch).toBe(epoch); + expect(announced?.active.peek()?.epoch).toBe(epoch); expect(siblingTrack.appendGroup().sequence).toBe(1); const reconfiguring = configured(); @@ -764,7 +766,7 @@ test("audio failure stays local through demand changes and capture format re-arm expect(encoder.out.active.peek()).toBe(true); expect(recovered.closed.peek()).toBeUndefined(); expect(broadcast.net.peek()).toBe(net); - expect(net.route?.epoch).toBe(epoch); + expect(announced?.active.peek()?.epoch).toBe(epoch); expect(siblingTrack.appendGroup().sequence).toBe(2); } finally { subscriber?.close(); @@ -773,6 +775,7 @@ test("audio failure stays local through demand changes and capture format re-arm recovered?.close(); encoder.close(); bandwidth.close(); + announced?.close(); broadcast.close(); } }); diff --git a/js/publish/src/broadcast.test.ts b/js/publish/src/broadcast.test.ts index 9631e92b857..4a0661965a5 100644 --- a/js/publish/src/broadcast.test.ts +++ b/js/publish/src/broadcast.test.ts @@ -244,12 +244,13 @@ test("a re-announce continues the catalog's group sequence", async () => { const net = broadcast.net.peek(); if (!net) throw new Error("expected a network producer once connected"); + const announced = broadcast.in.origin.peek()?.request(broadcast.in.name.peek(), { announced: true }); const first = net.track(Broadcast.CATALOG_TRACK).subscribe().ordered(); expect((await first.nextGroup())?.sequence).toBe(0); announce.set(false); await settle(); - expect(net.route).toBeUndefined(); + expect(announced?.active.peek()).toBeUndefined(); broadcast.audio("audio").config.set(audioConfig); await settle(); announce.set(true); @@ -263,6 +264,7 @@ test("a re-announce continues the catalog's group sequence", async () => { expect(Object.keys(catalog.audio?.renditions ?? {})).toEqual(["audio"]); first.close(); + announced?.close(); broadcast.close(); }); @@ -464,7 +466,8 @@ test("a served decoder identity change moves the rendition to a fresh track name const net = broadcast.net.peek(); if (!net) throw new Error("expected a network producer once connected"); - const epoch = net.route?.epoch; + const announced = broadcast.in.origin.peek()?.request(broadcast.in.name.peek(), { announced: true }); + const epoch = announced?.active.peek()?.epoch; const catalog = new Json.Snapshot.Consumer({ track: net.track(Broadcast.CATALOG_TRACK).subscribe(), }); @@ -499,7 +502,7 @@ test("a served decoder identity change moves the rendition to a fresh track name expect(track.closed.peek()).toBeNull(); expect(video.track.peek()).toBeUndefined(); expect(broadcast.net.peek()).toBe(net); - expect(net.route?.epoch).toBe(epoch); + expect(announced?.active.peek()?.epoch).toBe(epoch); const second = net.track("video/1").subscribe(); await settle(); @@ -520,6 +523,7 @@ test("a served decoder identity change moves the rendition to a fresh track name second.close(); } finally { first.close(); + announced?.close(); broadcast.close(); } }); diff --git a/js/publish/src/broadcast.ts b/js/publish/src/broadcast.ts index 8e5562f058f..62c20977638 100644 --- a/js/publish/src/broadcast.ts +++ b/js/publish/src/broadcast.ts @@ -343,7 +343,8 @@ export class Broadcast { // Creating into the origin outlives any single session: a reconnect re-announces the // broadcast and new subscriptions land on the same producer. - const broadcast = origin.createBroadcast(name); + const broadcastPublication = origin.createBroadcast(name); + const broadcast = broadcastPublication.broadcast; effect.cleanup(() => broadcast.close()); // One publisher instance per broadcast, kept across unannounce and announce, so a // restart is a new broadcast rather than a resume into the old one. @@ -373,7 +374,7 @@ export class Broadcast { let serving = false; effect.run((inner) => { if (!inner.get(this.in.announce)) { - broadcast.unannounce(); + broadcastPublication.unannounce(); return; } @@ -402,7 +403,7 @@ export class Broadcast { } } - broadcast.announce({ epoch }); + broadcastPublication.announce({ epoch }); }); // Static tracks fan out to every subscriber. Keep the encoder-facing handle demand-gated diff --git a/js/publish/src/video/encoder.test.ts b/js/publish/src/video/encoder.test.ts index c5a663372b2..60dcebf392c 100644 --- a/js/publish/src/video/encoder.test.ts +++ b/js/publish/src/video/encoder.test.ts @@ -1060,11 +1060,13 @@ test("a failed video encoder disables only its rendition and recovers on a fresh let siblingSubscriber: Moq.Track.Subscriber | undefined; let catalogSubscriber: Moq.Track.Subscriber | undefined; let recovered: Moq.Track.Subscriber | undefined; + let announced: Moq.Origin.Requesting | undefined; try { await encoder.out.catalog.changed(); const net = broadcast.net.peek(); if (!net) throw new Error("expected network broadcast"); - const epoch = net.route?.epoch; + announced = broadcast.in.origin.peek()?.request(broadcast.in.name.peek(), { announced: true }); + const epoch = announced?.active.peek()?.epoch; const sibling = broadcast.video("video/sd"); sibling.config.set({ codec: "avc1.640028", container: { kind: "legacy" } }); await flush(); @@ -1089,7 +1091,7 @@ test("a failed video encoder disables only its rendition and recovers on a fresh expect((await catalog.latest())?.value.video?.renditions["video/hd"]?.enabled).toBe(false); expect(encoder.out.active.peek()).toBe(false); expect(broadcast.net.peek()).toBe(net); - expect(net.route?.epoch).toBe(epoch); + expect(announced?.active.peek()?.epoch).toBe(epoch); expect(siblingTrack.closed.peek()).toBeUndefined(); expect(siblingTrack.appendGroup().sequence).toBe(1); const instances = FakeVideoEncoder.instances.length; @@ -1133,7 +1135,7 @@ test("a failed video encoder disables only its rendition and recovers on a fresh expect(encoder.out.active.peek()).toBe(true); expect(recovered.closed.peek()).toBeUndefined(); expect(broadcast.net.peek()).toBe(net); - expect(net.route?.epoch).toBe(epoch); + expect(announced?.active.peek()?.epoch).toBe(epoch); expect(siblingTrack.appendGroup().sequence).toBe(2); } finally { subscriber?.close(); @@ -1142,6 +1144,7 @@ test("a failed video encoder disables only its rendition and recovers on a fresh recovered?.close(); encoder.close(); bandwidth.close(); + announced?.close(); broadcast.close(); } }); diff --git a/js/watch/src/audio/decoder.test.ts b/js/watch/src/audio/decoder.test.ts index 1c449b169a0..f01bae01c47 100644 --- a/js/watch/src/audio/decoder.test.ts +++ b/js/watch/src/audio/decoder.test.ts @@ -948,7 +948,7 @@ async function livePlayback(burst: number) { // Serve the same rendition from another path, as a `broadcast` override does. move(path: string) { media.peek().close(); - producer = new Moq.Origin.Producer().createBroadcast(Moq.Path.from(path)); + producer = new Moq.Origin.Producer().createBroadcast(Moq.Path.from(path)).broadcast; track = producer.createTrack("audio", { timescale: Moq.Time.Timescale.MILLI }); sequence = 0; media.set(producer.consume()); diff --git a/js/watch/src/broadcast.test.ts b/js/watch/src/broadcast.test.ts index 0deccde227c..ed76cf9a62f 100644 --- a/js/watch/src/broadcast.test.ts +++ b/js/watch/src/broadcast.test.ts @@ -8,8 +8,9 @@ import { Effect, Signal } from "@moq/signals"; import { Broadcast } from "./broadcast"; function publish(origin: Origin.Producer, path: Path.Valid) { - const broadcast = origin.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); return broadcast; } @@ -39,7 +40,7 @@ function serve(claim: Origin.Dynamic, epoch?: Moq.Epoch.Valid): Moq.Broadcast.Pr const answers: Moq.Broadcast.Producer[] = []; void (async () => { for await (const request of claim.requested()) - answers.push(request.prepare(epoch ?? Moq.Epoch.mint()).publish()); + answers.push(request.prepare(epoch ?? Moq.Epoch.mint()).publish().broadcast); })(); return answers; } @@ -338,10 +339,12 @@ describe("relativeBroadcast", () => { it("reads a reference only from the epoch it names, best effort without one", async () => { const owner = new Origin.Producer(); const [catalogEpoch, siblingEpoch, other] = [Moq.Epoch.mint(), Moq.Epoch.mint(), Moq.Epoch.mint()]; - const catalog = owner.createBroadcast(Path.from("a/b")); - catalog.announce({ epoch: catalogEpoch }); - const sibling = owner.createBroadcast(Path.from("a/source")); - sibling.announce({ epoch: siblingEpoch }); + const catalogPublication = owner.createBroadcast(Path.from("a/b")); + const catalog = catalogPublication.broadcast; + catalogPublication.announce({ epoch: catalogEpoch }); + const siblingPublication = owner.createBroadcast(Path.from("a/source")); + const sibling = siblingPublication.broadcast; + siblingPublication.announce({ epoch: siblingEpoch }); const plain = publish(owner, Path.from("a/plain")); const source = new Broadcast({ origin: owner, @@ -380,8 +383,9 @@ describe("relativeBroadcast", () => { const owner = new Origin.Producer(); const main = publish(owner, Path.from("a/b")); const [oldEpoch, newEpoch] = [Moq.Epoch.mint(), Moq.Epoch.mint()]; - const old = owner.createBroadcast(Path.from("a/source")); - old.announce({ epoch: oldEpoch }); + const oldPublication = owner.createBroadcast(Path.from("a/source")); + const old = oldPublication.broadcast; + oldPublication.announce({ epoch: oldEpoch }); const source = new Broadcast({ origin: owner, name: Path.from("a/b"), @@ -405,8 +409,9 @@ describe("relativeBroadcast", () => { // The announcement is the online signal: the plain reference follows the restart, // while one naming the old instance never reads the new one. - const replacement = owner.createBroadcast(Path.from("a/source")); - replacement.announce({ epoch: newEpoch }); + const replacementPublication = owner.createBroadcast(Path.from("a/source")); + const replacement = replacementPublication.broadcast; + replacementPublication.announce({ epoch: newEpoch }); await settle(); expect(following?.epoch).toBe(newEpoch); expect(pinned).toBeUndefined(); @@ -424,8 +429,9 @@ describe("relativeBroadcast", () => { const owner = new Origin.Producer(); const main = publish(owner, Path.from("a/b")); const [oldEpoch, newEpoch] = [Moq.Epoch.mint(), Moq.Epoch.mint()]; - const old = owner.createBroadcast(Path.from("a/source")); - old.announce({ epoch: oldEpoch }); + const oldPublication = owner.createBroadcast(Path.from("a/source")); + const old = oldPublication.broadcast; + oldPublication.announce({ epoch: oldEpoch }); const named = { ...video("avc1.64001f", "./source"), epoch: oldEpoch } as Catalog.VideoConfig; const source = new Broadcast({ origin: owner, @@ -438,8 +444,9 @@ describe("relativeBroadcast", () => { await settle(); expect(videoRenditions(source)).toEqual(["named", "plain"]); - const replacement = owner.createBroadcast(Path.from("a/source")); - replacement.announce({ epoch: newEpoch }); + const replacementPublication = owner.createBroadcast(Path.from("a/source")); + const replacement = replacementPublication.broadcast; + replacementPublication.announce({ epoch: newEpoch }); await withoutWarnings(settle); expect(videoRenditions(source)).toEqual(["plain"]); replacement.close(); @@ -491,7 +498,7 @@ describe("blind resolution", () => { const requests = route.requested(); const next = await requests.next(); if (!next.value) throw new Error("expected a request"); - const upstream = next.value.prepare(Moq.Epoch.mint()).publish(); + const upstream = next.value.prepare(Moq.Epoch.mint()).publish().broadcast; await settle(); const active = source.out.active.peek(); @@ -608,8 +615,9 @@ describe("restart", () => { expect(old).toBeDefined(); // A newer publisher announces the path, which now serves the name: a restart. - const exact = owner.createBroadcast(name); - exact.announce({ epoch: Moq.Epoch.mint() }); + const exactPublication = owner.createBroadcast(name); + const exact = exactPublication.broadcast; + exactPublication.announce({ epoch: Moq.Epoch.mint() }); await settle(); const moved = source.out.active.peek(); expect(moved).toBeDefined(); @@ -632,8 +640,9 @@ describe("restart", () => { const owner = new Origin.Producer(); const name = Path.from("pool/job.hang"); const epoch = Moq.Epoch.mint(); - const exact = owner.createBroadcast(name); - exact.announce({ epoch }); + const exactPublication = owner.createBroadcast(name); + const exact = exactPublication.broadcast; + exactPublication.announce({ epoch }); const source = new Broadcast({ origin: owner, name, enabled: true, announced: true, catalogFormat: "manual" }); await settle(); const old = source.out.active.peek(); @@ -1039,7 +1048,7 @@ describe("wildcard renditions", () => { prepared.broadcast .createTrack("video", { timescale: Moq.Time.Timescale.MILLI }) .writeFrame({ payload: new TextEncoder().encode("frame"), timestamp: Moq.Time.Timestamp.now() }); - const produced = prepared.publish(); + const produced = prepared.publish().broadcast; await settle(); expect(active).toBeDefined(); const track = active?.track("video").subscribe().ordered(); diff --git a/js/watch/src/video/decoder.test.ts b/js/watch/src/video/decoder.test.ts index e9927a0d5f7..e69bbdc6963 100644 --- a/js/watch/src/video/decoder.test.ts +++ b/js/watch/src/video/decoder.test.ts @@ -366,7 +366,7 @@ async function guardedPlayback( republishMedia(path?: string): Moq.Track.Producer | undefined { media.peek().close(); mediaProducer = path - ? new Moq.Origin.Producer().createBroadcast(Moq.Path.from(path)) + ? new Moq.Origin.Producer().createBroadcast(Moq.Path.from(path)).broadcast : new Moq.Broadcast.Producer(); const track = live ? mediaProducer.createTrack("video", { timescale: Moq.Time.Timescale.MILLI }) diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index 955517b3214..6270fbe2c91 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -77,7 +77,6 @@ Rejected in planning: ## Required - [Two broadcast resolutions](/quest/m1/api-audit/resolve.md) - Rust, JS, and the bindings resolve a broadcast with `request(path, epoch)` or `follow(path, epoch)`, and every racy two-step caller is gone -- [JS origin publications](/quest/m1/api-audit/js-publication.md) - `@moq/net` origins hand out a `Publication` that announces, so `Prepared` and standalone broadcasts lose methods that always throw - [Binding publications](/quest/m1/api-audit/binding-publication.md) - moq-ffi, the C ABI, and every wrapper hand out a publication handle that announces, replacing the private adapters - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them - [Session-only items stay private](/quest/m1/api-audit/model-private.md) - model items only the built-in sessions use are private diff --git a/quest/m1/api-audit/binding-publication.md b/quest/m1/api-audit/binding-publication.md index f4692196473..21d83165e8f 100644 --- a/quest/m1/api-audit/binding-publication.md +++ b/quest/m1/api-audit/binding-publication.md @@ -13,7 +13,7 @@ prepared broadcast no longer exposes announcement calls that can only fail. Decided 2026-10-23 (maintainer, on [moq-dev/moq#6464](https://github.com/moq-dev/moq/pull/6464)): ✅ land the announcement split in three PRs: native Rust in that PR, the -[JS mirror](/quest/m1/api-audit/js-publication.md), and this binding rollout +[JS mirror](https://github.com/moq-dev/moq/pull/6795), and this binding rollout once the native PR merges. The native PR keeps private adapters that hold the `Publication` beside each binding fill handle (`rs/moq-ffi/src/producer.rs` `from_publication`, `rs/moq-c/src/publish.rs` `create_publication` and diff --git a/quest/m1/api-audit/js-publication.md b/quest/m1/api-audit/js-publication.md deleted file mode 100644 index 93883fbf44e..00000000000 --- a/quest/m1/api-audit/js-publication.md +++ /dev/null @@ -1,39 +0,0 @@ -# [M] JS origin publications - -## Goal - -A `@moq/net` `broadcast.Producer` fills tracks; it cannot announce. Creating a -broadcast on an origin and publishing a `Prepared` answer return a -`Publication` that carries `announce`, `unannounce`, and `route`, so a -standalone broadcast and `prepared.broadcast` no longer expose announcement -calls that can only throw. Mirrors Rust's `origin::Publication`. - -## Plan - -Decided 2026-10-23 (maintainer, on -[moq-dev/moq#6464](https://github.com/moq-dev/moq/pull/6464)): ✅ land the -announcement split in three PRs: native Rust in that PR, this `@moq/net` -mirror, and the [binding publications](/quest/m1/api-audit/binding-publication.md). -This one does not depend on the native PR and can start now. - -The shape is Rust's, approved on the same PR: an origin-owned record holding -the fill broadcast and its announcement capability. - -- `Producer.createBroadcast` (`js/net/src/origin.ts`) and `Prepared.publish` - return `Publication`. It exposes the fill producer as `broadcast`, plus - `announce(route)`, `unannounce()`, and `route`. No way to swap in another - producer, and no `finish`: a broadcast has no finished state. -- `broadcast.Producer` loses its public `announce`, `unannounce`, and `route` - (`js/net/src/broadcast.ts`); the announcer stays a module hook. -- Lifetime matches Rust: publication clones (or references) and fill handles - keep the existing broadcast lifetime; dropping a `Publication` does not - retract on its own. `unannounce` retracts sooner. -- Migrate every caller: `js/net` tests, benches, and examples, `js/publish`, - `js/moq-boy`, the `js/e2ee` docs, `js/watch` and `js/hang` tests, and - `demo/carla`. Add the `js/net/MIGRATE.md` entry and update - `doc/lib/js/net.md`. - -Test: clone/drop, explicit unannounce, Prepared admission, -and a stale publication against a replacement under a new epoch. - -Public API: breaking in `@moq/net`. Wire: none. diff --git a/quest/m1/broadcast-lifecycle/README.md b/quest/m1/broadcast-lifecycle/README.md index dd7df0a96fa..effc1e6ad3c 100644 --- a/quest/m1/broadcast-lifecycle/README.md +++ b/quest/m1/broadcast-lifecycle/README.md @@ -55,4 +55,3 @@ timer on main at 1582fdc07d: - [moq-transport finish interop](/quest/m2/ietf-finish.md) - the interop `finish` case, whose late readers rely on retention - [CLI and gst flags](https://github.com/moq-dev/moq/pull/6522) - removed the opt-in `linger` booleans, leaving completion to this epic -- [JS origin publications](/quest/m1/api-audit/js-publication.md) and [binding publications](/quest/m1/api-audit/binding-publication.md) - mirror `origin::Publication`, which has no `finish` diff --git a/test/audio-quality/clients/js/src/capture.test.ts b/test/audio-quality/clients/js/src/capture.test.ts index 9e3a0d1053e..b50854e607b 100644 --- a/test/audio-quality/clients/js/src/capture.test.ts +++ b/test/audio-quality/clients/js/src/capture.test.ts @@ -74,9 +74,10 @@ test("arrival capture refuses an unsupported container", () => { test("the observer shares playback demand and closes without ending playback", async () => { const origin = new Origin.Producer(); - const published = origin.createBroadcast(Path.from("tone.hang")); + const publishedPublication = origin.createBroadcast(Path.from("tone.hang")); + const published = publishedPublication.broadcast; const track = published.createTrack("audio", { timescale: Time.Timescale.MILLI }); - published.announce(); + publishedPublication.announce(); const source = new Broadcast({ origin, name: Path.from("tone.hang"), announced: false, catalogFormat: "manual" }); const capture = new Capture({ broadcast: new Signal(source), @@ -116,9 +117,10 @@ test("at auto the observer asks for the decoder's 2 s budget, not the target", a // A capture cut to the target would drop the frames later than it, which are what the target // is measured from, so a trace recorded that way can never show the player falling short. const origin = new Origin.Producer(); - const published = origin.createBroadcast(Path.from("tone.hang")); + const publishedPublication = origin.createBroadcast(Path.from("tone.hang")); + const published = publishedPublication.broadcast; const track = published.createTrack("audio", { timescale: Time.Timescale.MILLI }); - published.announce(); + publishedPublication.announce(); const source = new Broadcast({ origin, name: Path.from("tone.hang"), announced: false, catalogFormat: "manual" }); const capture = new Capture({ broadcast: new Signal(source), @@ -145,9 +147,10 @@ test("at auto the observer asks for the decoder's 2 s budget, not the target", a /** Capture a legacy track, then end its first group with `end` once that group's reader is in flight. */ async function captureGroupEnd(end: (group: Group.Producer) => void): Promise { const origin = new Origin.Producer(); - const published = origin.createBroadcast(Path.from("tone.hang")); + const publishedPublication = origin.createBroadcast(Path.from("tone.hang")); + const published = publishedPublication.broadcast; const track = published.createTrack("audio", { timescale: Time.Timescale.MILLI }); - published.announce(); + publishedPublication.announce(); const source = new Broadcast({ origin, name: Path.from("tone.hang"), announced: false, catalogFormat: "manual" }); const capture = new Capture({ broadcast: new Signal(source), diff --git a/test/drain/drain.ts b/test/drain/drain.ts index 834381e909a..7572e1dfb5a 100644 --- a/test/drain/drain.ts +++ b/test/drain/drain.ts @@ -109,10 +109,11 @@ await new Promise((resolve, reject) => { // ── publisher on B ──────────────────────────────────────────────────────────── const published = new Moq.Origin.Producer(); -const broadcast = published.createBroadcast(path); +const broadcastPublication = published.createBroadcast(path); +const broadcast = broadcastPublication.broadcast; const track = broadcast.createTrack(trackName, { timescale: Moq.Time.Timescale.MILLI }); // The epoch is what says B's route serves the same broadcast as A's. -broadcast.announce({ epoch: Moq.Epoch.mint() }); +broadcastPublication.announce({ epoch: Moq.Epoch.mint() }); const publisher = new Moq.Connection({ url: new URL(`http://127.0.0.1:${bPort}/`), publish: published.consume(), diff --git a/test/interop/clients/js-native/control.ts b/test/interop/clients/js-native/control.ts index 8493f3d4c4e..35382ba4f1b 100644 --- a/test/interop/clients/js-native/control.ts +++ b/test/interop/clients/js-native/control.ts @@ -78,10 +78,11 @@ if (mode === "status") { } else { assert(legacy, "request credit only exists on drafts 14 and 16"); const origin = new Moq.Origin.Producer(); - const broadcast = origin.createBroadcast(Moq.Path.from("control")); + const broadcastPublication = origin.createBroadcast(Moq.Path.from("control")); + const broadcast = broadcastPublication.broadcast; const first = broadcast.createTrack("first"); const second = broadcast.createTrack("second"); - broadcast.announce(); + broadcastPublication.announce(); const connection = new Connection({ url, quic, diff --git a/test/interop/clients/js-native/server.ts b/test/interop/clients/js-native/server.ts index e44b127ac69..a362a808bcb 100644 --- a/test/interop/clients/js-native/server.ts +++ b/test/interop/clients/js-native/server.ts @@ -22,17 +22,19 @@ const legacy = version === Ietf.Version.DRAFT_14 || version === Ietf.Version.DRA const alpns = version === Ietf.Version.DRAFT_14 ? ["moql", "moq-00"] : [`moqt-${Number(version) - 0xff000000}`]; const extensions = Extensions.offered(); const origin = new Moq.Origin.Producer(); -const broadcast = origin.createBroadcast(Moq.Path.from("handover")); +const broadcastPublication = origin.createBroadcast(Moq.Path.from("handover")); +const broadcast = broadcastPublication.broadcast; const firstTrack = broadcast.createTrack("first"); const secondTrack = broadcast.createTrack("second"); const video = broadcast.createTrack("video"); -broadcast.announce({ epoch: Moq.Epoch.mint() }); -const marker = origin.createBroadcast(Moq.Path.from("replacement")); +broadcastPublication.announce({ epoch: Moq.Epoch.mint() }); +const markerPublication = origin.createBroadcast(Moq.Path.from("replacement")); +const marker = markerPublication.broadcast; const ready = marker.createTrack("ready"); const markerGroup = ready.appendGroup(); markerGroup.writeFrame({ payload: new Uint8Array([42]) }); markerGroup.close(); -marker.announce(); +markerPublication.announce(); type Socket = { protocol: string; diff --git a/test/interop/clients/js-native/tail.ts b/test/interop/clients/js-native/tail.ts index a3cdc4154ec..7639b92543e 100644 --- a/test/interop/clients/js-native/tail.ts +++ b/test/interop/clients/js-native/tail.ts @@ -43,9 +43,10 @@ const connection = await Moq.Connection.connect({ }); try { if (role === "publish") { - const broadcast = origin.createBroadcast(Moq.Path.from(path)); + const broadcastPublication = origin.createBroadcast(Moq.Path.from(path)); + const broadcast = broadcastPublication.broadcast; const track = broadcast.createTrack("tail", { compress, ...(status ? { priority: 2 } : {}) }); - broadcast.announce(); + broadcastPublication.announce(); const lines = createInterface({ input: process.stdin }); const commands = lines[Symbol.asyncIterator](); if (!status) { diff --git a/test/interop/clients/js/src/finish.ts b/test/interop/clients/js/src/finish.ts index 57c4b99b751..0e89dcf9fed 100644 --- a/test/interop/clients/js/src/finish.ts +++ b/test/interop/clients/js/src/finish.ts @@ -70,7 +70,8 @@ export function finish( state.result = await readFinish(origin, path, frames, Milli(timeout)); return; } - const broadcast = origin.createBroadcast(Moq.Path.from(path)); + const broadcastPublication = origin.createBroadcast(Moq.Path.from(path)); + const broadcast = broadcastPublication.broadcast; const config: Catalog.VideoConfig = { codec: "avc1.42001e", codedWidth: Catalog.u53(320), @@ -132,19 +133,20 @@ export function finish( serving, ); serving.close(); - broadcast.announce(); + broadcastPublication.announce(); // Keep the finished broadcast up for the late readers, then take it offline. state.lingerUs = LINGER * 1000; state.ready = true; await sleep(Nano.fromMilli(LINGER)); broadcast.close(); - const health = origin.createBroadcast(Moq.Path.from(`${path}.health`)); + const healthPublication = origin.createBroadcast(Moq.Path.from(`${path}.health`)); + const health = healthPublication.broadcast; const healthTrack = health.createTrack("live"); const group = healthTrack.appendGroup(); group.writeFrame({ payload: new TextEncoder().encode("session-alive") }); group.close(); healthTrack.close(); - health.announce(); + healthPublication.announce(); state.ended = true; await ack; health.close(); diff --git a/test/interop/compat/client.ts b/test/interop/compat/client.ts index eddb6343fd9..763411ee61c 100644 --- a/test/interop/compat/client.ts +++ b/test/interop/compat/client.ts @@ -89,7 +89,7 @@ const AUDIO_BUDGET = 5000; /** The subscription field holding the latency budget, which the checkout renamed from `maxAge`. */ function budgetField(): "maxAge" | "maxDelay" { // The checkout refuses the old name, which a throwaway local track shows without a session. - const broadcast = new Moq.Origin.Producer().createBroadcast(Moq.Path.from("probe")); + const broadcast = new Moq.Broadcast.Producer(); try { broadcast .createTrack("probe") @@ -287,9 +287,11 @@ if (command === "sign") { } else if (command === "publish") { install(); const origin = new Moq.Origin.Producer(); - const broadcast = origin.createBroadcast(Moq.Path.from(args[1])); + const publication = origin.createBroadcast(Moq.Path.from(args[1])); + // Released packages return the fill producer directly; current ones return its publication. + const broadcast = ("broadcast" in publication ? publication.broadcast : publication) as Moq.Broadcast.Producer; const track = broadcast.createTrack("data", { timescale: Moq.Time.Timescale.MICRO }); - broadcast.announce(); + publication.announce(); const connection = await connect({ url: new URL(args[0]), publish: origin.consume(), diff --git a/test/interop/track-status.ts b/test/interop/track-status.ts index 87f8bac7663..2e3cd324988 100644 --- a/test/interop/track-status.ts +++ b/test/interop/track-status.ts @@ -20,8 +20,9 @@ const version = (0xff000000 + input.draft) as IetfVersion; const pair = createMockTransportPair(); const origin = new OriginProducer(); -const broadcast = origin.createBroadcast(Path.from("room")); -broadcast.announce(); +const broadcastPublication = origin.createBroadcast(Path.from("room")); +const broadcast = broadcastPublication.broadcast; +broadcastPublication.announce(); const track = broadcast.createTrack("video", { timescale: Timescale.MILLI, maxAge: Milli(5000), priority: 2 }); const group = new GroupProducer(4); group.writeFrame({ payload: new Uint8Array([1]), timestamp: Timestamp.fromMillis(0) }); diff --git a/test/interop/tracks.ts b/test/interop/tracks.ts index bc32f3d4170..a0ccec2deb6 100644 --- a/test/interop/tracks.ts +++ b/test/interop/tracks.ts @@ -23,8 +23,9 @@ const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); if (input.mode === "answer") { const origin = new OriginProducer(); - const broadcast = origin.createBroadcast(Path.from("room")); - broadcast.announce(); + const broadcastPublication = origin.createBroadcast(Path.from("room")); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); broadcast.createTrack("a", { timescale: Timescale.MILLI, maxAge: Milli(3000), priority: 9, refresh: 4 }); broadcast.createTrack("b"); diff --git a/test/max-age/client.ts b/test/max-age/client.ts index 4b33dd47fc2..64b10455b86 100644 --- a/test/max-age/client.ts +++ b/test/max-age/client.ts @@ -14,9 +14,10 @@ if (!address || !protocol || !["publish", "subscribe"].includes(mode)) const origin = new Producer(); const ages = [undefined, Milli(0), Milli(30_000)]; if (mode === "publish") { - const broadcast = origin.createBroadcast(Path.from("age")); + const broadcastPublication = origin.createBroadcast(Path.from("age")); + const broadcast = broadcastPublication.broadcast; for (const [i, maxAge] of ages.entries()) broadcast.createTrack(String(i), { maxAge }); - broadcast.announce(); + broadcastPublication.announce(); } const url = new URL(address); const transport = new Session(url, { diff --git a/test/p2p/src/main.ts b/test/p2p/src/main.ts index 84dea2b2e91..0b903812534 100644 --- a/test/p2p/src/main.ts +++ b/test/p2p/src/main.ts @@ -50,8 +50,9 @@ window.checkTransport = async () => { const publishing = new Moq.Origin.Producer(); const consuming = new Moq.Origin.Producer(); const path = Moq.Path.from("peer/data"); - const broadcast = publishing.createBroadcast(path); - broadcast.announce(); + const broadcastPublication = publishing.createBroadcast(path); + const broadcast = broadcastPublication.broadcast; + broadcastPublication.announce(); const track = broadcast.createTrack("data", { timescale: Moq.Time.Timescale.MILLI }); let client: Awaited> | undefined; let server: Awaited> | undefined; diff --git a/test/receive-credit/sender.ts b/test/receive-credit/sender.ts index b2fb5d7b3cb..ece438d052d 100644 --- a/test/receive-credit/sender.ts +++ b/test/receive-credit/sender.ts @@ -32,10 +32,11 @@ if (!address || !version || !certificate || !Number.isInteger(window) || window } const origin = new Moq.Origin.Producer(); -const broadcast = origin.createBroadcast(Moq.Path.from("room")); +const broadcastPublication = origin.createBroadcast(Moq.Path.from("room")); +const broadcast = broadcastPublication.broadcast; const low = broadcast.createTrack("low"); const high = broadcast.createTrack("high"); -broadcast.announce(); +broadcastPublication.announce(); const session = await Moq.Connection.connect({ url: new URL(address), diff --git a/test/video/src/captions.ts b/test/video/src/captions.ts index a28f6a03349..d23969bc455 100644 --- a/test/video/src/captions.ts +++ b/test/video/src/captions.ts @@ -11,7 +11,8 @@ import { Source } from "../../../js/watch/src/text/source"; /** Caption selection controls exercised by the browser harness. */ export function captions() { const origin = new Origin.Producer(); - const producer = origin.createBroadcast(Path.from("captions")); + const producerPublication = origin.createBroadcast(Path.from("captions")); + const producer = producerPublication.broadcast; const textCatalog: NonNullable = { renditions: {} }; const catalog: Catalog.Root = { text: textCatalog }; for (const [name, text] of [ @@ -24,7 +25,7 @@ export function captions() { group.writeFrame({ payload: Legacy.encodeFrame(new TextEncoder().encode(text), Micro(1_000_000)) }); group.close(); } - producer.announce(); + producerPublication.announce(); const broadcast = new Broadcast({ origin, name: Path.from("captions"), catalogFormat: "manual", catalog }); const target = new Signal(undefined); const source = new Source({ broadcast, target }); diff --git a/test/video/src/poster.ts b/test/video/src/poster.ts index 18a50b0bd11..81a8707db9e 100644 --- a/test/video/src/poster.ts +++ b/test/video/src/poster.ts @@ -8,7 +8,8 @@ import { Player } from "@moq/watch"; async function poster(audio: boolean, buffered: boolean) { const origin = new Origin.Producer(); const name = Path.from("poster"); - const producer = origin.createBroadcast(name); + const producerPublication = origin.createBroadcast(name); + const producer = producerPublication.broadcast; const track = producer.createTrack("video"); const group = track.appendGroup(); const audioTrack = audio ? producer.createTrack("audio") : undefined; @@ -37,7 +38,7 @@ async function poster(audio: boolean, buffered: boolean) { }, }), }; - producer.announce(); + producerPublication.announce(); const canvas = document.createElement("canvas"); document.body.append(canvas); const player = new Player({ diff --git a/test/wasm/baseline/publisher.ts b/test/wasm/baseline/publisher.ts index 2a0ade8def1..ec655d2a833 100644 --- a/test/wasm/baseline/publisher.ts +++ b/test/wasm/baseline/publisher.ts @@ -8,10 +8,11 @@ async function prepare(config: Workload): Promise { const fixtures = Array.from({ length: config.frames }, (_, frame) => payload(config.size, frame)); for (let index = 0; index < config.publishers; index++) { const origin = new Moq.Origin.Producer(); - const broadcast = origin.createBroadcast(Moq.Path.from(`${config.id}/${index}`)); + const broadcastPublication = origin.createBroadcast(Moq.Path.from(`${config.id}/${index}`)); + const broadcast = broadcastPublication.broadcast; const track = broadcast.createTrack("data", { timescale: Moq.Time.Timescale.MILLI }); const demand = track.demand().wanted; - broadcast.announce(); + broadcastPublication.announce(); const connection = await Moq.Connection.connect({ url: new URL(config.url), publish: origin.consume(), diff --git a/test/wasm/src/main.ts b/test/wasm/src/main.ts index 2412828ded7..d04db898604 100644 --- a/test/wasm/src/main.ts +++ b/test/wasm/src/main.ts @@ -143,12 +143,13 @@ async function withPublisher( // The session serves an origin rather than individual broadcasts, so the path is // published into the origin and the session announces the table. const origin = new Moq.Origin.Producer(); - const broadcast = origin.createBroadcast(Moq.Path.from(path)); + const broadcastPublication = origin.createBroadcast(Moq.Path.from(path)); + const broadcast = broadcastPublication.broadcast; const track = broadcast.createTrack(TRACK, { timescale: Moq.Time.Timescale.MILLI }); const missing = broadcast.createTrack("missing"); const used = track.demand().wanted; const missingUsed = missing.demand().wanted; - broadcast.announce(); + broadcastPublication.announce(); const connection = await Moq.Connection.connect({ url: new URL(relay.url), publish: origin.consume() }); let stopped = false;