diff --git a/Cargo.lock b/Cargo.lock index 08130ec9..9047d301 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -435,30 +435,6 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" -[[package]] -name = "aws-lc-rs" -version = "1.18.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "faac5829c2b74c28f830747e7818ccfb684261b5f48a1118b1e2a13d36dfab13" -dependencies = [ - "aws-lc-sys", - "untrusted 0.7.1", - "zeroize", -] - -[[package]] -name = "aws-lc-sys" -version = "0.46.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1622d8446a2d4b2ce0c7eefc73dd43a99779028d5ee5c2dd8073a658ba8a2bc" -dependencies = [ - "cc", - "cmake", - "dunce", - "fs_extra", - "pkg-config", -] - [[package]] name = "axum" version = "0.8.9" @@ -1116,12 +1092,6 @@ dependencies = [ "cc", ] -[[package]] -name = "cmov" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" - [[package]] name = "cobs" version = "0.3.0" @@ -1494,15 +1464,6 @@ dependencies = [ "syn 3.0.6", ] -[[package]] -name = "ctutils" -version = "0.4.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe3e9d21d39e6f39ff46e03d5b621424f88ee23d8a1672c6bfc9ccc7f2da4f6a" -dependencies = [ - "cmov", -] - [[package]] name = "cudarc" version = "0.19.10" @@ -1745,7 +1706,6 @@ dependencies = [ "block-buffer", "const-oid 0.10.2", "crypto-common", - "ctutils", ] [[package]] @@ -1860,12 +1820,6 @@ dependencies = [ "dtoa", ] -[[package]] -name = "dunce" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" - [[package]] name = "duplicate" version = "2.0.1" @@ -2409,12 +2363,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "fs_extra" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" - [[package]] name = "fsevent-sys" version = "4.1.0" @@ -2553,10 +2501,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", - "js-sys", "libc", "wasi", - "wasm-bindgen", ] [[package]] @@ -2578,11 +2524,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", - "js-sys", "libc", "r-efi 6.0.0", "rand_core 0.10.1", - "wasm-bindgen", ] [[package]] @@ -2942,15 +2886,6 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5491a308e0214554f07a81d8944abe45f552871c12e3c3c6e7e5d354039a6c4c" -[[package]] -name = "hmac" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" -dependencies = [ - "digest", -] - [[package]] name = "html-escape" version = "0.2.15" @@ -4119,9 +4054,8 @@ dependencies = [ "pkcs8 0.11.0", "rand", "reqwest", - "rind", + "rind 0.1.12", "rind-jsonwebtoken", - "rind-rustls", "rustls", "rustls-platform-verifier", "serde", @@ -4147,9 +4081,16 @@ dependencies = [ name = "moq-crypto" version = "0.0.11" dependencies = [ + "bytes", "data-encoding", + "kio", + "moq-net", + "rind 0.1.12", + "rind-rustls", + "rustls", "serde", "thiserror 2.0.21", + "zeroize", ] [[package]] @@ -4234,11 +4175,13 @@ dependencies = [ "num_enum", "rand", "rangemap", + "rind 0.1.12", "serde", "slab", "thiserror 2.0.21", "tracing", "url", + "zeroize", ] [[package]] @@ -4261,14 +4204,15 @@ dependencies = [ name = "moq-quic" version = "0.1.0" dependencies = [ - "aws-lc-rs", "bytes", "fastbloom", "indexmap 2.14.2", "kio", "lru-slab", "moq-congestion", + "moq-crypto", "rand", + "rind 0.1.12", "rustc-hash 2.1.3", "rustls", "slab", @@ -4282,7 +4226,6 @@ name = "moq-relay" version = "0.17.2" dependencies = [ "anyhow", - "aws-lc-rs", "axum", "axum-server", "bytes", @@ -4294,6 +4237,7 @@ dependencies = [ "hyper-util", "kio", "moq-auth", + "moq-crypto", "moq-net", "moq-quic", "moq-stats", @@ -4302,6 +4246,7 @@ dependencies = [ "rand", "reqwest", "reqwest-middleware", + "rind 0.1.12", "rustls", "sd-notify", "serde", @@ -4451,7 +4396,6 @@ dependencies = [ "dns-lookup", "futures", "hex", - "hmac", "http", "humantime", "humantime-serde", @@ -4459,6 +4403,7 @@ dependencies = [ "kio", "libc", "moq-congestion", + "moq-crypto", "moq-net", "moq-quic", "moq-sock", @@ -4467,13 +4412,14 @@ dependencies = [ "rand", "rcgen", "reqwest", + "rind 0.1.12", + "rind-rcgen", "rustc-hash 2.1.3", "rustls", "rustls-platform-verifier", "rustls-webpki", "serde", "serde_with", - "sha2", "thiserror 2.0.21", "time", "tokio", @@ -4499,9 +4445,9 @@ dependencies = [ [[package]] name = "moq-vaapi" -version = "0.1.0" +version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a697143830ed57c5ce0e96016e845e0b097f607c536847c12890e3bd23ff4db9" +checksum = "fb08b9dab1b19b21fe0e909755a9c17601d6dcb0ced2b325fda0f0a8ba594cb8" dependencies = [ "anyhow", "bindgen 0.70.1", @@ -4573,9 +4519,9 @@ dependencies = [ [[package]] name = "mp4-atom" -version = "0.17.0" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "718b5227fa90c52ed2e72d4d553560130d9e128acb6217d9fe0ea57312ee7063" +checksum = "327faeb7d11fc3735194fdaeaa82af72167801bf39dd3d6b2684c8fc5cd7fc19" dependencies = [ "bytes", "derive_more", @@ -6043,9 +5989,11 @@ dependencies = [ "futures", "kio", "libc", + "moq-crypto", "moq-net", "moq-quic", "rustls", + "rustls-native-certs", "thiserror 2.0.21", "tokio", "tokio-rustls", @@ -6075,63 +6023,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "quinn" -version = "0.11.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" -dependencies = [ - "bytes", - "cfg_aliases", - "pin-project-lite", - "quinn-proto", - "quinn-udp", - "rustc-hash 2.1.3", - "rustls", - "socket2", - "thiserror 2.0.21", - "tokio", - "tracing", - "web-time", -] - -[[package]] -name = "quinn-proto" -version = "0.11.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9746dbde176634f4f2f1faf2404e30a31b2bc1e9cafb5329c95d8177a18c9fc" -dependencies = [ - "aws-lc-rs", - "bytes", - "getrandom 0.4.3", - "lru-slab", - "rand", - "rand_pcg", - "ring", - "rustc-hash 2.1.3", - "rustls", - "rustls-pki-types", - "slab", - "thiserror 2.0.21", - "tinyvec", - "tracing", - "web-time", -] - -[[package]] -name = "quinn-udp" -version = "0.5.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" -dependencies = [ - "cfg_aliases", - "libc", - "once_cell", - "socket2", - "tracing", - "windows-sys 0.61.2", -] - [[package]] name = "quote" version = "1.0.47" @@ -6269,10 +6160,8 @@ version = "0.14.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8" dependencies = [ - "aws-lc-rs", "rustls-pki-types", "time", - "x509-parser", "yasna", ] @@ -6389,7 +6278,6 @@ dependencies = [ "log", "percent-encoding", "pin-project-lite", - "quinn", "rustls", "rustls-pki-types", "rustls-platform-verifier", @@ -6451,6 +6339,24 @@ dependencies = [ "libcrux-traits", ] +[[package]] +name = "rind" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa730caf5eb8b28fe22f68cdd2423365696e2bbb4a95d06f332c8063e7a0e941" +dependencies = [ + "cfg-if", + "curve25519-dalek", + "fiat-crypto", + "getrandom 0.4.3", + "libc", + "libcrux-aes", + "libcrux-hacl-rs", + "libcrux-macros", + "libcrux-poly1305", + "libcrux-traits", +] + [[package]] name = "rind-jsonwebtoken" version = "0.1.12" @@ -6459,7 +6365,19 @@ checksum = "6bf395502633219077e6e360d458ac1c040bf70212d6119a7dabcd123cacf745" dependencies = [ "jsonwebtoken", "pkcs1", - "rind", + "rind 0.1.12", +] + +[[package]] +name = "rind-rcgen" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e812d797a202e1f7dcfa34f7f4c25a49bc0d1f22525804f4f057cbaf41cbb05a" +dependencies = [ + "rcgen", + "rind 0.2.0", + "rustls-pki-types", + "zeroize", ] [[package]] @@ -6468,7 +6386,7 @@ version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e67845e30d1c9a194485cdcb585abf7dcc011a4413488d57634ec39d7e832f0" dependencies = [ - "rind", + "rind 0.1.12", "rustls", ] @@ -6482,7 +6400,7 @@ dependencies = [ "cfg-if", "getrandom 0.2.17", "libc", - "untrusted 0.9.0", + "untrusted", "windows-sys 0.52.0", ] @@ -6588,7 +6506,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6610,7 +6528,6 @@ version = "0.23.46" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" dependencies = [ - "aws-lc-rs", "log", "once_cell", "ring", @@ -6638,7 +6555,6 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ - "web-time", "zeroize", ] @@ -6675,10 +6591,9 @@ version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ - "aws-lc-rs", "ring", "rustls-pki-types", - "untrusted 0.9.0", + "untrusted", ] [[package]] @@ -7031,17 +6946,6 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" -[[package]] -name = "sha2" -version = "0.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47d7069beb7d6ac7b9acd1039986e73443f24234f41074da099d6f994ac9ad19" -dependencies = [ - "cfg-if", - "cpufeatures", - "digest", -] - [[package]] name = "sharded-slab" version = "0.1.7" @@ -8412,12 +8316,6 @@ dependencies = [ "num-traits", ] -[[package]] -name = "untrusted" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" - [[package]] name = "untrusted" version = "0.9.0" @@ -9624,7 +9522,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" dependencies = [ "asn1-rs", - "aws-lc-rs", "data-encoding", "der-parser", "lazy_static", diff --git a/bun.lock b/bun.lock index 0e3bbc08..f555971f 100644 --- a/bun.lock +++ b/bun.lock @@ -56,11 +56,11 @@ "name": "@moq/json", "version": "0.4.2", "dependencies": { + "@moq/clock": "workspace:^", "@moq/net": "workspace:^", "@moq/signals": "workspace:^", }, "devDependencies": { - "@moq/clock": "workspace:^", "@types/bun": "^1.4.4", "fast-check": "^4.10.2", "typescript": "7.0.2", @@ -117,7 +117,7 @@ }, "devDependencies": { "@types/bun": "^1.4.4", - "@types/node": "^26.6.4", + "@types/node": "^26.6.6", "@typescript/lib-dom": "npm:@types/web@^0.0.358", "fast-check": "^4.10.2", "typescript": "7.0.2", @@ -145,7 +145,7 @@ "@moq/signals": "workspace:^", "@moq/ui": "workspace:^", "@moq/video": "workspace:^", - "mediabunny": "^1.62.1", + "mediabunny": "^1.63.0", }, "devDependencies": { "@types/audioworklet": "^0.0.100", diff --git a/js/join/src/renew.test.ts b/js/join/src/renew.test.ts index b13be845..4dba3255 100644 --- a/js/join/src/renew.test.ts +++ b/js/join/src/renew.test.ts @@ -25,12 +25,12 @@ test("each token the studio publishes reaches the page, and only for its own nam ["bob", 0], ["alice", 1], ] as const) { - const other = origin.createBroadcast(path(name, generation)); + const other = origin.route(path(name, generation), { hidden: true }); other.broadcast.createTrack("token").writeFrame({ payload: encode(`${name} ${generation}`) }); other.announce(); } - const publication = origin.createBroadcast(path("alice", 0)); + const publication = origin.route(path("alice", 0), { hidden: true }); const track = publication.broadcast.createTrack("token"); track.writeFrame({ payload: encode("first") }); publication.announce(); @@ -50,7 +50,7 @@ test("a group superseded before it completes does not end the renewals", async ( tokens.update((all) => [...all, token]), ); - const publication = origin.createBroadcast(path("alice", 0)); + const publication = origin.route(path("alice", 0), { hidden: true }); const track = publication.broadcast.createTrack("token"); track.writeFrame({ payload: encode("first") }); publication.announce(); diff --git a/moq b/moq index ffcb7cd9..d0181980 160000 --- a/moq +++ b/moq @@ -1 +1 @@ -Subproject commit ffcb7cd9748422daa30e935ec464108b5a8ebb71 +Subproject commit d01819803e3e93e52d20366febfe9eaca9c8e232 diff --git a/quest/m0/README.md b/quest/m0/README.md index 4aac5003..2dd66f39 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -109,7 +109,6 @@ repository. - [moq-audio's filter chain is pinned](/quest/m0/pin-audio-filters.md) - the chain audio filters configure - [moq-video's browser backends are pinned](/quest/m0/pin-browser-video.md) - the web studio's video slice and browser composition - [moq-audio's browser backends are pinned](/quest/m0/pin-browser-audio.md) - the web studio's audio slice -- [moq-video's program readback is pinned](/quest/m0/pin-readback.md) - the readback the Linux virtual camera uses - [moq-audio's source tap is pinned](/quest/m0/pin-audio-tap.md) - the samples clap-to-measure sync times a clap from ## Related diff --git a/quest/m0/pin-readback.md b/quest/m0/pin-readback.md deleted file mode 100644 index dc8d4283..00000000 --- a/quest/m0/pin-readback.md +++ /dev/null @@ -1,23 +0,0 @@ -# [XS] moq-video's program readback is pinned - -## Goal - -The `moq` submodule includes moq-video's public readback, which copies a -wgpu texture into packed RGBA with no encoder, from -[Read a texture back without an encoder](https://github.com/moq-dev/moq/blob/main/quest/m1/video-readback.md). - -## Plan - -Once the wait clears, advance it with `just upgrade-moq` in its own PR, -fixing whatever the new API breaks here. - -Decided 2026-11-02 (maintainer, on -[#9](https://github.com/moq-dev/studio/pull/9)): the Linux virtual camera's -readback lives in moq-video, not in a copy here. - -## Waiting - -- moq-dev/moq: the upstream quest above lands. Check: that quest's PR has - merged on moq's `main`, deleting its file, and `moq/rs/moq-video` there has - a public readback with no encoder probe. A quest file that moved, or went - away without its work landing, does not count. diff --git a/quest/m0/vcam/linux.md b/quest/m0/vcam/linux.md index 8a4b008d..73151e61 100644 --- a/quest/m0/vcam/linux.md +++ b/quest/m0/vcam/linux.md @@ -7,7 +7,3 @@ PipeWire-aware apps (browsers and conferencing apps) list as a camera. It falls back to an existing v4l2loopback device when one is present. The studio's Devices panel leaves out its own virtual camera, so picking it cannot feed the program back into itself. - -## Required - -- [moq-video's program readback is pinned](/quest/m0/pin-readback.md) - the readback the camera reads the program through diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index ac9fd00b..088bfc63 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -978,15 +978,15 @@ impl Driver { /// Follow the broadcast at `path` for as long as the feed lives, playing /// each instance of it that comes online. async fn broadcast(&self, path: &str) -> Result<(), Error> { - let mut instances = moq_mux::follow::Instances::new(&self.context.origin, path, None)?; + let mut instances = self.context.origin.follow(path, None)?; set(&self.status, Status::Offline); let mut clock = self.clock.clone().expect("a broadcast source has a clock"); let mut playing: Option> = None; loop { tokio::select! { - event = instances.next() => match event { - Ok(moq_mux::follow::Event::Start(broadcast)) => { + event = instances.available() => match event { + Ok(Some(broadcast)) => { set(&self.status, Status::Live); *self.played.lock().unwrap() = Played { broadcast: Some(broadcast.clone()), @@ -996,7 +996,7 @@ impl Driver { clock = clock.reset(VIDEO_DELAY); playing = Some(Box::pin(self.play(broadcast, path, clock.clone()))); } - Ok(moq_mux::follow::Event::End) => { + Ok(None) => { playing = None; *self.played.lock().unwrap() = Played::default(); set(&self.status, Status::Offline); @@ -2305,7 +2305,9 @@ mod tests { local(async { let dir = tempfile::tempdir().unwrap(); let (context, origin) = context(dir.path()); - let publication = origin.create_broadcast("room/guest").unwrap(); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); let mut broadcast = publication.broadcast().clone(); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); // A known bitrate ranks a rendition above one without. @@ -2324,7 +2326,7 @@ mod tests { .unwrap() }; let first = audio("first", None); - publication.announce(moq_net::origin::Route::default()).unwrap(); + publication.announce().unwrap(); let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); @@ -2349,7 +2351,9 @@ mod tests { local(async { let dir = tempfile::tempdir().unwrap(); let (context, origin) = context(dir.path()); - let publication = origin.create_broadcast("room/guest").unwrap(); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); let mut broadcast = publication.broadcast().clone(); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); let info = catalog.track_info(hang::catalog::PRIORITY.audio); @@ -2365,7 +2369,7 @@ mod tests { .unwrap(); }; point(&mut catalog, "first"); - publication.announce(moq_net::origin::Route::default()).unwrap(); + publication.announce().unwrap(); let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); @@ -2390,7 +2394,9 @@ mod tests { local(async { let dir = tempfile::tempdir().unwrap(); let (context, origin) = context(dir.path()); - let publication = origin.create_broadcast("room/guest").unwrap(); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); let mut broadcast = publication.broadcast().clone(); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); let mut config = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2); @@ -2400,7 +2406,7 @@ mod tests { catalog.audio.renditions.insert("audio".into(), config); }) .unwrap(); - publication.announce(moq_net::origin::Route::default()).unwrap(); + publication.announce().unwrap(); let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); @@ -2421,7 +2427,7 @@ mod tests { /// A live broadcast at `room/guest`, writing nothing until the test does. struct Live { - _publication: moq_net::origin::Publication, + _publication: moq_net::origin::Route, broadcast: moq_net::broadcast::Producer, catalog: moq_mux::catalog::Producer, video: moq_video::encode::Producer, @@ -2444,7 +2450,9 @@ mod tests { /// A broadcast with a video track, and an audio one with `audio`. async fn live(origin: &moq_net::origin::Producer, audio: bool) -> Live { - let publication = origin.create_broadcast("room/guest").unwrap(); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); let mut broadcast = publication.broadcast().clone(); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); let (video, encoder) = rendition(&broadcast, catalog.timebase()).await; @@ -2454,7 +2462,7 @@ mod tests { options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Pcm, &input); moq_audio::encode::Producer::new(&mut broadcast, catalog.clone(), input, &options).unwrap() }); - publication.announce(moq_net::origin::Route::default()).unwrap(); + publication.announce().unwrap(); Live { _publication: publication, broadcast, diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index d3e2c860..1bca7df2 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -60,14 +60,14 @@ kio = { workspace = true, features = ["async", "runtime"] } moq-audio = { workspace = true, features = ["bus"] } # Signs guest tokens with the room key. moq-auth = { workspace = true } -moq-json = { workspace = true } +moq-json = { workspace = true, features = ["async"] } moq-mux = { workspace = true } moq-net = { workspace = true } # The guest room: its path convention, roster, and token claims. -moq-room = { workspace = true } +moq-room = { workspace = true, features = ["async"] } moq-scene = { workspace = true } moq-stats = { workspace = true } -moq-tokio = { workspace = true, default-features = false, features = ["aws-lc-rs", "net"] } +moq-tokio = { workspace = true, default-features = false, features = ["net"] } moq-video = { workspace = true } pollster = { workspace = true } # The join link's QR code. No dependencies of its own. diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 79ab35da..51017e43 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -573,7 +573,7 @@ pub struct Relay { pub struct Program { /// Where the broadcast is announced, which the relay connections publish. _origin: moq_net::origin::Producer, - publication: moq_net::origin::Publication, + publication: moq_net::origin::Route, /// What the broadcast lists and the clock its timestamps are on, for a /// replay of the program or a source. catalog: moq_scene::publish::Catalog, @@ -581,7 +581,7 @@ pub struct Program { /// Where the local sources keep their own tracks, on the same clock. sources: moq_scene::publish::Publish, /// What a recording reads the program from, on the same clock. - recording: moq_net::origin::Publication, + recording: moq_net::origin::Route, /// The recording broadcast's catalog, held for as long as the engine runs, /// so a recording still reading as the studio stops finds it open. _recorded: moq_scene::publish::Catalog, @@ -595,7 +595,7 @@ pub struct Program { impl Drop for Program { fn drop(&mut self) { - self.publication.close(); + self.publication.broadcast().close(); } } @@ -1432,7 +1432,9 @@ async fn publish( canvas: Canvas, live: Option<&Live>, ) -> anyhow::Result<(Program, Publish)> { - let publication = origin.create_broadcast(path).context("cannot create the broadcast")?; + let publication = origin + .route(path, moq_net::origin::RouteOptions::hidden()) + .context("cannot create the broadcast")?; let mut broadcast = publication.broadcast().clone(); // Each track reserves its ceiling on the first relay's estimate, which // that connection then probes to fund: the bandwidth the stream is granted. @@ -1489,10 +1491,10 @@ async fn publish( // published program. let local = moq_tokio::origin::spawn(); let recording = local - .create_broadcast("recording") + .route("recording", moq_net::origin::RouteOptions::hidden()) .context("cannot create the recording broadcast")?; recording - .announce(Default::default()) + .announce() .context("cannot announce the recording broadcast")?; let mut recorder = recording.broadcast().clone(); let config = moq_mux::catalog::Config::default() @@ -1552,10 +1554,8 @@ async fn publish( // One instance on every relay, so a viewer moving between them resumes // mid-group rather than restarting. - let instance = moq_net::Instance::random(); - publication - .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) - .context("cannot announce the broadcast")?; + let instance = publication.instance(); + publication.announce().context("cannot announce the broadcast")?; let health = live.zip(viewers).map(|(live, viewers)| { let feedback = health::Feedback::default(); @@ -2044,7 +2044,7 @@ type Lines = BTreeMap>; /// A further canvas's broadcast, whose catalog finishes and which closes /// when dropped. struct Broadcast { - publication: moq_net::origin::Publication, + publication: moq_net::origin::Route, /// Lists the canvas's video and the main broadcast's audio. catalog: moq_scene::publish::Catalog, } @@ -2053,7 +2053,7 @@ impl Drop for Broadcast { fn drop(&mut self) { // Fails only once the catalog failed, which ended its track already. let _ = self.catalog.finish(); - self.publication.close(); + self.publication.broadcast().close(); } } @@ -2142,7 +2142,7 @@ impl Extras { let publication = self .origin - .create_broadcast(&path) + .route(&path, moq_net::origin::RouteOptions::hidden()) .with_context(|| format!("cannot create the broadcast of canvas `{name}`"))?; let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default() @@ -2159,7 +2159,7 @@ impl Extras { } })?; publication - .announce(moq_net::origin::Route::default().with_instance(moq_net::Instance::random(), 0)) + .announce() .with_context(|| format!("cannot announce the broadcast of canvas `{name}`"))?; let renditions = Renditions { @@ -2666,6 +2666,7 @@ fn clocked(name: String, task: impl AsyncFnOnce() + Send + 'static) -> anyhow::R #[cfg(test)] mod tests { use moq_mux::catalog::Stream; + use moq_net::origin::RouteEvent; use super::*; @@ -3052,7 +3053,7 @@ mod tests { let mut following = relay.follow("studio/tall", None).unwrap(); let (broadcast, mut catalogs) = kio::clock::timeout(TIMEOUT, async { loop { - let broadcast = following.available().await.expect("broadcast"); + let broadcast = following.available().await.expect("broadcast").expect("an instance"); let format = moq_mux::catalog::CatalogFormat::default(); if let Ok(catalogs) = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await { return (broadcast, catalogs); @@ -3202,7 +3203,14 @@ mod tests { let format = moq_mux::catalog::CatalogFormat::Hang; let feed = Room::feed_path("alice", 0).unwrap(); - let broadcast = origin.consume().follow(feed, None).unwrap().available().await.unwrap(); + let broadcast = origin + .consume() + .follow(feed, None) + .unwrap() + .available() + .await + .unwrap() + .expect("an instance"); let mut listings = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await.unwrap(); let listed = kio::clock::timeout(TIMEOUT, async { loop { @@ -3264,7 +3272,8 @@ mod tests { let broadcast = kio::clock::timeout(TIMEOUT, relay.follow("studio", None).unwrap().available()) .await .expect("the relay routes the program") - .expect("broadcast"); + .expect("broadcast") + .expect("an instance"); let format = moq_mux::catalog::CatalogFormat::default(); let mut catalogs = moq_mux::catalog::Consumer::<()>::new(&broadcast, format) .await @@ -3537,32 +3546,31 @@ mod tests { let origin = moq_tokio::origin::spawn(); let subscribe = moq_tokio::origin::spawn(); let (_connections, links) = connect(relays, &origin, &subscribe); - let publication = origin.create_broadcast("studio").unwrap(); - let instance = moq_net::Instance::random(); - publication - .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) - .unwrap(); + let publication = origin.route("studio", Default::default()).unwrap(); + let instance = publication.instance(); // Publish `path` under `instance` through one relay's guest. let publish = |path: &str, instance: &moq_net::Instance, (client, url): &(moq_tokio::Client, url::Url)| { let origin = moq_tokio::origin::spawn(); - let source = origin.create_broadcast(path).unwrap(); - source - .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) + let source = origin + .route( + path, + moq_net::origin::RouteOptions::default().with_instance(instance.clone(), 0), + ) .unwrap(); let connection = client.clone().with_publisher(&origin).connect(url.clone()); (origin, source, connection) }; let _guest = publish("guest", &moq_net::Instance::random(), &guests[1]); - let mut sources = subscribe.consume().announced(); + let mut sources = subscribe.consume().routes(); let routed = async { for relay in &routes { let mut announced = relay.announced(); loop { match announced.next().await.expect("announcements") { - moq_net::announce::Event::Start(announce) if announce.prefix.as_str() == "studio" => { - assert_eq!(announce.route.instance, Some(instance.clone())); + moq_net::announce::Event::Present(announce) if announce.prefix.as_str() == "studio" => { + assert_eq!(announce.route.instance(), &instance); break; } _ => continue, @@ -3594,7 +3602,7 @@ mod tests { let first = publish("show", &older, &guests[0]); routed_at(&mut sources, &older).await; let _restarted = publish("show", &newer, &guests[1]); - routed_at(&mut routes[1].announced(), &newer).await; + routed_at(&mut routes[1].routes(), &newer).await; // Announced after the restart on the failover's one announce // stream, so once it reaches the studio the restart has too. let _marker = publish("marker", &moq_net::Instance::random(), &guests[1]); @@ -3606,13 +3614,13 @@ mod tests { .await .expect("the marker"); while let Some(event) = sources.try_next() { - if let moq_net::announce::Event::Start(announce) - | moq_net::announce::Event::Update(announce) - | moq_net::announce::Event::Restart(announce) - | moq_net::announce::Event::Takeover { announce, .. } = event + if let RouteEvent::Start(announce) + | RouteEvent::Update(announce) + | RouteEvent::Restart(announce) + | RouteEvent::Takeover { announce, .. } = event { assert!( - announce.prefix.as_str() != "show" || announce.route.instance.as_ref() != Some(&newer), + announce.prefix.as_str() != "show" || announce.route.instance() != &newer, "the failover's instance won while the first relay has one" ); } @@ -3663,10 +3671,9 @@ mod tests { connections, clients: vec![client], }; - let publication = origin.create_broadcast("studio").unwrap(); - publication.announce(Default::default()).unwrap(); + let publication = origin.route("studio", Default::default()).unwrap(); let info = || moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI); - let tracks = ["video", "audio"].map(|name| publication.create_track(name, info()).unwrap()); + let tracks = ["video", "audio"].map(|name| publication.broadcast().create_track(name, info()).unwrap()); // Wait for an event `want` picks on `announced`. let routed = async |announced: &mut moq_net::announce::Consumer, want: fn(&Event) -> bool| { @@ -3678,14 +3685,15 @@ mod tests { let mut announced = relay.announced(); routed( &mut announced, - |event| matches!(event, Event::Start(announce) if announce.prefix.as_str() == "studio"), + |event| matches!(event, Event::Present(announce) if announce.prefix.as_str() == "studio"), ) .await .expect("the relay routes the broadcast"); let broadcast = kio::clock::timeout(TIMEOUT, relay.follow("studio", None).unwrap().available()) .await .expect("the relay routes the broadcast") - .expect("broadcast"); + .expect("broadcast") + .expect("an instance"); let mut subscriptions = Vec::new(); for (name, track) in ["video", "audio"].into_iter().zip(&tracks) { let mut subscription = broadcast.track(name).unwrap().subscribe(None).await.unwrap(); @@ -3742,22 +3750,21 @@ mod tests { } routed( &mut announced, - |event| matches!(event, Event::End(announce) if announce.prefix.as_str() == "studio"), + |event| matches!(event, Event::Absent(announce) if announce.prefix.as_str() == "studio"), ) .await .expect("the broadcast is withdrawn"); } - /// Wait until `announced` routes `show` under `instance`. - async fn routed_at(announced: &mut moq_net::announce::Consumer, instance: &moq_net::Instance) { - use moq_net::announce::Event; + /// Wait until `routes` routes `show` under `instance`. + async fn routed_at(routes: &mut moq_net::origin::Routes, instance: &moq_net::Instance) { loop { - let announce = match announced.next().await.expect("announcements") { - Event::Start(announce) | Event::Update(announce) | Event::Restart(announce) => announce, - Event::Takeover { announce, .. } => announce, - Event::End(_) => continue, + let announce = match routes.next().await.expect("routes") { + RouteEvent::Start(announce) | RouteEvent::Update(announce) | RouteEvent::Restart(announce) => announce, + RouteEvent::Takeover { announce, .. } => announce, + RouteEvent::End(_) => continue, }; - if announce.prefix.as_str() == "show" && announce.route.instance.as_ref() == Some(instance) { + if announce.prefix.as_str() == "show" && announce.route.instance() == instance { return; } } diff --git a/rs/moq-studio/src/feed.rs b/rs/moq-studio/src/feed.rs index c1fb8a45..fc148ac0 100644 --- a/rs/moq-studio/src/feed.rs +++ b/rs/moq-studio/src/feed.rs @@ -64,15 +64,15 @@ impl Feeds { clock: moq_mux::Clock, epoch: Duration, ) -> anyhow::Result { - let requests = origin - .requests(PROGRAM) + let route = origin + .route_prefix(PROGRAM, Default::default()) .context("cannot serve the program in the room")?; let served = program.clone(); let serving = tokio::spawn(async move { - while let Ok(request) = requests.requested_broadcast().await { + while let Some(request) = route.requested().await { // The route claims everything beneath it too, which holds nothing. match request.path().as_str() == PROGRAM { - true => request.serve(&served), + true => request.serve(served.clone()), false => request.reject(moq_net::Error::NotFound), } } @@ -136,7 +136,7 @@ impl Feeds { let program = Path::new(PROGRAM)? .relative(&path) .context("the program cannot be named from a feed")?; - let publication = self.origin.create_broadcast(path)?; + let publication = self.origin.route(path, moq_net::origin::RouteOptions::hidden())?; let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default().with_clock(self.clock.clone()); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; @@ -149,7 +149,7 @@ impl Feeds { options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Opus, &input); options.config.bitrate = Some(BITRATE); let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog.clone(), input, &options)?; - publication.announce(moq_net::origin::Route::default())?; + publication.announce()?; let pick = mix::Pick::default(); let (stop, stopped) = tokio::sync::oneshot::channel(); @@ -199,7 +199,7 @@ struct Feed { /// A feed's broadcast, as its task runs it. struct Running { /// Ends the broadcast once the task returns. - _publication: moq_net::origin::Publication, + _publication: moq_net::origin::Route, catalog: moq_mux::catalog::Producer, audio: moq_audio::encode::Producer, follow: mix::Follow, @@ -429,7 +429,14 @@ mod tests { std::thread::scope(|scope| scope.spawn(|| feeds.sync(&collection, &sources)).join().unwrap()); let path = Room::feed_path("alice", 0).unwrap(); - let broadcast = origin.consume().follow(path, None).unwrap().available().await.unwrap(); + let broadcast = origin + .consume() + .follow(path, None) + .unwrap() + .available() + .await + .unwrap() + .expect("an instance"); let catalog = listed(&broadcast).await; let video = &catalog.video.renditions["video"]; assert_eq!(video.broadcast.as_ref().unwrap().as_str(), "../../.program"); @@ -450,7 +457,14 @@ mod tests { assert_eq!(bob_hears.data[0], 0.25, "bob hears only alice"); let path = Room::feed_path("bob", 0).unwrap(); - let bob = origin.consume().follow(path, None).unwrap().available().await.unwrap(); + let bob = origin + .consume() + .follow(path, None) + .unwrap() + .available() + .await + .unwrap() + .expect("an instance"); room.stop_renewing("bob").unwrap(); feeds.sync(&collection, &sources); assert_eq!(feeds.feeds.keys().collect::>(), ["alice"]); @@ -473,7 +487,10 @@ mod tests { let read = moq_tokio::origin::spawn(); let link = room.invite("alice").unwrap().link; let _alice = relay.join(&link, &moq_tokio::origin::spawn(), &read); - let follow = async |path: Path| read.consume().follow(path, None)?.available().await; + let follow = async |path: Path| { + let broadcast = read.consume().follow(path, None)?.available().await?; + moq_net::Result::Ok(broadcast.expect("the first change is an instance")) + }; let path = Room::feed_path("alice", 0).unwrap(); let feed = kio::clock::timeout(TIMEOUT, follow(path.clone())) diff --git a/rs/moq-studio/src/guests.rs b/rs/moq-studio/src/guests.rs index 197c3af6..ea507d7e 100644 --- a/rs/moq-studio/src/guests.rs +++ b/rs/moq-studio/src/guests.rs @@ -752,14 +752,14 @@ struct Renewals { /// One guest's token track, ended when dropped. struct Renewed { generation: u64, - publication: moq_net::origin::Publication, + publication: moq_net::origin::Route, track: moq_net::track::Producer, } impl Drop for Renewed { fn drop(&mut self) { let _ = self.track.finish(); - self.publication.close(); + self.publication.broadcast().close(); } } @@ -793,17 +793,18 @@ impl Renewals { /// The token track for `identity` in `generation`, holding a fresh token. fn publish(&self, identity: &str, generation: u64) -> anyhow::Result { - let publication = self - .origin - .create_broadcast(Room::renewal_path(identity, generation)?)?; - let track = publication.create_track(TOKEN_TRACK, None)?; + let publication = self.origin.route( + Room::renewal_path(identity, generation)?, + moq_net::origin::RouteOptions::hidden(), + )?; + let track = publication.broadcast().create_track(TOKEN_TRACK, None)?; let mut renewed = Renewed { generation, publication, track, }; self.write(identity, &mut renewed)?; - renewed.publication.announce(Default::default())?; + renewed.publication.announce()?; Ok(renewed) } @@ -1255,8 +1256,7 @@ pub(crate) mod tests { ); let guest = moq_tokio::origin::spawn(); - let camera = guest.create_broadcast("alice/camera.hang").unwrap(); - camera.announce(moq_net::origin::Route::default()).unwrap(); + let _camera = guest.route("alice/camera.hang", Default::default()).unwrap(); let renewals = moq_tokio::origin::spawn(); let alice = relay.join(&room.invite("alice").unwrap().link, &guest, &renewals); @@ -1302,7 +1302,8 @@ pub(crate) mod tests { let broadcast = origin .follow(Room::renewal_path(identity, generation)?, None)? .available() - .await?; + .await? + .expect("the first change is an instance"); read_token(broadcast).await } diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 9fd11ca3..c929b171 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -859,51 +859,35 @@ impl Feedback { let mut readers = BTreeMap::::new(); let mut unread = 0usize; while let Some(event) = announced.next().await { - let (announce, restart) = match event { - moq_net::announce::Event::Start(announce) => (announce, false), - moq_net::announce::Event::Restart(announce) => (announce, true), - moq_net::announce::Event::End(announce) => { + match event { + moq_net::announce::Event::Present(announce) => { + let path = announce.prefix; + let Some(name) = name(&prefix, &path).filter(|_| hang::echo::is_echo(&path)) else { + continue; + }; + if readers.len() < MAX_VIEWERS { + let reader = self.read(origin.clone(), path.clone(), name); + readers.insert(path, reader); + } else { + unread += 1; + } + } + moq_net::announce::Event::Absent(announce) => { if readers.remove(&announce.prefix).is_none() && hang::echo::is_echo(&announce.prefix) { unread = unread.saturating_sub(1); } - self.0.lock().unwrap().unread = unread; - continue; } - // A takeover continues the same instance, and an update moves its route. - _ => continue, - }; - let path = announce.prefix; - let Some(name) = name(&prefix, &path).filter(|_| hang::echo::is_echo(&path)) else { - continue; - }; - // A new instance of a viewer being read replaces its reader; one not - // read was counted when it started. - let read = if restart { - readers.remove(&path).is_some() - } else if readers.len() < MAX_VIEWERS { - true - } else { - unread += 1; - false - }; - if read { - let reader = self.read( - origin.clone(), - Observer::new(path.clone(), announce.route.instance), - name, - ); - readers.insert(path, reader); } self.0.lock().unwrap().unread = unread; } Ok(()) } - /// Start reading the viewer `observer` names. - fn read(&self, origin: moq_net::origin::Consumer, observer: Observer, name: String) -> Reader { + /// Start reading the viewer at `path`, named `name`. + fn read(&self, origin: moq_net::origin::Consumer, path: moq_net::Path, name: String) -> Reader { let slot = Slot::default(); self.0.lock().unwrap().slots.insert(name.clone(), slot.clone()); - let task = tokio::spawn(read(origin, observer, slot.clone())); + let task = tokio::spawn(read(origin, path, slot.clone())); Reader { name, slot, @@ -949,29 +933,67 @@ fn name(prefix: &moq_net::Path, path: &moq_net::Path) -> Option { name.as_str().strip_suffix(hang::echo::SUFFIX).map(str::to_string) } -/// Keep the latest reports of the viewer `observer` names in `slot`, until -/// its broadcast ends or the report cannot be read. -async fn read(origin: moq_net::origin::Consumer, observer: Observer, slot: Slot) { - let read = async { - let broadcast = origin.request(&observer.path, observer.instance.clone()).await?; - let track = broadcast.track(hang::echo::TRACK)?.subscribe(None).await?; - let mut config = moq_json::snapshot::consumer::Config::default(); - config.max_size = Some(MAX_REPORT); - let mut reports = moq_json::snapshot::Consumer::::new(track, config); - while let Some(report) = reports.latest().await? { - let report = health::Report::new(observer.clone(), report.value); - let mut reports = slot.lock().unwrap(); - reports.previous = reports.current.take(); - reports.current = Some((report, kio::clock::now())); - } - anyhow::Ok(()) +/// Keep the latest reports of the viewer at `path` in `slot`, from each +/// instance that serves it in turn, until dropped. A new instance starts its +/// reports afresh. +async fn read(origin: moq_net::origin::Consumer, path: moq_net::Path, slot: Slot) { + let mut following = match origin.follow(&path, None) { + Ok(following) => following, + Err(err) => return failed(&slot, &path, err.into()), }; - if let Err(err) = read.await { - tracing::debug!(%err, viewer = %observer.path, "stopped reading feedback"); - slot.lock().unwrap().error = Some(format!("{err:#}")); + let mut next = following.available().await; + loop { + let broadcast = match next { + Ok(Some(broadcast)) => broadcast, + // Offline until another instance serves the path. + Ok(None) => { + next = following.available().await; + continue; + } + // A closed origin answers every later call at once. + Err(err @ moq_net::Error::Closed) => return failed(&slot, &path, err.into()), + // The follow waits for the route to change before it asks again. + Err(err) => { + failed(&slot, &path, err.into()); + next = following.available().await; + continue; + } + }; + *slot.lock().unwrap() = Reports::default(); + let observer = Observer::new(path.clone(), broadcast.info().instance.clone()); + tokio::select! { + result = reports(&broadcast, observer, &slot) => { + if let Err(err) = result { + failed(&slot, &path, err); + } + next = following.available().await; + } + changed = following.available() => next = changed, + } } } +/// Keep the latest reports `broadcast` publishes in `slot`, until its track ends. +async fn reports(broadcast: &moq_net::broadcast::Consumer, observer: Observer, slot: &Slot) -> anyhow::Result<()> { + let track = broadcast.track(hang::echo::TRACK)?; + let mut config = moq_json::snapshot::consumer::Config::default(); + config.max_size = Some(MAX_REPORT); + let mut reports = moq_json::snapshot::Consumer::::new(track, config); + while let Some(report) = reports.latest().await? { + let report = health::Report::new(observer.clone(), report.value); + let mut reports = slot.lock().unwrap(); + reports.previous = reports.current.take(); + reports.current = Some((report, kio::clock::now())); + } + Ok(()) +} + +/// Record why reading the viewer at `path` stopped. +fn failed(slot: &Slot, path: &moq_net::Path, err: anyhow::Error) { + tracing::debug!(%err, viewer = %path, "stopped reading feedback"); + slot.lock().unwrap().error = Some(format!("{err:#}")); +} + /// A running feedback reader, which forgets its viewer when it stops. struct Reader { name: String, @@ -1418,7 +1440,7 @@ mod tests { async fn viewer_feedback_times_each_track() { let _ = moq_tokio::crypto::install_default(); let origin = moq_tokio::origin::spawn(); - let publication = origin.create_broadcast("show").unwrap(); + let publication = origin.route("show", moq_net::origin::RouteOptions::hidden()).unwrap(); let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); @@ -1458,10 +1480,12 @@ mod tests { .unwrap(), })); - let alice = origin.create_broadcast("show/viewers/alice.echo").unwrap(); - let track = alice.create_track(hang::echo::TRACK, None).unwrap(); + let alice = origin + .route("show/viewers/alice.echo", moq_net::origin::RouteOptions::hidden()) + .unwrap(); + let track = alice.broadcast().create_track(hang::echo::TRACK, None).unwrap(); let mut reports = moq_json::snapshot::Producer::::new(track, Default::default()); - alice.announce(Default::default()).unwrap(); + alice.announce().unwrap(); let timestamp = moq_net::Timestamp::from_millis(20_000).unwrap(); let wall = clock.wall_clock(timestamp).unwrap() + Duration::from_millis(250); @@ -1494,11 +1518,76 @@ mod tests { ); assert_eq!(viewer.latency, Some(latency)); - alice.unannounce(); + alice.withdraw(); until(&mut sampler, |report| report.viewers.reports.is_empty()).await; follow.abort(); } + /// A viewer that restarts under a new instance is read from that + /// instance, its reports starting afresh. + #[tokio::test(start_paused = true)] + async fn a_restarted_viewer_is_read_afresh() { + let origin = moq_tokio::origin::spawn(); + let path = moq_net::Path::new("show/viewers/alice.echo").unwrap(); + let publish = || { + let route = origin.route(&path, moq_net::origin::RouteOptions::hidden()).unwrap(); + let track = route.broadcast().create_track(hang::echo::TRACK, None).unwrap(); + let mut reports = moq_json::snapshot::Producer::::new(track, Default::default()); + reports.update(&hang::echo::Snapshot::default()).unwrap(); + route.announce().unwrap(); + (route, reports) + }; + let slot = Slot::default(); + let reading = tokio::spawn(read(origin.consume(), path.clone(), slot.clone())); + let read_from = async |instance: moq_net::Instance| { + let wait = async { + loop { + let read = { + let reports = slot.lock().unwrap(); + let observer = reports.current.as_ref().map(|(report, _)| &report.observer); + let read = observer.is_some_and(|observer| observer.instance.as_ref() == Some(&instance)); + assert!( + !read || reports.previous.is_none(), + "a new instance keeps no older report" + ); + read + }; + if read { + return; + } + kio::clock::sleep(INTERVAL).await; + } + }; + kio::clock::timeout(Duration::from_secs(30), wait) + .await + .expect("the instance is read"); + }; + + let (first, _first_reports) = publish(); + read_from(first.instance()).await; + // Creating the path again supersedes the first route. + let (second, _second_reports) = publish(); + assert_ne!(first.instance(), second.instance()); + read_from(second.instance()).await; + reading.abort(); + } + + /// A reader stops, saying why, once its origin closes. + #[tokio::test(start_paused = true)] + async fn a_reader_stops_with_its_origin() { + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let slot = Slot::default(); + let path = moq_net::Path::new("show/viewers/alice.echo").unwrap(); + let reading = tokio::spawn(read(origin.consume(), path, slot.clone())); + tokio::task::yield_now().await; + drop(driver); + kio::clock::timeout(Duration::from_secs(30), reading) + .await + .expect("the reader stops") + .unwrap(); + assert!(slot.lock().unwrap().error.is_some()); + } + /// A relay that shares its stats counts each session watching the /// broadcast, and only those on the broadcast itself. #[tokio::test(flavor = "multi_thread")] @@ -1523,9 +1612,11 @@ mod tests { let client = client.init(Default::default()).unwrap(); let published = moq_tokio::origin::spawn(); - let show = published.create_broadcast("show").unwrap(); - let track = show.create_track("video", None).unwrap(); - show.announce(Default::default()).unwrap(); + let show = published + .route("show", moq_net::origin::RouteOptions::hidden()) + .unwrap(); + let track = show.broadcast().create_track("video", None).unwrap(); + show.announce().unwrap(); let subscribe = moq_tokio::origin::spawn(); let watching = Watching::default(); tokio::spawn( @@ -1548,7 +1639,8 @@ mod tests { .unwrap() .available() .await - .unwrap(); + .unwrap() + .expect("an instance"); let mut video = broadcast.track("video").unwrap().subscribe(None).await.unwrap(); track.append_group().unwrap().finish().unwrap(); video.recv_group().await.unwrap(); @@ -1572,8 +1664,13 @@ mod tests { let follow = tokio::spawn(feedback.clone().follow(origin.consume(), prefix)); let announce = |n: usize| { - let viewer = origin.create_broadcast(format!("show/viewers/{n:03}.echo")).unwrap(); - viewer.announce(Default::default()).unwrap(); + let viewer = origin + .route( + format!("show/viewers/{n:03}.echo"), + moq_net::origin::RouteOptions::hidden(), + ) + .unwrap(); + viewer.announce().unwrap(); viewer }; let read = |feedback: &Feedback, name: &str| { @@ -1589,8 +1686,8 @@ mod tests { while read(&feedback, "065") != (MAX_VIEWERS, 2, false) { kio::clock::sleep(Duration::from_millis(10)).await; } - viewers[0].unannounce(); - viewers[MAX_VIEWERS].unannounce(); + viewers[0].withdraw(); + viewers[MAX_VIEWERS].withdraw(); while read(&feedback, "000") != (MAX_VIEWERS - 1, 1, false) { kio::clock::sleep(Duration::from_millis(10)).await; } diff --git a/rs/moq-studio/src/record.rs b/rs/moq-studio/src/record.rs index b5507537..9eef9a42 100644 --- a/rs/moq-studio/src/record.rs +++ b/rs/moq-studio/src/record.rs @@ -513,19 +513,19 @@ impl Run { let mut files: tokio::task::JoinSet<(u64, bool)> = tokio::task::JoinSet::new(); let mut current: Option = None; let mut count = 0; - let failed = match moq_mux::follow::Instances::new(&self.origin, path.as_str(), None) { + let failed = match self.origin.follow(path.as_str(), None) { Err(err) => Some(anyhow::Error::from(err)), Ok(mut instances) => loop { tokio::select! { - event = instances.next() => match event { - Ok(moq_mux::follow::Event::Start(broadcast)) => { + event = instances.available() => match event { + Ok(Some(broadcast)) => { current = None; if !*self.closing.read() { count += 1; current = self.instance(&name, format, broadcast, count, &mut files); } } - Ok(moq_mux::follow::Event::End) => current = None, + Ok(None) => current = None, Err(err) => break Some(err.into()), }, Some(joined) = files.join_next() => { @@ -1185,7 +1185,7 @@ mod tests { origin: moq_net::origin::Producer, /// What the relays feed, where guests publish. relay: moq_net::origin::Producer, - publication: moq_net::origin::Publication, + publication: moq_net::origin::Route, catalog: moq_scene::publish::Catalog, /// The local broadcast the sources' tracks are in, and its catalog. local: moq_net::broadcast::Producer, @@ -1202,12 +1202,12 @@ mod tests { impl Studio { fn new() -> Self { let origin = moq_tokio::origin::spawn(); - let publication = origin.create_broadcast("studio").unwrap(); + let publication = origin.route("studio", moq_net::origin::RouteOptions::hidden()).unwrap(); let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); let program = Some(Tone::new(&mut broadcast, catalog.clone())); - publication.announce(Default::default()).unwrap(); + publication.announce().unwrap(); let mut local = moq_net::broadcast::Info::new().produce(); let config = moq_mux::catalog::Config::default() .with_catalog(Default::default()) @@ -1333,7 +1333,7 @@ mod tests { /// A guest's microphone, published through the relay as the join page /// publishes it: Opus packets, as received, in a catalog of its own. struct Guest { - _publication: moq_net::origin::Publication, + _publication: moq_net::origin::Route, _catalog: moq_mux::catalog::Producer, audio: moq_mux::container::Producer, /// Each packet sent, in order. @@ -1348,7 +1348,7 @@ mod tests { impl Guest { /// Join at `path` on `relay`. fn join(relay: &moq_net::origin::Producer, path: &str) -> Self { - let publication = relay.create_broadcast(path).unwrap(); + let publication = relay.route(path, moq_net::origin::RouteOptions::hidden()).unwrap(); let mut broadcast = publication.broadcast().clone(); let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, moq_mux::catalog::Config::default()).unwrap(); @@ -1367,7 +1367,7 @@ mod tests { }) .unwrap(); let container = moq_mux::catalog::hang::Container::Legacy(moq_mux::container::Kind::Audio); - publication.announce(Default::default()).unwrap(); + publication.announce().unwrap(); Self { _publication: publication, _catalog: catalog, @@ -1886,7 +1886,7 @@ mod tests { async fn an_empty_recording_is_removed() { let dir = tempfile::tempdir().unwrap(); let origin = moq_tokio::origin::spawn(); - let publication = origin.create_broadcast("studio").unwrap(); + let publication = origin.route("studio", moq_net::origin::RouteOptions::hidden()).unwrap(); let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); let _catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); @@ -1914,11 +1914,11 @@ mod tests { async fn a_failure_before_any_media_says_why() { let dir = tempfile::tempdir().unwrap(); let origin = moq_tokio::origin::spawn(); - let publication = origin.create_broadcast("studio").unwrap(); + let publication = origin.route("studio", moq_net::origin::RouteOptions::hidden()).unwrap(); let mut broadcast = publication.broadcast().clone(); let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); - publication.announce(Default::default()).unwrap(); + publication.announce().unwrap(); let source = moq_mux::Source::new(origin.consume(), publication.broadcast().consume()); let local = moq_mux::Source::new(origin.consume(), moq_net::broadcast::Info::new().produce().consume()); let inputs = Inputs { diff --git a/rs/moq-studio/src/uplink.rs b/rs/moq-studio/src/uplink.rs index b5e42902..b96dec04 100644 --- a/rs/moq-studio/src/uplink.rs +++ b/rs/moq-studio/src/uplink.rs @@ -40,15 +40,13 @@ pub async fn measure( ) -> anyhow::Result { let publish = moq_tokio::origin::spawn(); let publication = publish - .create_broadcast(path) + .route(path, moq_net::origin::RouteOptions::hidden()) .context("cannot create the probe broadcast")?; let track = publication .broadcast() .create_track(TRACK, None) .context("cannot create the probe track")?; - publication - .announce(moq_net::origin::Route::default()) - .context("cannot announce the probe")?; + publication.announce().context("cannot announce the probe")?; // One-shot, so a relay that refuses either side ends the probe with its reason // instead of being redialed. @@ -93,13 +91,17 @@ pub async fn measure( Err(err) = fill(track, &reservation, ceiling) => Err(err), result = measured => result, }; - publication.close(); + publication.broadcast().close(); result } /// Wait for the probe through the relay, and subscribe to its track. async fn receive(origin: moq_net::origin::Consumer, path: &str) -> anyhow::Result { - let broadcast = origin.follow(path, None)?.available().await?; + let broadcast = origin + .follow(path, None)? + .available() + .await? + .expect("the first change is an instance"); let mut track = broadcast.track(TRACK)?.subscribe(None).await?; // Wait for the first group, so the window measures data that flows. track.recv_group().await?.context("the probe ended before it began")?;