diff --git a/README.md b/README.md index 37fb688..a3b91ff 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,8 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ Security. - The `--connect-*`, `--quic-*`, and `--log-level` flags are [moq-cli's](https://doc.moq.dev/bin/cli); see `moq-studio --help`. +- `moq-studio --watch ` opens no scene file and plays a studio broadcast + instead, as a viewer sees it; see [Watching](#watching). ## First launch @@ -993,6 +995,29 @@ A take changes the layout on the frame the program starts showing the new scene, so the published sources of the scene leaving go off and those of the scene coming in go on at once, even through a transition. +#### Watching + +`moq-studio --watch https://relay.example/anon/my-show` plays a studio +broadcast in a window as a viewer sees it: the program, with each picture of +its layer layout over it and each sound mixed with it. The URL's path names +the broadcast, as `moq play`'s does, and `--connect-token` and the `--quic-*` +flags apply to it; `--connect`, `--failover`, and `--room` are refused +beside it. It is a prototype for settling the layout's format, which goes +once `moq play` plays layers. + +- Each layout takes effect when the program plays the frame it is stamped + with, and its pictures' keyframes play from its `start`. +- Each source plays its own track on the program's clock: a picture shows + when the program's picture stamped alike does, and a sound plays with the + program's sound stamped alike, at its `gain`, `muted`, `pan`, and `delay`. + A frame too late for its moment is dropped. +- A broadcast without a layout, or with a layout of another `version`, plays + its program alone; the window says why for the latter. +- Guests' broadcasts are not in the layout yet, so only sources local to the + studio play as layers. `priority` and `toggle` are not applied. +- Sound needs a build with `capture`; without it the window says so in the + log and plays the pictures alone. + ### Recording **Start recording** in the Output panel saves the program at the canvas size, diff --git a/quest/m2/compose/controls.md b/quest/m2/compose/controls.md index a79b3c2..c3b2390 100644 --- a/quest/m2/compose/controls.md +++ b/quest/m2/compose/controls.md @@ -11,8 +11,8 @@ the studio. ## Plan Decided 2026-11-08 (maintainer, /quest-plan): the controls land in the native -player, the [watch prototype](/quest/m2/compose/watch.md) (rejected: browser -only). moq's players take them from hang's layer spec. +player, the `moq-studio --watch` prototype (`rs/moq-studio/src/watch.rs`; +rejected: browser only). moq's players take them from hang's layer spec. Decided 2026-11-09 (maintainer, /quest-audit): ✅ viewer toggles also land in moq's native player, @@ -28,7 +28,3 @@ layer is unsubscribed, so it costs the viewer and the relay nothing. The studio keeps encoding it, so toggling it back on can reach its recent past. Subscriptions are per track, not per layer: a source shown by several layers stays subscribed while any of them is on, at the highest priority among them. - -## Required - -- [moq-studio watch prototype](/quest/m2/compose/watch.md) - the viewer that exposes the toggles diff --git a/quest/m2/compose/hang-layers.md b/quest/m2/compose/hang-layers.md index 0df6526..1a9e38d 100644 --- a/quest/m2/compose/hang-layers.md +++ b/quest/m2/compose/hang-layers.md @@ -13,7 +13,7 @@ Decided 2026-11-08 (maintainer, /quest-plan): the layer layout becomes official moq, so `moq play` and `@moq/watch` play a studio broadcast. Once the wait clears, advance the gitlink with `just upgrade-moq` in this PR, move the studio to hang's types, and delete the -[watch prototype](/quest/m2/compose/watch.md) and its flag. +`moq-studio --watch` prototype (`rs/moq-studio/src/watch.rs`) and its flag. Decided 2026-11-09 (maintainer, /quest-audit): ✅ the wait names moq's three quests (rejected: "both quests", unnamed, which left out `@moq/watch`). @@ -21,7 +21,7 @@ quests (rejected: "both quests", unnamed, which left out `@moq/watch`). ## Waiting - moq-dev/moq: its three layer quests land, planned in moq's m3 from the - [watch prototype](/quest/m2/compose/watch.md)'s format: + `moq-studio --watch` prototype's format, the README's "Layer layout": [hang specifies a layer layout track](https://github.com/moq-dev/moq/blob/main/quest/m3/layers/spec.md), [moq play composites layers](https://github.com/moq-dev/moq/blob/main/quest/m3/layers/play.md), and @@ -34,6 +34,5 @@ quests (rejected: "both quests", unnamed, which left out `@moq/watch`). ## Required -- [moq-studio watch prototype](/quest/m2/compose/watch.md) - the format moq specifies - [Viewer controls](/quest/m2/compose/controls.md) - toggles and priorities the spec carries - [The studio is caught up to moq's main](/quest/m0/pin-main.md) - moq's breaking changes since moq-dev/moq#8272, so this pin takes only its own diff --git a/quest/m2/compose/layout.md b/quest/m2/compose/layout.md index c91e27d..9f415c9 100644 --- a/quest/m2/compose/layout.md +++ b/quest/m2/compose/layout.md @@ -19,6 +19,11 @@ studio's broadcast. Bump the layout's `version` if the change is not additive. +The `moq-studio --watch` prototype (`rs/moq-studio/src/watch.rs`) plays only +local sources' layers today: its layers play tracks of the program's own +broadcast on the program's clock, so a guest's layer needs the prototype to +follow the named broadcast, on the offset the layout carries. + ## Waiting - Maintainer: decide how the layout names a guest's broadcast. The epic @@ -43,7 +48,3 @@ Bump the layout's `version` if the change is not additive. Check: the maintainer answers in /quest-unblock, and the choice is recorded in this Plan and the epic's. - -## Related - -- [moq-studio watch prototype](/quest/m2/compose/watch.md) - the first player of the layout diff --git a/quest/m2/compose/watch.md b/quest/m2/compose/watch.md deleted file mode 100644 index 0ce3816..0000000 --- a/quest/m2/compose/watch.md +++ /dev/null @@ -1,30 +0,0 @@ -# [M] moq-studio watch prototype - -## Goal - -`moq-studio --watch ` plays a studio broadcast: the program, with each -source the layer layout names composited over it on the media clock, and -their audio mixed with the program's. A broadcast without a layer layout plays -its program alone. - -## Plan - -Decided 2026-11-08 (maintainer, /quest-plan): a throwaway prototype, so the -layer format settles without moq round trips before hang specifies it -(rejected: going straight to `moq play`). It lives in `moq-studio`, not its -own binary, since [hang-layers](/quest/m2/compose/hang-layers.md) deletes it -once `moq play` plays layers. - -Build it on moq-scene, which already plays remote broadcasts as sources for -guests (`rs/moq-studio/src/guests.rs`): a scene that the layout drives rather -than a scene file. `README.md` documents the flag while it exists. - -It plays the layer layout the studio publishes on `live.kixel.studio.layers` -(`rs/moq-scene/src/layout.rs`, and the README's "Layer layout"), refusing a -layout of another `version` loudly and playing the program alone. Its working -format is the starting point for moq's hang layer spec, which waits on this -quest. - -## Related - -- [Guests in the layer layout](/quest/m2/compose/layout.md) - guests' layers, once the maintainer decides how the layout names their broadcasts diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index 46b6c02..56784c7 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -469,6 +469,13 @@ impl Collection { "`{name}` draws scene `{scene}`, which is not a scene" ))); } + if let Source::Layer { broadcast, .. } = source + && !matches!(self.sources.get(broadcast), Some(Source::Broadcast { .. })) + { + return Err(Error::Invalid(format!( + "`{name}` plays on `{broadcast}`, which is not a broadcast source" + ))); + } let Source::Clone { source } = source else { continue }; match self.sources.get(source) { Some(Source::Clone { .. }) => { @@ -1848,6 +1855,19 @@ pub enum Source { /// same frame. A scene drawing it cannot go on program, and the top layer /// cannot draw it, since either would draw the program into itself. Program, + /// A source a studio broadcast publishes independently, played from the + /// broadcast the [`Source::Broadcast`] called `broadcast` plays, on that + /// source's playout clock, so it lands with the program as the + /// [layer layout](crate::layout) places it. Never in a scene file: only + /// a player of the layout makes one. + #[serde(skip)] + Layer { + /// The broadcast source whose broadcast and clock it plays on. + broadcast: String, + /// Its entry under the broadcast catalog's + /// [`SECTION`](crate::publish::SECTION) `sources`. + source: String, + }, } impl Source { @@ -1904,6 +1924,7 @@ impl Source { | Self::Broadcast { .. } | Self::Browser { .. } | Self::Replay + | Self::Layer { .. } ) } diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 963f935..83b0940 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -655,6 +655,12 @@ impl Compositor { Ok(()) } + /// Play the program scene's keyframes as if it went on program at + /// `start`, such as one a viewer joined after it went on. + pub fn set_program_start(&mut self, start: Instant) { + self.program_start = Some(start); + } + /// Refuse to take a preview scene to program that cannot go there. fn can_take(&self, collection: &Collection) -> Result<(), Error> { match self.preview_scene.as_deref() { diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 127f18f..eb7b616 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -30,7 +30,7 @@ pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); /// represent, and so how long a stalled group is waited on before it is /// skipped, as under `moq play --delay auto`. A shorter wait would cap the /// estimate at it, since a skipped group is never measured. -pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); +pub const MAX_DELAY: Duration = Duration::from_secs(2); /// How far ahead of its moment a sound is handed to the bus. /// @@ -179,16 +179,56 @@ impl Drop for Speaker<'_> { } } +/// How soon a picture held by [`heard`] looks at the playhead again: another +/// broadcast's playhead moves, or re-anchors across a gap, without telling +/// anyone reading its clock, so a held picture polls it. +pub const RECHECK: Duration = Duration::from_millis(10); + /// Whether a picture stamped `timestamp` may show at `now` on `schedule`: /// while audio plays, only once the sound stamped with it has played, so a /// stall in the audio alone holds the picture too. -pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> bool { +pub fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> bool { schedule.target().is_none() || schedule .playhead(now) .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) } +/// What to do at `now` with a sound stamped `timestamp` that plays on +/// `schedule`, another track's clock. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Follow { + /// Write it to sound at this moment. + Write(Instant), + /// Look again at this moment. + Wait(Instant), + /// Its moment has passed, or the clock has none. + Drop, +} + +/// Decide what to do with a followed sound from the clock as it is now, not +/// as it was when the sound arrived, so a re-anchor moves it. The clock's own +/// sound stamped alike must have been written first: when that stalls, this +/// one waits with the pictures, until its moment has passed. +pub(crate) fn follow(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> Follow { + let Some(sounds) = schedule.due(timestamp) else { + return Follow::Drop; + }; + if sounds < now { + return Follow::Drop; + } + // The clock re-anchors without telling its readers, so a wait for a + // moment it may yet move earlier is rechecked too. + let write = sounds - WRITE_LEAD; + if write > now { + return Follow::Wait(write.min(now + RECHECK)); + } + if !heard(schedule, timestamp, sounds) { + return Follow::Wait(now + RECHECK); + } + Follow::Write(sounds) +} + /// Writes placed frames into a positioned bus channel, each at the bus /// position its moment maps to, so it sounds exactly then. /// @@ -330,6 +370,31 @@ mod tests { assert!(!heard(&schedule, stamp(1_100), at(500)), "shown past its sound"); } + /// A followed sound, rechecked until then, is written the bus's latency + /// before its moment once the clock's own sound stamped alike has been, + /// waits while that has not, and is dropped once its moment has passed. + #[test] + fn a_followed_sound_waits_for_the_clocks_own() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let schedule = clock.consume(); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + assert_eq!( + follow(&schedule, stamp(1_000), at(0)), + Follow::Wait(at(10)), + "rechecked" + ); + assert_eq!(follow(&schedule, stamp(1_000), at(95)), Follow::Wait(at(100))); + assert_eq!(follow(&schedule, stamp(1_000), at(100)), Follow::Write(at(150))); + // Past what the clock's own sound has written: it stalled. + assert_eq!(follow(&schedule, stamp(1_100), at(200)), Follow::Wait(at(210))); + assert_eq!(follow(&schedule, stamp(1_100), at(250)), Follow::Wait(at(260))); + assert_eq!(follow(&schedule, stamp(1_100), at(251)), Follow::Drop); + let fresh = clock.clone().reset(VIDEO_DELAY).consume(); + assert_eq!(follow(&fresh, stamp(1_000), at(0)), Follow::Drop, "no moment yet"); + } + /// A stall longer than the target re-buffers: the next frame sounds the /// target after it arrived, and the pictures stamped with it wait as long. #[test] diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 8dc5e9a..b101060 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -31,7 +31,8 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, WRITE_LEAD, Writer}; +use crate::playout::{Follow, Place, Speaker, VIDEO_DELAY, WRITE_LEAD, Writer, follow}; +pub use crate::playout::{MAX_DELAY, heard}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -370,6 +371,20 @@ impl Sources { } keep }); + // A layer plays on its broadcast source's clock, so it stops with it + // and starts again on the replacement's. + let orphaned: Vec = self + .feeds + .iter() + .filter( + |(_, feed)| matches!(&feed.source, Source::Layer { broadcast, .. } if !self.feeds.contains_key(broadcast)), + ) + .map(|(name, _)| name.clone()) + .collect(); + for name in orphaned { + let feed = self.feeds.remove(&name).expect("listed above"); + leave(&self.mixes, &feed); + } // A mix ends once its last handle drops, which may be a reader's. self.mixes.retain(|name, _| collection.audio_scene(name).is_some()); for scene in &collection.audio_scenes { @@ -386,10 +401,25 @@ impl Sources { .collect(); let mut drivers = Vec::new(); - for (name, source) in collection.sources.iter().filter(|(_, source)| source.opens()) { + // Layers last, once the broadcast sources whose clocks they play on run. + let (layers, opens): (Vec<_>, Vec<_>) = collection + .sources + .iter() + .filter(|(_, source)| source.opens()) + .partition(|(_, source)| matches!(source, Source::Layer { .. })); + for (name, source) in opens.into_iter().chain(layers) { if !self.feeds.contains_key(name) { let held = collection.transition.stinger() == Some(name.as_str()); - let (feed, driver) = open(name, source, &self.context, held); + let follows = match source { + Source::Layer { broadcast, .. } => match collection.sources.get(broadcast) { + Some(Source::Broadcast { path }) => { + self.feeds[broadcast].clock.clone().map(|clock| (path.clone(), clock)) + } + _ => None, + }, + _ => None, + }; + let (feed, driver) = open(name, source, &self.context, held, follows); self.feeds.insert(name.clone(), feed); drivers.push(driver); } @@ -498,7 +528,7 @@ pub struct Feed { runs: Runs, /// Set once a media source's file turns out to hold no sound. silent: Arc, - /// A broadcast source's playout clock. + /// A broadcast source's playout clock, or the one a layer plays on. clock: Option, } @@ -661,7 +691,7 @@ impl Feed { /// picture alone. pub fn stalled(&self, now: Instant) -> Option { let stall = match self.source { - Source::Camera { .. } | Source::Broadcast { .. } => Stall::Picture, + Source::Camera { .. } | Source::Broadcast { .. } | Source::Layer { .. } => Stall::Picture, Source::Microphone { .. } | Source::SystemAudio => Stall::Sound, _ => return None, }; @@ -681,6 +711,13 @@ impl Feed { (self.status() == Status::Live).then(|| clock.target().unwrap_or(VIDEO_DELAY)) } + /// When a broadcast source plays what its broadcast stamps, or a layer + /// what its broadcast source's does, so what else that broadcast carries + /// can play on the same clock. `None` for any other source. + pub fn clock(&self) -> Option<&timeline::Consumer> { + self.clock.as_ref() + } + /// The source's gain, mute, delay, and meter in every mix, if it has sound. /// A media file without any has none once it opens. pub fn strip(&self) -> Option<&Strip> { @@ -727,6 +764,9 @@ pub struct Driver { runs: Runs, silent: Arc, clock: Option, + /// A layer's broadcast path, and the clock of the broadcast source it + /// plays on. + follows: Option<(String, timeline::Consumer)>, } /// Where a source's own track goes, its claim on the local catalog's listing @@ -844,6 +884,7 @@ impl Driver { self.media(&self.context.dir.join(path), *looping, cues, listing).await } Source::Broadcast { path } => self.broadcast(path).await, + Source::Layer { source, .. } => self.layer(source).await, Source::Browser { url, width, height } => self.browser(url, *width, *height).await, Source::Replay => { let plays = self.plays.clone().expect("a replay source has plays"); @@ -894,6 +935,7 @@ impl Driver { | Source::Image { .. } | Source::Media { .. } | Source::Broadcast { .. } + | Source::Layer { .. } | Source::Browser { .. } | Source::Overlay { .. } | Source::Clone { .. } @@ -1072,23 +1114,52 @@ 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 clock = self.clock.clone().expect("a broadcast source has a clock"); + self.follow(path, |broadcast| { + *self.played.lock().unwrap() = Played { + broadcast: Some(broadcast.clone()), + ..Played::default() + }; + // Each instance's timestamps are its own. + clock = clock.clone().reset(VIDEO_DELAY); + Box::pin(self.play(broadcast, path, clock.clone())) + }) + .await + } + + /// Play the source called `source` that the broadcast this layer follows + /// publishes independently, from each instance of it, on the clock of the + /// broadcast source playing it. + async fn layer(&self, source: &str) -> Result<(), Error> { + let Some((path, schedule)) = &self.follows else { + return Err(Error::Invalid(format!( + "layer `{}` follows no broadcast source", + self.name + ))); + }; + self.follow(path, |broadcast| { + Box::pin(self.play_layer(broadcast, path, source, schedule)) + }) + .await + } + + /// Play each instance of the broadcast at `path` with `start` as it comes + /// online, until the next replaces it or it goes offline. + async fn follow<'a>( + &'a self, + path: &str, + mut start: impl FnMut(moq_net::broadcast::Consumer) -> Playing<'a>, + ) -> Result<(), Error> { 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.available() => match event { Ok(Some(broadcast)) => { set(&self.status, Status::Live); - *self.played.lock().unwrap() = Played { - broadcast: Some(broadcast.clone()), - ..Played::default() - }; - // Each instance's timestamps are its own. - clock = clock.reset(VIDEO_DELAY); - playing = Some(Box::pin(self.play(broadcast, path, clock.clone()))); + playing = Some(start(broadcast)); } Ok(None) => { playing = None; @@ -1108,6 +1179,213 @@ impl Driver { } } + /// Decode one instance of a studio broadcast's source called `source`: + /// the first video and audio renditions its entry under the catalog's + /// [`SECTION`](crate::publish::SECTION) `sources` offers, switching as + /// the catalog changes them, until the catalog ends. Both play on + /// `schedule`, the program's clock, without moving it. + async fn play_layer( + &self, + broadcast: moq_net::broadcast::Consumer, + path: &str, + source: &str, + schedule: &timeline::Consumer, + ) -> Result<(), Error> { + let format = moq_mux::catalog::CatalogFormat::detect(path).unwrap_or_default(); + let mut catalogs = moq_mux::catalog::Consumer::::new(&broadcast, format).await?; + let mut video: Option = None; + let mut audio: Option = None; + let mut video_playing: Playing<'_> = Box::pin(std::future::pending()); + let mut audio_playing: Playing<'_> = Box::pin(std::future::pending()); + loop { + tokio::select! { + catalog = catalogs.next() => { + let Some(catalog) = catalog? else { + return Ok(()); + }; + let listed = catalog.ext.studio.sources.get(source); + let next_video = listed.map(|listed| pick(listed.video.enabled())).transpose()?.flatten(); + let next_audio = listed.map(|listed| pick(listed.audio.enabled())).transpose()?.flatten(); + // A source without video, such as an image or a color, is a still. + let still = listed.and_then(|listed| { + let mut stills = listed.binary.tracks.iter(); + let (id, config) = stills.find(|(_, config)| config.broadcast.is_none() && config.mime.as_deref().is_some_and(|mime| mime.starts_with("image/")))?; + Some(Reference::track(config, id).to_string()) + }); + let track = match &next_video { + Some((id, config)) => Some(Reference::track(config, id).to_string()), + None => still.clone(), + }; + if track != video { + video = track; + video_playing = match (next_video, still) { + (Some((id, config)), _) => Box::pin(self.follow_video(&broadcast, id, config, schedule)), + (None, Some(still)) => Box::pin(self.follow_still(&broadcast, still)), + (None, None) => { + self.picture.lock().unwrap().clear(); + Box::pin(std::future::pending()) + } + }; + } + let track = next_audio.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + if track != audio { + audio = track; + audio_playing = match next_audio { + Some((id, config)) => Box::pin(self.follow_audio(&broadcast, id, config, schedule)), + None => Box::pin(std::future::pending()), + }; + } + } + result = &mut video_playing => { + result?; + video_playing = Box::pin(std::future::pending()); + } + result = &mut audio_playing => { + result?; + audio_playing = Box::pin(std::future::pending()); + } + } + } + } + + /// Show the newest picture of the track `name`, a PNG or JPEG file per + /// group, as [`Still`] publishes one, whenever it arrives: a still has no + /// moment. + async fn follow_still(&self, broadcast: &moq_net::broadcast::Consumer, name: String) -> Result<(), Error> { + let mut track = broadcast.track(&name)?.subscribe(None).await?; + while let Some(mut group) = track.recv_group().await? { + if let Some(frame) = group.read_frame().await? { + self.show(decode(&frame.payload).map_err(|err| Error::Image(format!("{name}: {err}")))?); + } + } + Ok(()) + } + + /// Show each picture once `schedule`, another track's clock, says it is + /// due and its sound stamped alike has played, the newest such one when + /// several are, as [`play_video`](Self::play_video) does. + async fn follow_video( + &self, + broadcast: &moq_net::broadcast::Consumer, + name: String, + config: hang::catalog::VideoConfig, + schedule: &timeline::Consumer, + ) -> Result<(), Error> { + let mut options = moq_video::decode::Options::new(); + options.start = moq_video::decode::Start::Latest; + options.decoder.output = self.context.output; + options.delay = moq_video::decode::Delay { + min: Some(VIDEO_DELAY), + max: Some(MAX_DELAY), + }; + let mut consumer = moq_video::decode::Consumer::new(broadcast, &config, name, options).await?; + let mut queue = VecDeque::::new(); + let mut ended = false; + loop { + let now = kio::clock::now(); + // Counted first: `present` holds the clock's lock while it iterates. + let heard_count = queue + .iter() + .take_while(|frame| heard(schedule, frame.timestamp, now)) + .count(); + let ready = queue.iter().take(heard_count).map(|frame| frame.timestamp); + if let Some(index) = schedule.present(ready, now) { + let frame = queue.drain(..=index).next_back().expect("an index into the queue"); + self.show(Picture::Frame(frame)); + } + if ended && queue.is_empty() { + return Ok(()); + } + // Its moment, or once due, soon to see whether its sound has played. + let wake = queue.front().and_then(|frame| match schedule.due(frame.timestamp) { + Some(due) if due > now => Some(due), + _ if !heard(schedule, frame.timestamp, now) => Some(now + crate::playout::RECHECK), + _ => None, + }); + tokio::select! { + read = consumer.read(), if !ended && queue.len() < QUEUED => match read? { + Some(moq_video::decode::Read::Frame(frame)) => queue.push_back(frame), + Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => {} + None => ended = true, + }, + () = kio::clock::sleep_until(wake.unwrap_or(now)), if wake.is_some() => {} + } + } + } + + /// Play a track's sound into the bus when `schedule`, another track's + /// clock, says each frame sounds and its own sound stamped alike has been + /// written, dropping a frame too late for its moment, or before the clock + /// has one. + async fn follow_audio( + &self, + broadcast: &moq_net::broadcast::Consumer, + name: String, + config: hang::catalog::AudioConfig, + schedule: &timeline::Consumer, + ) -> Result<(), Error> { + let Some(strip) = self.strip.clone() else { + return Ok(()); + }; + let bus = self.context.bus.config(); + let mut options = moq_audio::decode::Options::new(); + options.start = moq_audio::decode::Start::Latest; + options.delay = moq_audio::decode::Delay { + min: None, + max: Some(MAX_DELAY), + }; + options.output.format = moq_audio::Format::F32; + options.output.sample_rate = Some(bus.sample_rate); + options.output.layout = Some(bus.layout); + let mut consumer = moq_audio::decode::Consumer::new(broadcast, &config, name, options).await?; + let mut writer = self.writer(strip)?; + + // Each frame waits here until its lead before it sounds. + let mut queue = VecDeque::::new(); + let mut ended = false; + loop { + let now = kio::clock::now(); + let mut write = None; + while let Some(frame) = queue.front() { + match follow(schedule, frame.timestamp, now) { + Follow::Write(sounds) => { + let frame = queue.pop_front().expect("peeked"); + writer.write(sounds, &frame.data).await?; + } + Follow::Wait(at) => { + write = Some(at); + break; + } + Follow::Drop => { + queue.pop_front(); + } + } + } + if ended && queue.is_empty() { + return Ok(()); + } + tokio::select! { + read = consumer.read(), if !ended => match read? { + Some(moq_audio::decode::Read::Frame(frame)) => { + // One due past the longest target would hold whatever a + // peer stamps. + let now = kio::clock::now(); + let placed = schedule.due(frame.timestamp).is_some_and(|sounds| { + sounds >= now && sounds <= now + MAX_DELAY + WRITE_LEAD + }); + if placed { + queue.push_back(frame); + } + } + // A hole or an encoder restart shows in the stamps that follow. + Some(_) => {} + None => ended = true, + }, + () = kio::clock::sleep_until(write.unwrap_or(now)), if write.is_some() => {} + } + } + } + /// Decode one instance of a broadcast: the first video and audio renditions /// each catalog offers, switching when a catalog update changes them, until /// the catalog ends. Both play on `clock`. @@ -2028,8 +2306,15 @@ pub(crate) fn pick<'a, C: Reference + Clone + 'a>( } /// Start the feed for `source`, and the driver that fills it, `held` from -/// playing a media file until the first [`Feed::play`]. -fn open(name: &str, source: &Source, context: &Context, held: bool) -> (Feed, Driver) { +/// playing a media file until the first [`Feed::play`]. A layer `follows` its +/// broadcast's path and the clock of the broadcast source playing it. +fn open( + name: &str, + source: &Source, + context: &Context, + held: bool, + follows: Option<(String, timeline::Consumer)>, +) -> (Feed, Driver) { let picture = Mailbox::default(); let delivered = Delivered::default(); let strip = source.has_audio().then(Strip::default); @@ -2082,7 +2367,10 @@ fn open(name: &str, source: &Source, context: &Context, held: bool) -> (Feed, Dr cues, runs: runs.clone(), silent: silent.clone(), - clock: clock.as_ref().map(timeline::Producer::consume), + clock: match &follows { + Some((_, schedule)) => Some(schedule.clone()), + None => clock.as_ref().map(timeline::Producer::consume), + }, }; let driver = Driver { name: name.to_string(), @@ -2100,6 +2388,7 @@ fn open(name: &str, source: &Source, context: &Context, held: bool) -> (Feed, Dr runs, silent, clock, + follows, }; (feed, driver) } @@ -2110,11 +2399,12 @@ impl Source { /// system audio, or media file as captured or decoded, before its gain and /// mute. /// - /// A broadcast is already published by its own publisher, and a clone shows - /// its source's track, its filters carried by the layout rather than encoded - /// again. Web pages and overlays are seen only in the program, since H.264 - /// would encode their transparency over black, a LUT is looked up, never - /// shown, and a scene is drawn from other sources. + /// A broadcast, or a layer of one, is already published by its own + /// publisher, and a clone shows its source's track, its filters carried by + /// the layout rather than encoded again. Web pages and overlays are seen + /// only in the program, since H.264 would encode their transparency over + /// black, a LUT is looked up, never shown, and a scene is drawn from other + /// sources. pub fn publishes(&self) -> bool { match self { Source::Camera { .. } @@ -2127,6 +2417,7 @@ impl Source { | Source::Media { .. } | Source::Replay => true, Source::Broadcast { .. } + | Source::Layer { .. } | Source::Browser { .. } | Source::Overlay { .. } | Source::Clone { .. } @@ -2175,14 +2466,17 @@ async fn capture(source: moq_audio::capture::Source) -> Result Result<(Picture, bytes::Bytes), Error> { let bytes = std::fs::read(path).map_err(|err| Error::Image(format!("{}: {err}", path.display())))?; - let image = image::load_from_memory(&bytes) - .map_err(|err| Error::Image(format!("{}: {err}", path.display())))? - .into_rgba8(); - let picture = Picture::Rgba { + let picture = decode(&bytes).map_err(|err| Error::Image(format!("{}: {err}", path.display())))?; + Ok((picture, bytes.into())) +} + +/// A PNG or JPEG file's picture, within the image crate's default limits. +fn decode(bytes: &[u8]) -> Result { + let image = image::load_from_memory(bytes)?.into_rgba8(); + Ok(Picture::Rgba { size: moq_video::Size::new(image.width(), image.height()), pixels: image.into_raw(), - }; - Ok((picture, bytes.into())) + }) } /// Helpers for tests across the crate. @@ -2400,7 +2694,7 @@ mod tests { let (context, _origin) = context(dir.path()); let start = kio::clock::now(); let opened = |source: Source| { - let (feed, driver) = open("source", &source, &context, false); + let (feed, driver) = open("source", &source, &context, false, None); set(&driver.status, Status::Live); (feed, driver) }; @@ -2693,9 +2987,9 @@ mod tests { } /// A video track on its own timestamp base. - async fn rendition( + async fn rendition( broadcast: &moq_net::broadcast::Producer, - timebase: moq_mux::catalog::Timebase, + timebase: moq_mux::catalog::Timebase, ) -> (moq_video::encode::Producer, moq_video::encode::Encoder) { let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); config.kind = moq_video::encode::Kind::Software; @@ -2776,6 +3070,155 @@ mod tests { .await; } + /// A layer shows each picture when its broadcast source's clock says, as + /// the program's picture stamped alike shows, not its own delay after it + /// arrives. + #[tokio::test(start_paused = true)] + async fn a_layer_plays_on_its_broadcasts_clock() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let mut catalog = crate::publish::Catalog::new(&mut broadcast, config).unwrap(); + let (mut program, mut encoder) = rendition(&broadcast, catalog.timebase()).await; + // The camera's track is in the program's broadcast, listed under its + // name in the catalog's own section rather than as the program's. + let mut local = moq_net::broadcast::Info::new().produce(); + let listing = moq_mux::catalog::Producer::new(&mut local, Default::default()).unwrap(); + let (mut cam, mut cam_encoder) = rendition(&broadcast, listing.timebase()).await; + publication.announce().unwrap(); + + let mut collection = collection(); + collection.sources.insert( + "cam".into(), + Source::Layer { + broadcast: "guest".into(), + source: "cam".into(), + }, + ); + collection.validate().unwrap(); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let layer = sources.get("cam").unwrap(); + let clock = sources.get("guest").unwrap().clock.clone().unwrap(); + + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + picture(&mut program, &mut encoder, 1_000); + let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + let due = until(|| clock.due(stamp)).await; + assert_eq!(due, arrived + VIDEO_DELAY); + + // Half the delay later, so on a clock of its own it would show late. + until_time(arrived + VIDEO_DELAY / 2).await; + picture(&mut cam, &mut cam_encoder, 1_000); + let listed = until(|| Some(listing.snapshot()).filter(|listed| !listed.video.renditions.is_empty())).await; + catalog + .mutate(|root| { + root.ext.studio.sources.insert("cam".into(), listed); + }) + .unwrap(); + until_time(due - Duration::from_millis(1)).await; + assert!(layer.next_picture().is_none(), "shown before its moment"); + until_time(due).await; + assert_eq!(shown(layer).await, Duration::from(stamp)); + assert_eq!(layer.delay(), Some(VIDEO_DELAY), "the program's"); + }) + .await; + } + + /// A layer whose source is a still, an image file on a binary track, + /// shows it as it arrives. + #[tokio::test] + async fn a_layer_shows_a_still() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let publication = origin + .route("room/guest", moq_net::origin::RouteOptions::hidden()) + .unwrap(); + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let mut catalog = crate::publish::Catalog::new(&mut broadcast, config).unwrap(); + let mut track = broadcast.create_track("source/logo.png", None).unwrap(); + let pixel = image::RgbaImage::from_raw(1, 1, vec![255, 0, 0, 255]).unwrap(); + let mut png = std::io::Cursor::new(Vec::new()); + pixel.write_to(&mut png, image::ImageFormat::Png).unwrap(); + track.write_frame(None, bytes::Bytes::from(png.into_inner())).unwrap(); + let mut still = hang::catalog::BinaryConfig::new(hang::catalog::Mode::Snapshot); + still.mime = Some("image/png".into()); + let mut logo = hang::Catalog::default(); + logo.binary.tracks.insert("source/logo.png".into(), still); + catalog + .mutate(|root| { + root.ext.studio.sources.insert("logo".into(), logo); + }) + .unwrap(); + publication.announce().unwrap(); + + let mut collection = collection(); + collection.sources.insert( + "logo".into(), + Source::Layer { + broadcast: "guest".into(), + source: "logo".into(), + }, + ); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let layer = sources.get("logo").unwrap(); + let pixels = until(|| match layer.next_picture() { + Some(Picture::Rgba { pixels, .. }) => Some(pixels), + _ => None, + }) + .await; + assert_eq!(pixels, [255, 0, 0, 255]); + }) + .await; + } + + /// A layer stops with its broadcast source, and starts again on its + /// replacement's clock. + #[tokio::test] + async fn a_layer_follows_its_broadcast_sources_replacement() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut collection = collection(); + collection.sources.insert( + "cam".into(), + Source::Layer { + broadcast: "guest".into(), + source: "cam".into(), + }, + ); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + assert_eq!(drivers.len(), 3); + start(&sources, drivers).await; + collection.sources.insert( + "guest".into(), + Source::Broadcast { + path: "room/other".into(), + }, + ); + let mut names: Vec = sources + .sync(&collection) + .iter() + .map(|driver| driver.name().to_string()) + .collect(); + names.sort(); + assert_eq!(names, ["cam", "guest"]); + }) + .await; + } + /// A video track replaced by one stamped earlier starts its own delay over, /// rather than showing every picture on arrival against the old track's. #[tokio::test(start_paused = true)] diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 208ed9f..541a880 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -838,7 +838,7 @@ pub const AUDIO_BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::fr /// How often the audio thread advances the bus. Short enough that a block is never /// close to the bus's buffering, long enough to cost nothing. -const AUDIO_PERIOD: Duration = Duration::from_millis(10); +pub(crate) const AUDIO_PERIOD: Duration = Duration::from_millis(10); /// The longest [`Engine::close`] waits for the sources to stop and the relays /// to drain, so a source or relay that never answers cannot hold up quitting. @@ -3044,7 +3044,7 @@ fn browser_profile() -> anyhow::Result { /// Run a source on its own thread until its feed is dropped, or fail the /// source when its thread cannot start. Returns what closes as the source /// ends, having finished its own track. -fn spawn(driver: Driver) -> tokio::sync::oneshot::Receiver<()> { +pub(crate) fn spawn(driver: Driver) -> tokio::sync::oneshot::Receiver<()> { // Handed over once the thread exists, since a failed spawn drops what it // was given and the driver must stay here to report why. let (handoff, received) = mpsc::sync_channel::(1); @@ -3069,7 +3069,10 @@ fn spawn(driver: Driver) -> tokio::sync::oneshot::Receiver<()> { /// Start a thread called `name` that runs `task` with its own clock and IO, /// returning why if either cannot start. -fn clocked(name: String, task: impl AsyncFnOnce() + Send + 'static) -> anyhow::Result> { +pub(crate) fn clocked( + name: String, + task: impl AsyncFnOnce() + Send + 'static, +) -> anyhow::Result> { // The thread starts before its runtime is built and handed over, since a // runtime dropped in the async caller, as a refused spawn would, panics. let (handoff, received) = mpsc::sync_channel::(1); diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index c7a3191..00349bc 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -32,6 +32,7 @@ mod snap; mod ui; mod uplink; mod vcam; +mod watch; #[derive(usage_rs::Cli, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] @@ -41,7 +42,15 @@ pub struct Config { /// The scene collection to open, saved as it changes. If it does not exist, /// guided setup proposes one and creates it. #[usage(value_hint = usage_rs::ValueHint::FilePath, extensions("toml"))] - pub scenes: PathBuf, + pub scenes: Option, + + /// Play a studio broadcast instead, as a viewer sees it: the program with + /// each source published independently over it. The URL's path names the + /// broadcast, as `moq play`'s does, such as + /// `https://relay.example/anon/studio`. A prototype, until `moq play` plays + /// the layer layout. + #[usage(long)] + pub watch: Option, /// The broadcast path the program is published at. #[usage(long, default = "studio")] @@ -85,6 +94,30 @@ pub struct Config { } impl Config { + /// The scene file to open, refusing `--watch` beside the studio's own flags. + fn scenes(&self) -> anyhow::Result> { + if self.watch.is_none() { + let scenes = self + .scenes + .as_deref() + .context("name a scene file to open, or --watch a broadcast")?; + return Ok(Some(scenes)); + } + anyhow::ensure!( + self.scenes.is_none(), + "--watch plays a broadcast, so it opens no scene file" + ); + anyhow::ensure!( + self.client.url.is_none() && self.failover.is_none(), + "--watch names the relay in its URL, so it takes no --connect or --failover" + ); + anyhow::ensure!( + self.guests.room.is_none(), + "--watch publishes nothing, so it has no --room" + ); + Ok(None) + } + /// Refuse a config parsed from released spellings, naming what replaced each. fn check_deprecated(&self) -> anyhow::Result<()> { let mut deprecated = self.client.deprecated(); @@ -149,6 +182,12 @@ fn main() -> anyhow::Result { config.log.init()?; config.check_deprecated()?; moq_tokio::cli::check_env(&[Config::spec()], &[])?; + let Some(scenes) = config.scenes()? else { + let url = config.watch.clone().expect("checked by scenes"); + watch::run(config.client.clone(), config.quic.clone(), url)?; + return Ok(ExitCode::SUCCESS); + }; + let scenes = scenes.to_path_buf(); let runtime = tokio::runtime::Runtime::new()?; // The QUIC client binds its socket on the current runtime. @@ -157,8 +196,7 @@ fn main() -> anyhow::Result { let gpu = pollster::block_on(engine::Gpu::new())?; let recordings = match &config.recordings { Some(dir) => dir.clone(), - None => config - .scenes + None => scenes .parent() .filter(|dir| !dir.as_os_str().is_empty()) .unwrap_or(Path::new(".")) @@ -177,20 +215,19 @@ fn main() -> anyhow::Result { recordings, room, }; - let exists = config - .scenes + let exists = scenes .try_exists() - .with_context(|| format!("reading {}", config.scenes.display()))?; + .with_context(|| format!("reading {}", scenes.display()))?; let app = match exists { true => { - let (autosave, loaded) = autosave::Autosave::open(config.scenes.clone())?; + let (autosave, loaded) = autosave::Autosave::open(scenes.clone())?; App::Starting(runtime.block_on(engine::Engine::start(launch, autosave, loaded))?) } // Nothing is saved until setup starts the studio, so quitting before then // leaves no file and the next launch runs setup again. false => App::Setup(Box::new(setup::Setup::new( launch, - config.scenes.clone(), + scenes.clone(), runtime.handle().clone(), )?)), }; @@ -427,7 +464,7 @@ mod tests { ]) .expect("a scene file and a relay should parse"); config.check_deprecated().unwrap(); - assert_eq!(config.scenes, PathBuf::from("show.toml")); + assert_eq!(config.scenes().unwrap(), Some(Path::new("show.toml"))); assert_eq!(config.broadcast, "studio"); assert!(!config.on_demand, "the program encodes whatever the demand by default"); assert!(config.guests.room.is_none(), "guests are opt-in"); @@ -463,4 +500,24 @@ mod tests { assert!(check(refused).is_err(), "{refused:?} was accepted"); } } + + /// `--watch` opens no scene file and names its own relay, and the studio + /// needs a scene file. + #[test] + fn watch_takes_no_scene_file_or_relay() { + let parse = |args: &[&str]| { + let args: Vec<_> = args.iter().map(std::ffi::OsStr::new).collect(); + Config::parse_from(&args).unwrap() + }; + let watch = parse(&["--watch", "https://relay.example/anon/studio"]); + assert_eq!(watch.scenes().unwrap(), None); + for refused in [ + &["show.toml", "--watch", "https://a/b"][..], + &["--watch", "https://a/b", "--connect", "https://a/"], + &["--watch", "https://a/b", "--room", "guests"], + &[], + ] { + assert!(parse(refused).scenes().is_err(), "{refused:?} was accepted"); + } + } } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 004cd8f..c4228f8 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -314,13 +314,13 @@ impl Studio { } /// A compositor texture as egui shows it. -struct Registered { - id: egui::TextureId, +pub(crate) struct Registered { + pub(crate) id: egui::TextureId, texture: wgpu::Texture, } impl Registered { - fn new(render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) -> Self { + pub(crate) fn new(render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) -> Self { let view = texture.create_view(&Default::default()); let id = render .renderer @@ -333,7 +333,7 @@ impl Registered { } /// Show `texture` from now on, if a resize replaced the one shown. - fn follow(&mut self, render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) { + pub(crate) fn follow(&mut self, render: &eframe::egui_wgpu::RenderState, texture: &wgpu::Texture) { if self.texture == *texture { return; } @@ -4508,6 +4508,7 @@ fn describe(source: &Source) -> String { Source::Replay => "replay".into(), Source::Lut { path } => format!("LUT: {}", path.display()), Source::Scene { scene } => format!("scene: {scene}"), + Source::Layer { broadcast, source } => format!("{source}, published in {broadcast}"), Source::Program => "program".into(), } } diff --git a/rs/moq-studio/src/watch.rs b/rs/moq-studio/src/watch.rs new file mode 100644 index 0000000..1f7c916 --- /dev/null +++ b/rs/moq-studio/src/watch.rs @@ -0,0 +1,712 @@ +//! `--watch`: play a studio broadcast as a viewer would, the program with each +//! source the [layer layout](moq_scene::layout) names composited over it and +//! mixed with it, on the program's media clock. +//! +//! A prototype for settling the layout's format before hang specifies one, +//! built on moq-scene: a collection the layout drives, rather than a scene +//! file, whose one scene draws the program, a [`Source::Broadcast`], under +//! each picture's [`Source::Layer`]. Each layer plays its own tracks on the +//! program's playout clock, each layout takes effect when the program reaches +//! the media time it is stamped with, and its pictures' keyframes play from +//! its `start` on that clock. A layout of another version is refused loudly, +//! and the program plays alone. + +use std::collections::{BTreeSet, VecDeque}; +use std::time::Duration; + +use anyhow::Context; +use kio::clock::Instant; +use moq_mux::timeline; +use moq_scene::layout::{self, Layout}; +use moq_scene::publish::Ext; +use moq_scene::source::{self, MAX_DELAY, Sources, Status}; +use moq_scene::{AudioScene, Canvas, Collection, Compositor, Item, Rect, Scene, Source}; + +use tokio::sync::watch; + +use crate::engine::{self, Gpu}; +use crate::mix; + +/// The program's source in the collection. +const PROGRAM: &str = "program"; + +/// What each layer's source name starts with, so none can take the +/// program's. +const LAYER: &str = "layer/"; + +/// The scene on program. +const SCENE: &str = "watch"; + +/// The canvas until the program's catalog or a layout gives its size. +const CANVAS: Canvas = Canvas { + width: 1280, + height: 720, + framerate: 30, +}; + +/// Play the broadcast at `url` until the window closes. Its path names the +/// broadcast, as `moq play`'s does, and `client` dials it. +pub fn run(client: moq_tokio::connect::Config, quic: moq_tokio::quic::Config, url: url::Url) -> anyhow::Result<()> { + let runtime = tokio::runtime::Runtime::new()?; + let _runtime = runtime.enter(); + let client = client.init(quic)?; + let origin = moq_tokio::origin::spawn(); + tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), "watching"); + let connection = client.with_subscriber(origin.clone()).connect(url.clone()); + + let gpu = pollster::block_on(Gpu::new())?; + let (bus, driver) = moq_audio::bus::Bus::new(Default::default())?; + let config = bus.config().clone(); + let start = Instant::now(); + let (mixer, mixed) = source::Mixer::new(driver, start); + engine::clocked("audio".into(), async move || advance(start, config, mixer).await)?; + let monitor = match mix::Monitor::start(runtime.handle(), AudioScene::DEFAULT, None) { + Ok(monitor) => Some(monitor), + Err(err) => { + tracing::warn!(error = format!("{err:#}"), "playing without sound"); + None + } + }; + + let sources = Sources::new(source::Context { + bus, + mixed, + origin: origin.consume(), + dir: std::env::current_dir()?, + output: moq_video::render::output(&gpu.device), + publish: None, + browser: None, + }); + let collection = collection(None, CANVAS)?; + let mut compositor = Compositor::new(&gpu.device, &gpu.queue, collection.canvas)?; + compositor.set_previewing(false); + compositor.set_preview(Some(SCENE.into())); + compositor.cut(&collection, &sources)?; + + let (read, _) = watch::channel(Read::default()); + runtime.spawn(follow(origin.consume(), read.clone())); + + let watch = Watch { + url: moq_tokio::RedactedUrl::new(&url).to_string(), + read, + collection, + synced: false, + layout: None, + size: None, + sources, + compositor, + monitor, + shown: None, + }; + let options = eframe::NativeOptions { + viewport: egui::ViewportBuilder::default() + .with_title("kixel.live Studio") + .with_app_id("moq-studio") + .with_inner_size([1280.0, 720.0]), + wgpu_options: eframe::egui_wgpu::WgpuConfiguration { + // The compositor's program reaches the screen without leaving the GPU. + wgpu_setup: eframe::egui_wgpu::WgpuSetup::Existing(eframe::egui_wgpu::WgpuSetupExisting { + instance: gpu.instance, + adapter: gpu.adapter, + device: gpu.device, + queue: gpu.queue, + }), + ..Default::default() + }, + ..Default::default() + }; + let ran = eframe::run_native( + "moq-studio-watch", + options, + Box::new(move |cc| { + let render = cc + .wgpu_render_state + .clone() + .context("the window is not drawn with wgpu")?; + Ok(Box::new(Window { render, watch })) + }), + ); + runtime.block_on(async { + let _ = connection.close().await; + }); + ran.map_err(|err| anyhow::anyhow!("the window failed: {err}")) +} + +/// Advance the bus on the wall clock from `start`, so each mix plays. +async fn advance(start: Instant, config: moq_audio::bus::Config, mut mixer: source::Mixer) { + let rate = u128::from(config.sample_rate); + let mut clock = kio::clock::interval_at(start, engine::AUDIO_PERIOD); + clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); + let mut written: u128 = 0; + loop { + let tick = clock.tick().await; + let due = tick.saturating_duration_since(start).as_nanos() * rate / 1_000_000_000; + let frames = (due - written) as usize; + if frames > 0 { + mixer.advance(frames); + written = due; + } + } +} + +/// What the broadcast's catalog and layout track say, for the window to apply. +#[derive(Default)] +struct Read { + /// The program's picture size, from its catalog. + size: Option<(u32, u32)>, + /// Each layout read, or `None` for the program alone, and the media time + /// it takes effect at, or `None` for at once, oldest first. + pending: VecDeque<(Option, Option)>, + /// Why the layout was refused, so the program plays alone. + refused: Option, +} + +impl Read { + /// Forget what an instance of the broadcast said, as another replaces it + /// or it goes offline: its stamps are not the next one's. + fn reset(&mut self) { + self.size = None; + self.refused = None; + self.pending.clear(); + self.pending.push_back((None, None)); + } + + /// Queue `layout` to take effect at `at`, or at once for `None`, which + /// supersedes every layout still waiting so a stalled program cannot hold + /// them up. The program plays at most [`MAX_DELAY`] behind what arrives, + /// so of the layouts stamped that far before it, all due by now, only the + /// newest is kept. + fn push(&mut self, layout: Option, at: Option) { + match at { + None => self.pending.clear(), + Some(at) => { + let passed = self.pending.iter().take_while(|(_, stamp)| { + stamp.is_none_or(|stamp| Duration::from(stamp) + MAX_DELAY <= Duration::from(at)) + }); + self.pending.drain(..passed.count().saturating_sub(1)); + } + } + self.pending.push_back((layout, at)); + } + + /// Take the newest layout due at `now`: each takes effect once the program + /// plays the frame it is stamped with on `schedule`, and its sound has + /// played. `None` while none is due; `Some(None)` for the program alone. + fn take_due(&mut self, schedule: Option<&timeline::Consumer>, now: Instant) -> Option> { + let mut taken = None; + while let Some((_, at)) = self.pending.front() { + let due = match at { + None => true, + Some(at) => schedule.is_some_and(|schedule| { + schedule.due(*at).is_some_and(|due| due <= now) && source::heard(schedule, *at, now) + }), + }; + if !due { + break; + } + taken = self.pending.pop_front().map(|(layout, _)| layout); + } + taken + } +} + +/// Read each instance of the root broadcast as it comes online: the program's +/// size from its catalog, and each layout from the layout track while the +/// catalog lists it. +async fn follow(origin: moq_net::origin::Consumer, read: watch::Sender) { + let mut instances = match origin.follow("", None) { + Ok(instances) => instances, + Err(err) => { + tracing::error!(%err, "cannot follow the broadcast"); + return; + } + }; + let mut playing = None; + loop { + tokio::select! { + event = instances.available() => match event { + Ok(Some(broadcast)) => { + read.send_modify(Read::reset); + playing = Some(Box::pin(instance(broadcast, read.clone()))); + } + Ok(None) => { + read.send_modify(Read::reset); + playing = None; + } + Err(err) => { + tracing::error!(%err, "the broadcast's announcements failed"); + return; + } + }, + result = async { playing.as_mut().expect("guarded").await }, if playing.is_some() => { + playing = None; + if let Err(err) = result { + tracing::warn!(error = format!("{err:#}"), "cannot read the broadcast's layout"); + } + } + } + } +} + +/// Read one instance of the broadcast until its catalog ends. +async fn instance(broadcast: moq_net::broadcast::Consumer, read: watch::Sender) -> anyhow::Result<()> { + let mut catalogs = + moq_mux::catalog::Consumer::::new(&broadcast, moq_mux::catalog::CatalogFormat::default()).await?; + let mut layouts: Option> = None; + // A refused layout stays refused for this instance. + let mut refused = false; + let alone = || read.send_modify(|read| read.push(None, None)); + loop { + tokio::select! { + catalog = moq_mux::catalog::Stream::next(&mut catalogs) => { + let Some(catalog) = catalog? else { + alone(); + return Ok(()); + }; + let size = catalog.video.renditions.values().find_map(|config| config.coded_width.zip(config.coded_height)); + read.send_if_modified(|read| std::mem::replace(&mut read.size, size) != size); + let listed = catalog.json.tracks.contains_key(layout::TRACK); + if listed && layouts.is_none() && !refused { + let track = broadcast.track(layout::TRACK)?; + layouts = Some(moq_json::snapshot::Consumer::new(track, Default::default())); + } else if !listed && layouts.take().is_some() { + alone(); + } + } + next = async { layouts.as_mut().expect("guarded").next().await }, if layouts.is_some() => match next { + Ok(Some(timed)) => read.send_modify(|read| read.push(Some(timed.value), timed.at)), + Ok(None) => { + layouts = None; + alone(); + } + Err(err) => { + tracing::error!(%err, "refusing the layer layout; playing the program alone"); + layouts = None; + refused = true; + read.send_modify(|read| { + read.refused = Some(err.to_string()); + read.push(None, None); + }); + } + }, + } + } +} + +/// The collection `layout` drives on a canvas the program's `size` when it +/// has none: the program, and over it each picture's layer, with the program +/// and each sound's layer in the audio scene on program. +fn collection(layout: Option<&Layout>, size: Canvas) -> anyhow::Result { + let mut collection = Collection::new(); + collection.canvas = match layout { + Some(layout) => Canvas { + width: layout.canvas.width, + height: layout.canvas.height, + framerate: size.framerate, + }, + None => size, + }; + collection + .sources + .insert(PROGRAM.into(), Source::Broadcast { path: String::new() }); + let mut items = vec![Item::new(PROGRAM, Rect::fill(&collection.canvas))]; + let mut sounds = BTreeSet::from([PROGRAM.to_string()]); + let mut layer = |source: &str| { + let name = format!("{LAYER}{source}"); + collection.sources.insert( + name.clone(), + Source::Layer { + broadcast: PROGRAM.into(), + source: source.into(), + }, + ); + name + }; + let mut audio = Vec::new(); + for picture in layout.iter().flat_map(|layout| &layout.pictures) { + items.push(Item { + source: layer(&picture.item.source), + ..picture.item.clone() + }); + } + for sound in layout.iter().flat_map(|layout| &layout.sounds) { + let name = layer(&sound.source); + audio.push((name.clone(), sound.audio.clone())); + sounds.insert(name); + } + collection.audio.extend(audio); + collection.scenes = vec![Scene { + items, + ..Scene::new(SCENE) + }]; + collection.audio_scenes = vec![AudioScene { + program: true, + sounds, + ..AudioScene::new(AudioScene::DEFAULT) + }]; + collection.validate().context("cannot play the layer layout")?; + Ok(collection) +} + +/// The player: the collection the layout drives, and what plays it. +struct Watch { + /// What the window says it waits for. + url: String, + read: watch::Sender, + collection: Collection, + /// Whether the sources run what `collection` holds. + synced: bool, + /// The layout in effect, or `None` for the program alone. + layout: Option, + /// The program's size, from its catalog. + size: Option<(u32, u32)>, + sources: Sources, + compositor: Compositor, + monitor: Option, + /// The program as the window shows it. + shown: Option, +} + +impl Watch { + /// Apply what was read and is now due at `now`, and draw the program. + fn render(&mut self, now: Instant) -> anyhow::Result<()> { + let schedule = self.sources.get(PROGRAM).and_then(|feed| feed.clock()).cloned(); + let mut changed = !self.synced; + self.read.send_if_modified(|read| { + if read.size != self.size { + self.size = read.size; + changed = true; + } + if let Some(layout) = read.take_due(schedule.as_ref(), now) { + self.layout = layout; + changed = true; + } + false + }); + + if changed { + let size = match self.size { + Some((width, height)) => Canvas { + width, + height, + ..CANVAS + }, + None => CANVAS, + }; + let collection = match collection(self.layout.as_ref(), size) { + Ok(collection) => collection, + Err(err) => { + tracing::error!(error = format!("{err:#}"), "playing the program alone"); + self.read.send_modify(|read| read.refused = Some(format!("{err:#}"))); + self.layout = None; + collection(None, size)? + } + }; + if collection.canvas != self.collection.canvas { + self.compositor.fit(&collection)?; + } + self.collection = collection; + for driver in self.sources.sync(&self.collection) { + engine::spawn(driver); + } + if let (Some(monitor), Some(mix)) = (&self.monitor, self.sources.mix(AudioScene::DEFAULT)) { + monitor.play(mix.clone()); + } + self.synced = true; + } + + // The keyframes play from `start` as far as the program has played, + // following the clock as it re-anchors or stalls with the sound. + if let (Some(layout), Some(schedule)) = (&self.layout, &schedule) { + let went_on = match schedule.playhead(now) { + Some(head) => now.checked_sub(head.saturating_sub(layout.start)), + None => moq_net::Timestamp::try_from(layout.start) + .ok() + .and_then(|start| schedule.due(start)), + }; + if let Some(went_on) = went_on { + self.compositor.set_program_start(went_on); + } + } + + self.compositor.render(&self.collection, &self.sources, now)?; + Ok(()) + } + + /// What the window says over the program, if anything. + fn status(&self) -> Option { + if let Some(refused) = &self.read.borrow().refused { + return Some(format!("Playing the program alone: {refused}")); + } + match self.sources.get(PROGRAM).map(|feed| feed.status()) { + Some(Status::Live) => None, + Some(Status::Failed(err)) => Some(format!("{}: {err}", self.url)), + _ => Some(format!("Waiting for {}", self.url)), + } + } +} + +/// The window around the player. +struct Window { + render: eframe::egui_wgpu::RenderState, + watch: Watch, +} + +impl eframe::App for Window { + fn ui(&mut self, ui: &mut egui::Ui, _frame: &mut eframe::Frame) { + let watch = &mut self.watch; + if let Err(err) = watch.render(kio::clock::now()) { + tracing::error!(error = format!("{err:#}"), "cannot draw the program"); + } + let texture = watch.compositor.program(); + match &mut watch.shown { + Some(shown) => shown.follow(&self.render, texture), + None => watch.shown = Some(crate::ui::Registered::new(&self.render, texture)), + } + let rect = ui.max_rect(); + let painter = ui.painter(); + painter.rect_filled(rect, 0.0, egui::Color32::BLACK); + let canvas = watch.collection.canvas; + let scale = (rect.width() / canvas.width as f32).min(rect.height() / canvas.height as f32); + let fit = egui::Rect::from_center_size( + rect.center(), + egui::vec2(canvas.width as f32 * scale, canvas.height as f32 * scale), + ); + let uv = egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(1.0, 1.0)); + let id = watch.shown.as_ref().expect("registered above").id; + painter.image(id, fit, uv, egui::Color32::WHITE); + if let Some(status) = watch.status() { + painter.text( + rect.left_top() + egui::vec2(12.0, 12.0), + egui::Align2::LEFT_TOP, + status, + egui::FontId::proportional(16.0), + egui::Color32::LIGHT_RED, + ); + } + ui.ctx().request_repaint(); + } +} + +#[cfg(test)] +mod tests { + use moq_scene::layout::{Picture, Sound, Version}; + use moq_scene::{Audio, Filter, Keyframe, Size}; + + use super::*; + + /// A camera named as the program is, placed twice, and a microphone. + fn layout(start: Duration) -> Layout { + let mut cam = Item::new( + "program", + Rect { + x: 10.0, + y: 10.0, + width: 64.0, + height: 36.0, + }, + ); + cam.filters = vec![Filter::Opacity { amount: 0.5 }]; + cam.keyframes = vec![Keyframe { + x: 100.0, + ..Keyframe::new(Duration::from_secs(1)) + }]; + let picture = Picture { + item: cam, + priority: 0, + toggle: false, + }; + Layout { + version: Version, + canvas: Size { + width: 640, + height: 360, + }, + start, + pictures: vec![picture.clone(), picture], + sounds: vec![Sound { + source: "mic".into(), + audio: Audio { + gain: 0.5, + delay: Duration::from_millis(80), + ..Default::default() + }, + priority: 0, + toggle: false, + }], + } + } + + fn stamp(secs: u64) -> moq_net::Timestamp { + moq_net::Timestamp::try_from(Duration::from_secs(secs)).unwrap() + } + + /// The layout's pictures draw over the program from layers named apart + /// from it, even one named as it is, a source placed twice shares one + /// layer, and the audio scene on program plays the program and each + /// sound at its gain and delay. + #[test] + fn a_layout_drives_the_collection() { + let layout = layout(Duration::ZERO); + let driven = collection(Some(&layout), CANVAS).unwrap(); + assert_eq!((driven.canvas.width, driven.canvas.height), (640, 360)); + let items = &driven.scene(SCENE).unwrap().items; + let sources: Vec<&str> = items.iter().map(|item| item.source.as_str()).collect(); + assert_eq!(sources, [PROGRAM, "layer/program", "layer/program"]); + assert_eq!(items[1].keyframes, layout.pictures[0].item.keyframes); + assert_eq!( + driven.sources["layer/program"], + Source::Layer { + broadcast: PROGRAM.into(), + source: "program".into() + } + ); + let heard = driven.heard(AudioScene::DEFAULT); + assert_eq!(heard, BTreeSet::from([PROGRAM, "layer/mic"])); + assert_eq!(driven.audio["layer/mic"], layout.sounds[0].audio); + + let alone = collection(None, CANVAS).unwrap(); + assert_eq!(alone.sources.len(), 1, "the program alone"); + assert_eq!(alone.canvas, CANVAS); + } + + /// Of the layouts the program must have reached by the time a new one + /// arrives, only the newest waits. + #[test] + fn only_the_newest_passed_layout_waits() { + let mut read = Read::default(); + for secs in [1, 2, 3, 4] { + read.push(None, Some(stamp(secs))); + } + read.push(None, Some(stamp(5))); + let stamps: Vec<_> = read.pending.iter().map(|(_, at)| *at).collect(); + assert_eq!(stamps, [Some(stamp(3)), Some(stamp(4)), Some(stamp(5))]); + } + + /// Another instance of the broadcast starts from nothing: the last one's + /// stamps and refusal are not its own. + #[test] + fn a_new_instance_forgets_the_last() { + let mut read = Read::default(); + read.push(None, Some(stamp(1))); + read.refused = Some("another version".into()); + read.reset(); + assert_eq!(read.pending.len(), 1); + assert!(read.pending[0].0.is_none() && read.pending[0].1.is_none()); + assert!(read.size.is_none() && read.refused.is_none()); + } + + /// Once the layout track is unlisted, the program plays alone at once, + /// even while a stalled program holds back a layout still waiting. + #[tokio::test(start_paused = true)] + async fn unlisting_supersedes_a_waiting_layout() { + let clock = timeline::Producer::new(Duration::from_millis(100)); + let audio = timeline::Audio { + end: Duration::from_secs(1), + buffered: Duration::ZERO, + target: Duration::from_millis(100), + }; + clock.audio(audio, Instant::now()); + let schedule = clock.consume(); + + let (broadcast, catalog, read, reader) = broadcast(); + let mut publisher = layout::Publisher::new(broadcast.clone(), catalog.clone()); + publisher.update(layout(Duration::ZERO), stamp(5)).unwrap(); + pending(&read, 1).await; + tokio::time::advance(Duration::from_secs(60)).await; + let later = Instant::now(); + let take = || { + let mut taken = None; + read.send_if_modified(|read| { + taken = read.take_due(Some(&schedule), later); + false + }); + taken + }; + assert_eq!(take(), None, "the program stalled before it"); + + publisher.finish().unwrap(); + alone(&read).await; + assert_eq!(take(), Some(None), "the program alone"); + assert!(read.borrow().pending.is_empty()); + reader.abort(); + } + + /// A layout of another version, as a newer studio would publish. + #[derive(serde::Serialize)] + struct Newer { + version: u32, + } + + /// A broadcast with a catalog, and the instance reader's view of it. + fn broadcast() -> ( + moq_net::broadcast::Producer, + moq_scene::publish::Catalog, + watch::Sender, + tokio::task::JoinHandle>, + ) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); + let catalog = moq_scene::publish::Catalog::new(&mut broadcast, config).unwrap(); + let (read, _) = watch::channel(Read::default()); + let reader = tokio::spawn(instance(broadcast.consume(), read.clone())); + (broadcast, catalog, read, reader) + } + + /// What `read` holds once it has `count` layouts waiting, with their + /// stamps as durations, since one read back carries its track's timescale. + async fn pending(read: &watch::Sender, count: usize) -> Vec<(Option, Option)> { + let mut reads = read.subscribe(); + let read = reads.wait_for(|read| read.pending.len() >= count).await.unwrap(); + let pending = read.pending.iter().cloned(); + pending.map(|(layout, at)| (layout, at.map(Duration::from))).collect() + } + + /// Wait until the newest thing `read` holds is the program alone. + async fn alone(read: &watch::Sender) { + let mut reads = read.subscribe(); + let alone = |read: &Read| read.pending.back().is_some_and(|(_, at)| at.is_none()); + reads.wait_for(alone).await.unwrap(); + } + + /// The reader hands over each layout at the media time it took effect, + /// and the program alone once the catalog stops listing the track, until + /// it lists it again. + #[tokio::test] + async fn reads_each_layout() { + let (broadcast, catalog, read, reader) = broadcast(); + let mut publisher = layout::Publisher::new(broadcast.clone(), catalog.clone()); + publisher.update(layout(Duration::ZERO), stamp(1)).unwrap(); + let first = (Some(layout(Duration::ZERO)), Some(Duration::from_secs(1))); + assert_eq!(pending(&read, 1).await, [first]); + publisher.finish().unwrap(); + alone(&read).await; + assert_eq!( + pending(&read, 1).await, + [(None, None)], + "the program alone once unlisted" + ); + publisher.update(layout(Duration::from_secs(2)), stamp(2)).unwrap(); + let again = (Some(layout(Duration::from_secs(2))), Some(Duration::from_secs(2))); + assert_eq!(pending(&read, 2).await[1], again, "listed again"); + reader.abort(); + } + + /// A layout of another version is refused, saying why, and the program + /// plays alone. + #[tokio::test] + async fn refuses_another_version() { + let (broadcast, catalog, read, reader) = broadcast(); + let info = moq_net::track::Info::default().with_timescale(moq_mux::Clock::TIMESCALE); + let track = broadcast.create_track(layout::TRACK, info).unwrap(); + let options = moq_mux::json::Options::default().with_deltas(false); + let mut newer = catalog + .json_snapshot::(track, moq_mux::json::Config::default(), options) + .unwrap(); + newer.update(&Newer { version: 2 }, stamp(1)).unwrap(); + assert_eq!(pending(&read, 1).await, [(None, None)]); + let refused = read.borrow().refused.clone().unwrap(); + assert!(refused.contains("layer layout version 2 is not supported"), "{refused}"); + reader.abort(); + } +}