diff --git a/Cargo.lock b/Cargo.lock index d603dd6..7e78405 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3484,6 +3484,7 @@ dependencies = [ "moq-stats", "moq-tokio", "moq-video", + "mp4-atom", "pollster", "qrcodegen", "tempfile", diff --git a/README.md b/README.md index e91d0bf..e848152 100644 --- a/README.md +++ b/README.md @@ -96,6 +96,7 @@ Every panel is a tab you can drag, split, or stack. | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | | Output | Set the canvas size and frame rate, and see the broadcast path, whether the program is encoding, and the encoder's work. | | Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem. | +| Replay | Replay the last moments of the program or a source in a replay source, slowed down if you like, or save them as a clip. | A new source, from the Sources or Devices panel, is placed over whatever the Items panel shows. The **top layer** is drawn over every scene, in preview and @@ -264,10 +265,10 @@ rect = { x = 1600, y = 40, width = 280, height = 120 } Source kinds are `camera`, `display`, and `microphone` (each with an optional `id`, the default device without one), `window` (with an `id`), `system_audio` (macOS), `image` (PNG or JPEG), `color` (`#rrggbb` or `#rrggbbaa`), -`broadcast`, `clone` (with the `source` it shows), and `lut` (a 3D Adobe or +`broadcast`, `clone` (with the `source` it shows), `lut` (a 3D Adobe or Resolve `.cube` file, which no item places, but which grades what a `lut` -filter names it in). The Devices panel fills in ids. `[[top]]` items make up the top -layer. +filter names it in), and `replay`, where the Replay panel plays replays. The +Devices panel fills in ids. `[[top]]` items make up the top layer. `[video.]` sets a source's `crop` and `filters`, wherever it is shown, and an item's `crop`, `filters`, and `blend` its own. A `crop` trims a fraction @@ -403,6 +404,37 @@ macOS. Every other source, every source on Windows or on a GPU that cannot import, and every still image delivers its frames from the CPU, which the compositor uploads. +## Replay + +Add a `replay` source and place it like any other, such as full screen in a +scene of its own. In the Replay panel, pick the program or a source, a length +up to 30 seconds, and a speed of 100%, 50%, or 25%, then **Replay** plays that +many seconds up to the click in the replay source, which is transparent before +and after. **Cut** to its scene to put it on program. Sound replays only at +100%, through the replay source's strip in the Audio panel. A new replay cuts +off the one playing, and a stretch the source was offline or paused replays +as the gap it was. + +**Save clip** writes the same seconds, in real time, beside the scene file as a +regular MP4: `show-clip-1.mp4` for `show.toml`, then `show-clip-2.mp4`, and so +on. Nothing is re-encoded, so a clip starts at the keyframe before those +seconds, up to one keyframe interval earlier. It shows nothing after them, so +video with B-frames can end early: the pictures shown before the end that refer +to one shown after it are left out, along with everything decoded after that +one. That is usually a fraction of a second, but it can reach back as far as +the keyframe for a slow or deeply reordered source. The panel says where once +it is written, shortly after the click. + +Nothing records in the background for this. The program and each source's +track keep their last 30 seconds for late viewers anyway, and a replay copies +from those, so a replay needs `--connect` and goes back at most that far. With +`--on-demand`, the program has no past from before someone watched it. A +broadcast source replays from what the studio received of it, and only over a +relay connection whose protocol version can address a track by time; an older +one refuses with the reason in the panel. An image or a color has no past to +replay. A replay source publishes each replay as its own track while it plays, +like a camera, so viewers composing the scene themselves see it too. + ## Health The Health panel grades the stream every second on the thresholds moq's diff --git a/quest/m0/README.md b/quest/m0/README.md index 6caa376..e6d37e1 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -87,7 +87,6 @@ repository. - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker - [Live encode control](/quest/m0/encode-control.md) - simulcast rungs that follow the bandwidth grant, live bitrate and resolution, keyframes on request - [Audio sources report stalls](/quest/m0/audio-stall.md) - a dead microphone or system audio reads stalled like a camera -- [Instant replay and DVR](/quest/m0/replay.md) - replays and clips from held groups, and viewer scrubbing through the archive - [HTML/CSS overlays](/quest/m0/overlays.md) - script-free HTML and CSS graphics through Blitz natively and the DOM in the browser - [Animation and transitions](/quest/m0/animation.md) - move, slide, wipe, and stinger transitions, and keyframed item properties - [ML video effects](/quest/m0/effects.md) - background removal and other ML effects, baked into their source's track diff --git a/quest/m0/compose/sources.md b/quest/m0/compose/sources.md index f3c69aa..fa7727b 100644 --- a/quest/m0/compose/sources.md +++ b/quest/m0/compose/sources.md @@ -28,7 +28,7 @@ in its ML effects, and a composing viewer mixes the levels itself. Encode like every first-party publisher (decided 2026-10-28, from [#7101](https://github.com/moq-dev/moq/issues/7101)): an unwatched source -still encodes, so a late viewer or a [replay](/quest/m0/replay.md) +still encodes, so a late viewer or a replay finds its recent past, but costs no upload. ✅ 2026-11-01 (maintainer, on diff --git a/quest/m0/replay.md b/quest/m0/replay.md deleted file mode 100644 index 6961543..0000000 --- a/quest/m0/replay.md +++ /dev/null @@ -1,34 +0,0 @@ -# [M] Instant replay and DVR - -## Goal - -`moq-studio` replays the last seconds of any source or the program on demand: -several lengths, slow motion, and a one-click clip saved to disk, with no -separate always-on recording. Viewers can scrub back through the live -broadcast as far as the relay or archive holds it. - -## Plan - -The groups are already on hand: the studio's own recent groups and the -relay's cache, addressed by presentation time through subscription ranges -(`doc/concept/moq-lite.md`). A replay plays from those rather than a RAM-only -frame buffer like OBS's. Retention past the relay cache is the archive's job. - -Sources and the program encode whatever the subscriber demand (decided -2026-10-28, from [#7101](https://github.com/moq-dev/moq/issues/7101); see -[source tracks](/quest/m0/compose/sources.md)), so a track nobody -watched still holds its recent groups, as far back as its `max_age`. Replay -needs no arming and no local subscriber. A studio started with `--on-demand` -encodes the program only while someone watches, so the program has no past -from before its first viewer to replay (decided ✅ 2026-11-01, maintainer, on -[moq-dev/moq#7554](https://github.com/moq-dev/moq/pull/7554)). A stretch where a source was offline -or removed replays as a gap rather than being skipped. Acceptance covers the -case where every viewer composes sources themselves and nothing outside the -studio subscribes to the program. -Replay published to viewers is a new source in the scene, so it composes like -any other. - -## Related - -- [Archive DVR](https://github.com/moq-dev/moq/blob/main/quest/m2/archive-dvr.md) - viewer scrubbing beyond the relay cache -- [Recording](/quest/m0/recording.md) - clips use the same file writer diff --git a/rs/moq-scene/Cargo.toml b/rs/moq-scene/Cargo.toml index e02278a..027cf1d 100644 --- a/rs/moq-scene/Cargo.toml +++ b/rs/moq-scene/Cargo.toml @@ -44,12 +44,12 @@ moq-video = { workspace = true, features = ["render"] } serde = { workspace = true } serde_with = { workspace = true } thiserror = { workspace = true } -tokio = { workspace = true, features = ["macros", "time"] } +tokio = { workspace = true, features = ["macros", "sync", "time"] } toml = { workspace = true } tracing = { workspace = true } [dev-dependencies] -# An encoder on every machine, for the source track tests. +# An encoder on every machine, for the source track and replay tests. moq-video = { workspace = true, features = ["openh264"] } tempfile = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "test-util"] } diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index 296ea80..a8e98ca 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -839,6 +839,10 @@ pub enum Source { /// The file, relative to the scene file's directory unless absolute. path: PathBuf, }, + /// Where instant replays play: the recent past of the program or another + /// source, on demand through [`Feed::replay`](crate::source::Feed::replay). + /// Transparent while nothing plays. + Replay, } impl Source { @@ -873,7 +877,7 @@ impl Source { pub fn has_audio(&self) -> bool { matches!( self, - Self::Microphone { .. } | Self::SystemAudio | Self::Broadcast { .. } + Self::Microphone { .. } | Self::SystemAudio | Self::Broadcast { .. } | Self::Replay ) } } diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 99f1be8..9a18b05 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -33,6 +33,7 @@ mod edit; mod effect; mod lut; pub mod publish; +pub mod replay; pub mod source; pub use clip::Clip; @@ -85,6 +86,10 @@ pub enum Error { #[error("unsupported source: {0}")] Unsupported(String), + /// The source stopped, with the reason, so it takes nothing more. + #[error("the source stopped: {0}")] + Stopped(String), + /// Capturing, decoding, or drawing video failed. #[error(transparent)] Video(#[from] moq_video::Error), diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 3d54ad3..35636e8 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -218,7 +218,7 @@ impl Drop for Listing<'_> { } } -/// A captured source's video, encoded frame by frame as it is shown. +/// A captured or replayed source's video, encoded frame by frame as it is shown. pub(crate) struct Video { /// Taken only to abort or finish it, as the track ends. producer: Option, @@ -237,10 +237,6 @@ impl Video { /// to learn exactly what the real one will write, so a machine that cannot /// encode another stream, such as a GPU at its hardware session cap, fails /// here with the reason. - #[cfg_attr( - not(feature = "capture"), - allow(dead_code, reason = "only capture sources encode video") - )] pub(crate) async fn new( listing: &Listing<'_>, size: moq_video::Size, @@ -267,10 +263,6 @@ impl Video { } /// Encode and publish `frame`, restamping it onto the studio clock. - #[cfg_attr( - not(feature = "capture"), - allow(dead_code, reason = "only capture sources encode video") - )] pub(crate) fn encode(&mut self, frame: &mut moq_video::Frame) -> Result<(), Error> { let captured = frame.timestamp.as_micros(); let epoch = *self @@ -299,6 +291,15 @@ impl Video { Ok(()) } + /// Mark a break, such as a gap in a replay: what follows does not continue + /// what came before, and a viewer arriving during it waits for what follows + /// rather than being served the last picture as if it were live. + pub(crate) fn discontinuity(&mut self) -> Result<(), Error> { + self.drain()?; + self.producer.as_mut().expect("open until it ends").discontinuity()?; + Ok(()) + } + /// Publish what the encoder still holds and close it, so a pipelined /// backend's last pictures are not lost when the encoder is replaced or the /// track ends. @@ -310,10 +311,6 @@ impl Video { } /// End the track, failed with `err` so viewers see why. - #[cfg_attr( - not(feature = "capture"), - allow(dead_code, reason = "only capture sources encode video") - )] pub(crate) fn abort(mut self, err: &Error) { if let Some(producer) = self.producer.take() { producer.abort(moq_net::Error::Transport(err.to_string())); @@ -392,7 +389,7 @@ mod tests { use super::*; use crate::collection::{Collection, Rgba, Source}; - use crate::source::test::{context, local, start}; + use crate::source::test::{context, local, start, until, until_time, video}; use crate::source::{Sources, Status}; /// A studio broadcast and catalog to publish into, and a reader of it. @@ -617,4 +614,64 @@ mod tests { }) .await; } + + /// A replay source publishes each replay as it shows, so viewers composing + /// the scene themselves see it too, gaps included. + #[tokio::test(start_paused = true)] + async fn a_replay_is_published_as_it_plays() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (publish, catalog, consumer) = studio(); + let (mut context, _origin) = context(dir.path()); + context.publish = Some(publish); + let mut collection = Collection::new(); + collection.sources.insert("replay".into(), Source::Replay); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + + // The broadcast left out its second second, a gap in the span. + let (broadcast, video, _live) = video(Some(1)).await; + let target = crate::replay::Target { + broadcast: broadcast.consume(), + video: Some(video), + audio: None, + end: moq_net::Timestamp::from_micros(2_500_000).unwrap(), + }; + let take = crate::replay::Take::new(&target, std::time::Duration::from_millis(2_500)).unwrap(); + let feed = sources.get("replay").unwrap(); + let begin = kio::clock::now(); + feed.replay(take, 1.0).unwrap(); + // The first picture is due at once, so the clock stays put while + // the encoder opens. + let listed = until(|| { + let listed = catalog.snapshot().ext.studio.sources.remove("replay"); + listed.filter(|listed| !listed.video.renditions.is_empty()) + }) + .await; + assert_eq!(feed.track(), Some(Status::Live)); + + // Into the gap, before the replay ends, a viewer reads the break. + let (name, config) = listed.video.renditions.iter().next().unwrap(); + let track = consumer + .track(hang::catalog::Reference::track(config, name)) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(None)) + .await + .unwrap(); + let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); + let mut track = moq_mux::container::Consumer::new(track, container); + until_time(begin + std::time::Duration::from_millis(1_500)).await; + let now = kio::Waiter::noop(); + until(|| match track.poll_read(&now) { + std::task::Poll::Ready(read) => match read.unwrap().expect("open while the replay plays") { + moq_mux::container::Read::End { .. } | moq_mux::container::Read::Gap { .. } => Some(()), + _ => None, + }, + std::task::Poll::Pending => None, + }) + .await; + }) + .await; + } } diff --git a/rs/moq-scene/src/replay.rs b/rs/moq-scene/src/replay.rs new file mode 100644 index 0000000..02dc9c7 --- /dev/null +++ b/rs/moq-scene/src/replay.rs @@ -0,0 +1,511 @@ +//! Instant replay: the recent past of a broadcast, held so it can be played +//! back at any speed or saved. +//! +//! The groups are already on hand. A track keeps its recent groups as far back +//! as its `max_age`, whoever watches, so a [`Take`] copies those covering a span +//! into a broadcast of its own. Nothing ages a take's groups, so a slow-motion +//! replay that ends long after its span left the track's window still has all +//! of it, and its tracks end where the span does, so an export of it finishes. +//! Copying shares each frame's payload rather than duplicating it. + +use std::collections::BTreeMap; +use std::task::Poll; +use std::time::Duration; + +use hang::catalog::{AudioConfig, Reference, VideoConfig}; + +use crate::Error; + +/// The longest span a take holds: as far back as a media track keeps its groups. +/// +/// Further back, a track holds nothing to copy, and a relay asked for it would +/// fetch it from an archive, so bounding the span bounds what a take can make +/// the studio hold to that many seconds of the broadcast. +pub const MAX_LENGTH: Duration = hang::container::MAX_AGE; + +/// The slowest a take plays, as a fraction of real time. +/// +/// A replay holds its take until it ends, so this bounds that to +/// [`MAX_LENGTH`] / `MIN_SPEED`, and keeps when each moment plays well within +/// the clock's range. +pub const MIN_SPEED: f32 = 0.01; + +/// What a take copies: a broadcast's video and audio renditions, and the +/// moment its span ends, on that broadcast's timeline. +#[derive(Clone)] +pub struct Target { + /// The broadcast whose tracks are copied. + pub broadcast: moq_net::broadcast::Consumer, + /// The video rendition, by catalog name. + pub video: Option<(String, VideoConfig)>, + /// The audio rendition, by catalog name. + pub audio: Option<(String, AudioConfig)>, + /// The newest moment the span covers. + pub end: moq_net::Timestamp, +} + +/// A span of a broadcast, copied into a broadcast of its own. +/// +/// It holds the groups covering the span, from the one in progress at its start +/// through the one in progress at its end, so the span decodes from its first +/// picture. Dropping it releases them. +/// +/// Frames are in decode order, so with B-frames a picture shown inside the span +/// can refer to one shown after its end. A take to play ([`Take::new`]) keeps +/// such references, since playing stops showing at the end anyway. A clip +/// ([`Take::clip`]) stops before the first frame shown after the end, since an +/// export shows everything it holds. It loses every frame decoded after that +/// one too, so it can end early by as much as the span that frame's group +/// reorders, back to the group's keyframe at worst. +pub struct Take { + broadcast: moq_net::broadcast::Producer, + /// Keep the copies and their listing readable. + _tracks: Vec, + _listing: moq_mux::catalog::Producer, + /// The video rendition, under its catalog name. + pub video: Option<(String, VideoConfig)>, + /// The audio rendition, under its catalog name. + pub audio: Option<(String, AudioConfig)>, + /// Where the span starts on the copied broadcast's timeline. + pub start: moq_net::Timestamp, + /// Where it ends. + pub end: moq_net::Timestamp, +} + +impl Take { + /// Hold the `length` of `target` that ends at its end, at most [`MAX_LENGTH`], + /// as far as its tracks hold it now. + /// + /// Copies only what is readable at once, so it never waits on a publisher: + /// a rendition its publisher has not described yet, or a frame or group + /// still to be written or still in transit, is left out, and a span ending + /// at the live edge may stop a frame or two short. + /// + /// Refuses a target with neither rendition, one served from another + /// broadcast, a length of zero or over [`MAX_LENGTH`], a broadcast that has + /// described neither rendition yet, and, over a wire that cannot address a + /// received track by time, a broadcast from a relay. + pub fn new(target: &Target, length: Duration) -> Result { + Self::cut(target, length, Cut::Decodable) + } + + /// Hold the `length` of `target` that ends at its end to save, as + /// [`Take::new`] does, but with nothing shown after the end. + pub fn clip(target: &Target, length: Duration) -> Result { + Self::cut(target, length, Cut::Shown) + } + + fn cut(target: &Target, length: Duration, cut: Cut) -> Result { + if length.is_zero() || length > MAX_LENGTH { + return Err(Error::Invalid(format!( + "a replay is above 0 and at most {} s long, not {length:?}", + MAX_LENGTH.as_secs() + ))); + } + if target.video.is_none() && target.audio.is_none() { + return Err(Error::Unsupported( + "the broadcast has no video or audio to replay".into(), + )); + } + let end = target.end; + let start = Duration::from(end).saturating_sub(length); + let start = moq_net::Timestamp::try_from(start) + .and_then(|start| start.convert(end.scale())) + .map_err(|err| Error::Invalid(format!("replay start: {err}")))?; + + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut tracks = Vec::new(); + let mut video = None; + if let Some((name, config)) = &target.video { + reference(config)?; + if let Some(track) = copy(target, &broadcast, Reference::track(config, name), start, cut)? { + tracks.push(track); + video = Some((name.clone(), config.clone())); + } + } + let mut audio = None; + if let Some((name, config)) = &target.audio { + reference(config)?; + if let Some(track) = copy(target, &broadcast, Reference::track(config, name), start, cut)? { + tracks.push(track); + audio = Some((name.clone(), config.clone())); + } + } + if tracks.is_empty() { + return Err(Error::Unsupported( + "the broadcast has described none of its tracks yet".into(), + )); + } + // Listed so an export of the take finds its renditions; HLS is for live + // viewers, which a take has none of. + let config = moq_mux::catalog::Config::default().with_hls(None); + let mut listing = moq_mux::catalog::Producer::new(&mut broadcast, config)?; + listing.mutate(|listed| { + listed.video.renditions = video.clone().into_iter().collect(); + listed.audio.renditions = audio.clone().into_iter().collect(); + })?; + listing.finish()?; + + Ok(Self { + broadcast, + _tracks: tracks, + _listing: listing, + video, + audio, + start, + end, + }) + } + + /// The take's broadcast, to decode or export. + pub fn consume(&self) -> moq_net::broadcast::Consumer { + self.broadcast.consume() + } +} + +/// Refuse a rendition served from another broadcast, which the take does not +/// subscribe to. +fn reference(config: &impl Reference) -> Result<(), Error> { + match config.broadcast() { + Some(_) => Err(Error::Unsupported( + "a rendition served from another broadcast cannot be replayed".into(), + )), + None => Ok(()), + } +} + +/// Where a take's groups stop in decode order, at the end of its span. +#[derive(Clone, Copy)] +enum Cut { + /// Through the last frame shown inside the span, with every reference it + /// needs, even one shown after the end. + Decodable, + /// Before the first frame shown after the end. + Shown, +} + +/// Copy the span of `name` in `target`, as far as it is readable now, into a +/// track of `broadcast`, and end that track. `None` while its publisher has +/// not described the track, so nothing of it is held. +fn copy( + target: &Target, + broadcast: &moq_net::broadcast::Producer, + name: &str, + start: moq_net::Timestamp, + cut: Cut, +) -> Result, Error> { + let range = moq_net::track::Range { + start: moq_net::track::Anchor::Time(start), + end: Some(moq_net::track::Anchor::Time(target.end)), + }; + // With the publisher's whole window, so none of the span is skipped as + // stale. + let subscription = moq_net::track::Subscription::default() + .with_ranges([range]) + .with_max_delay(None) + .with_order(moq_net::track::Order::Ascending); + let subscribing = target.broadcast.track(name)?.subscribe(subscription); + // Never woken, like the reads below: a publisher that never describes the + // track cannot hold the take up. + let mut source = match subscribing.poll_ok(&kio::Waiter::noop()) { + Poll::Ready(source) => source?, + Poll::Pending => return Ok(None), + }; + // Never stale: the take keeps its groups for as long as it lives. + let info = source.info().clone().with_max_age(None); + let mut track = broadcast.create_track(name, info)?; + match copy_held(&mut source, &mut track, target.end, cut) { + Ok(()) => track.finish()?, + Err(err) => { + let _ = track.clone().abort(moq_net::Error::Cancel); + return Err(err); + } + } + Ok(Some(track)) +} + +/// A group's frames as far as they were readable, and how reading it ended. +struct Held { + frames: Vec, + read: Result<(), moq_net::Error>, +} + +/// Copy every group and frame `source` has ready, up to `end` as `cut` says, +/// into `track`. +fn copy_held( + source: &mut moq_net::track::Subscriber, + track: &mut moq_net::track::Producer, + end: moq_net::Timestamp, + cut: Cut, +) -> Result<(), Error> { + // Never woken: anything not ready now is left out. + let now = kio::Waiter::noop(); + // Groups come in the order they reached the studio, which over a session + // need not be their own. + let mut held = BTreeMap::new(); + loop { + let mut group = match source.poll_recv_group(&now) { + Poll::Ready(group) => match group? { + Some(group) => group, + None => break, + }, + Poll::Pending => break, + }; + let mut frames = Vec::new(); + let read = loop { + match group.poll_read_frame(&now) { + Poll::Ready(Ok(Some(frame))) => frames.push(frame), + // A group still arriving is copied as far as it arrived. + Poll::Ready(Ok(None)) | Poll::Pending => break Ok(()), + Poll::Ready(Err(err)) => break Err(err), + } + }; + let inside = |frame: &moq_net::frame::Frame| frame.timestamp.is_none_or(|timestamp| timestamp <= end); + let kept = match cut { + Cut::Decodable => frames.iter().rposition(inside).map_or(0, |last| last + 1), + Cut::Shown => frames.iter().position(|frame| !inside(frame)).unwrap_or(frames.len()), + }; + // None kept is a group wholly past the span, or one the publisher gave + // up or that aged out before any of it was read, which replays as the + // gap it is. + if kept > 0 { + frames.truncate(kept); + held.insert(group.sequence, Held { frames, read }); + } + } + + if let Some(&first) = held.keys().next() { + track.start_at(first)?; + } + for (&sequence, held) in &held { + let mut into = track.create_group(moq_net::group::Info { sequence })?; + for frame in &held.frames { + into.write_frame(frame.timestamp, frame.payload.clone())?; + } + match &held.read { + Ok(()) => into.finish()?, + Err(err) => into.abort(err.clone())?, + } + } + + // Groups the span covers that are not here are gaps, not groups to come. + let copied: Vec = held.into_keys().collect(); + for pair in copied.windows(2) { + if pair[1] > pair[0] + 1 { + track.drop_groups(pair[0] + 1..pair[1], moq_net::StreamError::NotFound); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ts(millis: u64) -> moq_net::Timestamp { + moq_net::Timestamp::from_micros(millis * 1000).unwrap() + } + + /// Write group `sequence` of `track` with a frame at each of `at`, in + /// milliseconds and decode order, the frame at `at[i]` named + /// `{sequence}.{i}`. + fn write(track: &moq_net::track::Producer, sequence: u64, at: &[u64]) { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + for (index, at) in at.iter().enumerate() { + group + .write_frame(ts(*at), format!("{sequence}.{index}").into_bytes()) + .unwrap(); + } + group.finish().unwrap(); + } + + /// The frames of group `sequence` of [`broadcast`], three a third of a + /// second apart from its second. + fn second(sequence: u64) -> [u64; 3] { + [0, 333, 666].map(|at| sequence * 1000 + at) + } + + /// A video track with a group a second, from `0` up to but excluding + /// `groups` seconds, but for the group the publisher gave up at `lost`. + fn broadcast(groups: u64, lost: Option) -> (moq_net::broadcast::Producer, moq_net::track::Producer) { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("video", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + for sequence in 0..groups { + match lost == Some(sequence) { + true => track.drop_groups(sequence..sequence + 1, moq_net::StreamError::Cancel), + false => write(&track, sequence, &second(sequence)), + } + } + (broadcast, track) + } + + fn target(broadcast: &moq_net::broadcast::Producer, end: moq_net::Timestamp) -> Target { + let config = hang::catalog::VideoConfig::new(hang::catalog::H264 { + inline: true, + profile: 0x42, + constraints: 0, + level: 0x1f, + }); + Target { + broadcast: broadcast.consume(), + video: Some(("video".into(), config)), + audio: None, + end, + } + } + + /// Every frame of every group the take's track holds, by group. + async fn read(take: &Take) -> Vec<(u64, Vec)> { + let mut track = take + .consume() + .track("video") + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_delay(None)) + .await + .unwrap(); + let mut groups = Vec::new(); + while let Some(mut group) = track.recv_group().await.unwrap() { + let mut frames = Vec::new(); + while let Some(frame) = group.read_frame().await.unwrap() { + frames.push(String::from_utf8(frame.payload.to_vec()).unwrap()); + } + groups.push((group.sequence, frames)); + } + groups + } + + fn frames(group: u64, count: u64) -> Vec { + (0..count).map(|frame| format!("{group}.{frame}")).collect() + } + + /// A take holds the group in progress at its start through the frames up to + /// its end, and ends there. + #[tokio::test(start_paused = true)] + async fn a_take_holds_its_span() { + let (broadcast, _track) = broadcast(10, None); + let take = Take::new(&target(&broadcast, ts(7_500)), Duration::from_secs(3)).unwrap(); + assert_eq!(take.start, ts(4_500)); + assert_eq!( + read(&take).await, + [ + (4, frames(4, 3)), + (5, frames(5, 3)), + (6, frames(6, 3)), + (7, frames(7, 2)) + ] + ); + } + + /// A group missing from the span stays missing, rather than one its readers + /// wait for. + #[tokio::test(start_paused = true)] + async fn a_take_keeps_a_gap() { + let (broadcast, _track) = broadcast(10, Some(5)); + let take = Take::new(&target(&broadcast, ts(7_500)), Duration::from_secs(3)).unwrap(); + assert_eq!( + read(&take).await, + [(4, frames(4, 3)), (6, frames(6, 3)), (7, frames(7, 2))] + ); + } + + /// Groups that reached the studio out of order are held in their own order, + /// the first of them included. + #[tokio::test(start_paused = true)] + async fn a_take_orders_its_groups() { + let (broadcast, track) = broadcast(5, None); + write(&track, 6, &second(6)); + write(&track, 5, &second(5)); + let take = Take::new(&target(&broadcast, ts(6_500)), Duration::from_secs(3)).unwrap(); + assert_eq!( + read(&take).await, + [ + (3, frames(3, 3)), + (4, frames(4, 3)), + (5, frames(5, 3)), + (6, frames(6, 2)) + ] + ); + } + + /// A picture presented past the end is kept to play when one decoded after + /// it, and shown inside the span, may refer to it, and left out of a clip, + /// with the pictures that refer to it. + #[tokio::test(start_paused = true)] + async fn a_take_keeps_the_references_of_its_last_pictures() { + let (broadcast, track) = broadcast(4, None); + write(&track, 4, &[4_000, 4_400, 4_133, 4_266]); + let target = target(&broadcast, ts(4_200)); + let take = Take::new(&target, Duration::from_secs(1)).unwrap(); + assert_eq!(read(&take).await, [(3, frames(3, 3)), (4, frames(4, 3))]); + let clip = Take::clip(&target, Duration::from_secs(1)).unwrap(); + assert_eq!(read(&clip).await, [(3, frames(3, 3)), (4, frames(4, 1))]); + } + + /// A clip of a slow source that reorders deeply can end seconds early: at + /// one picture a second, decoded I0 P8 B1 ... B7, a clip to 7.5 s keeps + /// only the keyframe, since every picture after it refers to P8. + #[tokio::test(start_paused = true)] + async fn a_clip_can_end_back_at_its_keyframe() { + let (broadcast, track) = broadcast(0, None); + write(&track, 0, &[0, 8_000, 1_000, 2_000, 3_000, 4_000, 5_000, 6_000, 7_000]); + let target = target(&broadcast, ts(7_500)); + let clip = Take::clip(&target, Duration::from_secs(5)).unwrap(); + assert_eq!(read(&clip).await, [(0, frames(0, 1))]); + let take = Take::new(&target, Duration::from_secs(5)).unwrap(); + assert_eq!(read(&take).await, [(0, frames(0, 9))]); + } + + /// A take at the live edge holds what was written when it was taken, and + /// ends without waiting for more: a source may never write again, such as a + /// screen that stopped changing. + #[tokio::test(start_paused = true)] + async fn a_take_at_the_live_edge_holds_what_was_written() { + let (broadcast, track) = broadcast(7, None); + let mut live = track.create_group(moq_net::group::Info { sequence: 7 }).unwrap(); + live.write_frame(ts(7_000), b"7.0".to_vec()).unwrap(); + let take = Take::new(&target(&broadcast, ts(7_500)), Duration::from_secs(3)).unwrap(); + live.write_frame(ts(7_333), b"7.1".to_vec()).unwrap(); + assert_eq!( + read(&take).await, + [ + (4, frames(4, 3)), + (5, frames(5, 3)), + (6, frames(6, 3)), + (7, frames(7, 1)) + ] + ); + } + + /// A rendition its publisher has not described yet is left out rather than + /// waited for, and a take of nothing but such renditions is refused. + /// Outside any runtime, too, as the UI thread takes one. + #[test] + fn a_take_leaves_out_what_is_not_described() { + let (broadcast, _track) = broadcast(10, None); + // Answers no request, as a publisher that never replies. + let _requests = broadcast.requests(); + let mut target = target(&broadcast, ts(7_500)); + target.audio = Some(( + "audio".into(), + hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2), + )); + let take = Take::new(&target, Duration::from_secs(1)).unwrap(); + assert!(take.video.is_some()); + assert!(take.audio.is_none()); + + target.video = None; + let result = Take::new(&target, Duration::from_secs(1)); + assert!(matches!(result, Err(Error::Unsupported(_)))); + } + + #[test] + fn refuses_a_span_it_cannot_hold() { + let (broadcast, _track) = broadcast(1, None); + for length in [Duration::ZERO, MAX_LENGTH + Duration::from_millis(1)] { + let result = Take::new(&target(&broadcast, ts(500)), length); + assert!(matches!(result, Err(Error::Invalid(_))), "{length:?}"); + } + } +} diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 423f587..74880d4 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -15,15 +15,17 @@ use std::pin::Pin; use std::sync::{Arc, Mutex}; use std::time::Duration; -use hang::catalog::Reference; +use hang::catalog::{AudioConfig, Reference, VideoConfig}; use kio::clock::Instant; use moq_audio::bus::{Bus, Strip}; use moq_mux::catalog::Stream; +use tokio::sync::watch; use crate::Error; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::publish::{Claim, Listing, Publish, Still}; +use crate::replay::{MIN_SPEED, Take, Target}; /// A source's newest picture, waiting for the compositor to draw it. pub(crate) enum Picture { @@ -42,6 +44,37 @@ type Mailbox = Arc>>; /// When a source last delivered a picture, `None` before its first. type Shown = Arc>>; +/// What a broadcast source plays, for a replay of it to copy. +#[derive(Default)] +struct Played { + /// The instance of the broadcast online now. + broadcast: Option, + video: Option<(String, VideoConfig)>, + audio: Option<(String, AudioConfig)>, + /// The newest picture shown from it, where a replay of it ends. + newest: Option, +} + +type Playback = Arc>; + +/// A take for a replay source to play, and how fast. +struct Play { + take: Take, + speed: f32, +} + +/// A replay source's next replay, until its driver takes it, so a replay that +/// ended holds its take no longer. +type Next = Arc>>; + +/// A picture that shows nothing, for a replay source between replays. +fn clear() -> Picture { + Picture::Rgba { + size: moq_video::Size::new(1, 1), + pixels: vec![0; 4], + } +} + /// How long a camera or broadcast may go without a new picture before it reads /// as stalled. Either delivers several pictures a second while it works, so a /// second without one is a freeze a viewer sees. @@ -148,9 +181,51 @@ pub struct Feed { strip: Option, status: kio::Consumer, track: Option>, + played: Playback, + /// A replay source's next replay. + plays: Option>, } impl Feed { + /// What a replay of this broadcast source copies: the broadcast it plays, + /// up to the newest picture it showed, which is what was just seen. + /// + /// `None` until a broadcast source shows a picture, and for every other + /// kind of source, whose recent past is on its own track if anywhere. + pub fn target(&self) -> Option { + let played = self.played.lock().unwrap(); + Some(Target { + broadcast: played.broadcast.clone()?, + video: played.video.clone(), + audio: played.audio.clone(), + end: played.newest?, + }) + } + + /// Play `take` at `speed`, from [`MIN_SPEED`] to 1 for real time, in place + /// of whatever this replay source plays now. Its sound plays only in real + /// time. + /// + /// Refused by any other kind of source, and by a replay source that stopped. + pub fn replay(&self, take: Take, speed: f32) -> Result<(), Error> { + let Some(plays) = &self.plays else { + return Err(Error::Unsupported("only a replay source plays replays".into())); + }; + if !(MIN_SPEED..=1.0).contains(&speed) { + return Err(Error::Invalid(format!( + "a replay plays at {MIN_SPEED} to 1 times real time, not {speed}" + ))); + } + let play = Play { take, speed }; + // Fails once the driver is gone, which leaves the reason in the status. + plays.send(Arc::new(Mutex::new(Some(play)))).map_err(|_| { + Error::Stopped(match self.status() { + Status::Failed(reason) => reason, + status => format!("{status:?}"), + }) + }) + } + /// What this feed was opened for. pub fn source(&self) -> &Source { &self.source @@ -207,6 +282,8 @@ pub struct Driver { status: kio::Producer, /// The source's own track, when it publishes one. track: Option, + played: Playback, + plays: Option>, } /// Where a source's own track goes, its claim on the studio catalog's listing @@ -315,6 +392,10 @@ impl Driver { std::future::pending().await } Source::Broadcast { path } => self.broadcast(path).await, + Source::Replay => { + let plays = self.plays.clone().expect("a replay source has plays"); + self.replays(plays, listing).await + } Source::Clone { .. } => unreachable!("a clone opens no feed"), device => self.device(device, listing).await, } @@ -358,9 +439,8 @@ impl Driver { | Source::Image { .. } | Source::Broadcast { .. } | Source::Clone { .. } - | Source::Lut { .. } => { - unreachable!("not a device source") - } + | Source::Lut { .. } + | Source::Replay => unreachable!("not a device source"), }; let mut config = moq_video::capture::Config::default(); @@ -427,10 +507,15 @@ impl Driver { event = instances.next() => match event { Ok(moq_mux::follow::Event::Start(broadcast)) => { set(&self.status, Status::Live); + *self.played.lock().unwrap() = Played { + broadcast: Some(broadcast.clone()), + ..Played::default() + }; playing = Some(Box::pin(self.play(broadcast, path))); } Ok(moq_mux::follow::Event::End) => { playing = None; + *self.played.lock().unwrap() = Played::default(); set(&self.status, Status::Offline); } Err(err) => return Err(err.into()), @@ -473,6 +558,12 @@ impl Driver { let track = next.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); if track != video { video = track; + let mut played = self.played.lock().unwrap(); + played.video = next.clone(); + // The newest picture was the old track's, a moment the + // new one may not hold. + played.newest = None; + drop(played); video_playing = match next { Some((id, config)) => Box::pin(self.play_video(&broadcast, id, config)), None => Box::pin(std::future::pending()), @@ -482,6 +573,7 @@ impl Driver { let track = next.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); if track != audio { audio = track; + self.played.lock().unwrap().audio = next.clone(); audio_playing = match next { Some((id, config)) => Box::pin(self.play_audio(&broadcast, id, config)), None => Box::pin(std::future::pending()), @@ -513,6 +605,7 @@ impl Driver { let mut consumer = moq_video::decode::Consumer::new(broadcast, &config, name, options).await?; while let Some(read) = consumer.read().await? { if let moq_video::decode::Read::Frame(frame) = read { + self.played.lock().unwrap().newest = Some(frame.timestamp); self.show(Picture::Frame(frame)); } } @@ -552,8 +645,174 @@ impl Driver { } Ok(()) } + + /// Play each replay [`Feed::replay`] hands over, showing nothing between + /// them. A new one cuts off the one playing. + async fn replays(&self, mut plays: watch::Receiver, listing: Option<&Listing<'_>>) -> Result<(), Error> { + *self.picture.lock().unwrap() = Some(clear()); + set(&self.status, Status::Live); + + let mut playing: Option> = None; + loop { + tokio::select! { + changed = plays.changed() => { + // The feed is gone, which stops this driver anyway. + if changed.is_err() { + return std::future::pending().await; + } + let play = plays.borrow_and_update().lock().unwrap().take(); + // A replay cut off must not leave its last picture up until + // the new one's first. Between replays the picture is clear + // already, and clearing it again would show before the new + // one's first picture even when that is due at once. + if playing.is_some() { + *self.picture.lock().unwrap() = Some(clear()); + } + playing = play.map(|play| Box::pin(self.replay(play, listing)) as Playing<'_>); + } + result = async { playing.as_mut().expect("guarded").await }, if playing.is_some() => { + playing = None; + *self.picture.lock().unwrap() = Some(clear()); + set(&self.status, match result { + Ok(()) => Status::Live, + Err(err) => Status::Failed(err.to_string()), + }); + } + } + } + } + + /// Play one take at its speed, each picture and sound when its moment + /// comes: one `t` into the span plays `t / speed` after the replay starts. + /// Ends once the whole span has played. + /// + /// With `listing`, the pictures are also encoded onto the source's own + /// track as they show, so viewers composing the scene see the replay too. + async fn replay(&self, play: Play, listing: Option<&Listing<'_>>) -> Result<(), Error> { + set(&self.status, Status::Live); + let take = &play.take; + let begin = kio::clock::now(); + let at = |timestamp: moq_net::Timestamp| { + let into = Duration::from(timestamp).saturating_sub(Duration::from(take.start)); + begin + into.div_f32(play.speed) + }; + let broadcast = take.consume(); + + let video = async { + let Some((name, config)) = &take.video else { + return Ok(()); + }; + let mut options = moq_video::decode::Options::new(); + options.decoder.output = self.context.output; + options.delay = REPLAY; + let mut consumer = moq_video::decode::Consumer::new(&broadcast, config, name, options).await?; + // Opened at the first picture, which says its size. + let mut track: Option = None; + let mut opened = false; + while let Some(read) = consumer.read().await? { + match read { + // The group in progress at the start decodes from its + // keyframe, but shows only from the start on. + moq_video::decode::Read::Frame(frame) if frame.timestamp < take.start => {} + // One at the end would show for no time at all. + moq_video::decode::Read::Frame(frame) if frame.timestamp >= take.end => break, + moq_video::decode::Read::Frame(mut frame) => { + kio::clock::sleep_until(at(frame.timestamp)).await; + if let Some(listing) = listing.filter(|_| !std::mem::replace(&mut opened, true)) { + let framerate = config + .framerate + .filter(|rate| rate.is_finite() && *rate >= 1.0) + .and_then(|rate| moq_video::FrameRate::new(rate.round() as u32, 1).ok()) + .unwrap_or_else(|| moq_video::FrameRate::new(30, 1).expect("a valid frame rate")); + let video = + crate::publish::Video::new(listing, frame.size(), framerate, frame.surface.color()); + track = self.publish(video.await); + } + if let Some(video) = &mut track { + // Stamped when it shows, on a timeline of the replay's own. + frame.timestamp = moq_net::Timestamp::from_micros(begin.elapsed().as_micros() as u64) + .map_err(moq_video::Error::from)?; + if let Err(err) = video.encode(&mut frame) { + tracing::warn!(source = %self.name, %err, "cannot encode the replay"); + self.fail_track(Status::Failed(err.to_string())); + track.take().expect("encoding").abort(&err); + } + } + self.show(Picture::Frame(frame)); + } + // A stretch the broadcast was offline, paused, or lost + // replays as the gap it was, for viewers composing the + // scene too. Past the end, the replay is over anyway. + moq_video::decode::Read::Gap { timestamp, .. } + | moq_video::decode::Read::End { end: timestamp } => { + kio::clock::sleep_until(at(timestamp.clamp(take.start, take.end))).await; + *self.picture.lock().unwrap() = Some(clear()); + if let Some(video) = &mut track + && let Err(err) = video.discontinuity() + { + tracing::warn!(source = %self.name, %err, "cannot mark a gap in the replay"); + self.fail_track(Status::Failed(err.to_string())); + track.take().expect("encoding").abort(&err); + } + } + } + } + Ok::<_, Error>(()) + }; + + // Slowed down, sound would need stretching, so it plays only in real time. + let audio = async { + let (Some((name, config)), Some(strip), true) = (&take.audio, &self.strip, play.speed == 1.0) else { + return Ok(()); + }; + let bus = self.context.bus.config().clone(); + let mut options = moq_audio::decode::Options::new(); + options.delay = REPLAY; + 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.clone(), options).await?; + let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); + input.format = moq_audio::Format::F32; + let mut channel = self.context.bus.channel(input, strip.clone())?; + while let Some(read) = consumer.read().await? { + match read { + moq_audio::decode::Read::Frame(frame) if frame.timestamp < take.start => {} + moq_audio::decode::Read::Frame(frame) if frame.timestamp >= take.end => break, + moq_audio::decode::Read::Frame(frame) => { + kio::clock::sleep_until(at(frame.timestamp)).await; + channel.write(&frame.data)?; + } + _ => {} + } + } + // The bus plays what is written after a buffer's delay, and drops + // what a closed channel has not played, so it stays open until the + // replay ends. + std::future::pending().await + }; + + // The span ends where it does even when its tracks have not ended yet. + let played = async { + tokio::try_join!(video, audio)?; + std::future::pending().await + }; + // Played first, so a picture due at the very end shows before it does. + tokio::select! { + biased; + result = played => result, + () = kio::clock::sleep_until(at(take.end)) => Ok(()), + } + } } +/// How a replay reads its take: from the first group, where the take's tracks +/// declare they start, skipping nothing as late, since a take is history. +const REPLAY: moq_mux::container::Delay = moq_mux::container::Delay { + min: Duration::ZERO, + max: None, +}; + /// The best enabled rendition the broadcast serves itself, of `renditions` /// ranked best first. /// @@ -596,6 +855,15 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { status: kio::Producer::new(Status::Starting), }); + let played = Playback::default(); + let (plays, replays) = match source { + Source::Replay => { + let (plays, replays) = watch::channel(Next::default()); + (Some(plays), Some(replays)) + } + _ => (None, None), + }; + let feed = Feed { source: source.clone(), picture: picture.clone(), @@ -603,6 +871,8 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { strip: strip.clone(), status: status.consume(), track: track.as_ref().map(|track| track.status.consume()), + played: played.clone(), + plays, }; let driver = Driver { name: name.to_string(), @@ -613,6 +883,8 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { strip, status, track, + played, + plays: replays, }; (feed, driver) } @@ -688,6 +960,76 @@ pub(crate) mod test { panic!("sources never started"); } + /// Three seconds of 30 fps H.264 with a keyframe each second, but for the + /// second `skip` the broadcast paused through, its track open for as long + /// as the returned producer lives, as a live one is. + pub(crate) async fn video( + skip: Option, + ) -> ( + moq_net::broadcast::Producer, + (String, VideoConfig), + moq_video::encode::Producer, + ) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + let rendition = config.probe().await.unwrap(); + let reserved = catalog.reserve().with_hls_hint(config.gop.duration(config.framerate)); + let mut producer = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + let mut encoder = moq_video::encode::Encoder::new(&config).unwrap(); + let gray = vec![0x80; 64 * 64 * 4]; + for index in 0..90u64 { + if skip == Some(index / 30) { + if index % 30 == 0 { + producer.discontinuity().unwrap(); + } + continue; + } + if index % 30 == 0 { + encoder.cut().unwrap(); + } + let surface = moq_video::Surface::rgba(&gray, moq_video::Size::new(64, 64)).unwrap(); + let frame = moq_video::Frame::new( + surface, + moq_net::Timestamp::from_micros(index * 1_000_000 / 30).unwrap(), + ); + producer.publish(&encoder.encode(&frame).unwrap()).unwrap(); + } + let video = catalog.snapshot().video.renditions.into_iter().next().unwrap(); + (broadcast, video, producer) + } + + /// Wait until `f` returns something, without letting a paused clock move. + /// + /// Decoding and probing an encoder run on moq-video's own threads off Apple + /// platforms, which tokio cannot see: a runtime left idle while one works + /// advances a paused clock to its earliest timer, which may lie past whatever + /// the test was waiting for. Keeping a timer due now holds the clock where it + /// is, but only on a whole millisecond: the timer wheel rounds a deadline up + /// to one, so from between two a zero sleep is due at the next, and the idle + /// runtime moves the clock a millisecond per loop. Move it with + /// [`until_time`], which lands on one, never `tokio::time::advance`, which + /// may not. A wait that never ends is a hang nextest kills. + pub(crate) async fn until(mut f: impl FnMut() -> Option) -> T { + loop { + if let Some(value) = f() { + return value; + } + // Through the timer driver, so the replay's due timers fire, but + // due now, so a paused clock has nowhere later to jump to. + #[expect(clippy::disallowed_methods, reason = "a zero sleep that holds the paused clock")] + tokio::time::sleep(Duration::ZERO).await; + } + } + + /// Move a paused clock to `at`, rounded up to its millisecond, and no + /// further: the timer due then is the earliest pending while the runtime + /// waits for it. See [`until`]. + pub(crate) async fn until_time(at: Instant) { + kio::clock::sleep_until(at).await; + } + /// Run `f` on a local task set, as drivers need. pub(crate) async fn local(f: F) -> F::Output { tokio::task::LocalSet::new().run_until(f).await @@ -696,7 +1038,7 @@ pub(crate) mod test { #[cfg(test)] mod tests { - use super::test::{context, local, start}; + use super::test::{context, local, start, until, until_time, video}; use super::*; use crate::collection::{Audio, Rgba}; @@ -1042,4 +1384,98 @@ mod tests { }) .await; } + + /// A replay shows the span's pictures, and only those, at its speed: half + /// speed spreads a second of them over two. Then it shows nothing. + #[tokio::test(start_paused = true)] + async fn a_replay_plays_its_span_at_its_speed() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let mut collection = Collection::new(); + collection.sources.insert("replay".into(), Source::Replay); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let feed = sources.get("replay").unwrap(); + assert!(matches!(feed.take(), Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4])); + + let (broadcast, video, _live) = video(None).await; + let target = Target { + broadcast: broadcast.consume(), + video: Some(video), + audio: None, + end: moq_net::Timestamp::from_micros(2_500_000).unwrap(), + }; + let take = Take::new(&target, Duration::from_secs(1)).unwrap(); + let speed = 0.5; + let begin = kio::clock::now(); + feed.replay(take, speed).unwrap(); + + // From the start, not from the keyframe of its group at 1 s, each + // picture `t / speed` into the replay, and none sooner. The clock + // moves only to each picture's moment, which the timer wheel rounds + // up to its millisecond. + let start = moq_net::Timestamp::from_micros(1_500_000).unwrap(); + for index in 45..75u64 { + let timestamp = moq_net::Timestamp::from_micros(index * 1_000_000 / 30).unwrap(); + let due = begin + (Duration::from(timestamp) - Duration::from(start)).div_f32(speed); + assert!(feed.take().is_none(), "a picture before {timestamp:?} is due"); + until_time(due).await; + let shown = until(|| match feed.take() { + Some(Picture::Frame(frame)) => Some(frame.timestamp), + Some(_) => panic!("cleared before {timestamp:?}"), + None => None, + }) + .await; + assert_eq!(shown, timestamp); + } + + // Nothing more, and the replay ends once the span's second has + // played over two. + assert!(feed.take().is_none(), "the picture at the end shows for no time"); + let end = begin + Duration::from_secs(2); + until_time(end - Duration::from_millis(1)).await; + assert!(feed.take().is_none(), "cleared before the end"); + until_time(end).await; + until(|| matches!(feed.take(), Some(Picture::Rgba { .. })).then_some(())).await; + assert_eq!(feed.status(), Status::Live, "ready for the next replay"); + }) + .await; + } + + /// A new replay cuts off the one playing, whose picture does not linger + /// until the new one's first is due. + #[tokio::test(start_paused = true)] + async fn a_new_replay_clears_the_one_playing() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let mut collection = Collection::new(); + collection.sources.insert("replay".into(), Source::Replay); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let feed = sources.get("replay").unwrap(); + + let (broadcast, video, _live) = video(None).await; + let take = |end| { + let target = Target { + broadcast: broadcast.consume(), + video: Some(video.clone()), + audio: None, + end: moq_net::Timestamp::from_micros(end).unwrap(), + }; + Take::new(&target, Duration::from_secs(1)).unwrap() + }; + // Its first picture, at 1.5 s, is due at once. + feed.replay(take(2_500_000), 1.0).unwrap(); + until(|| matches!(feed.take(), Some(Picture::Frame(_))).then_some(())).await; + // This one's first picture, at the same 1.5 s, is due 20 ms in. + feed.replay(take(2_480_000), 1.0).unwrap(); + let shown = until(|| feed.take()).await; + assert!(matches!(shown, Picture::Rgba { pixels, .. } if pixels == [0; 4])); + }) + .await; + } } diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index ac9c56a..3a2b3ce 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -66,6 +66,8 @@ url = { workspace = true } usage-rs = { workspace = true, features = ["completions", "config"] } [dev-dependencies] +# Reads a saved clip's timeline back; the version moq-mux writes it with. +mp4-atom = "0.17.0" # The auth server a test relay verifies guest tokens with. moq-auth = { workspace = true, features = ["serve"] } moq-relay = { workspace = true, features = ["test-support"] } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index d0e1e51..504b363 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -345,6 +345,10 @@ pub struct Output { /// while held. The Health panel reads their states. _connections: Vec, publication: moq_net::origin::Publication, + /// 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, + clock: moq_mux::Clock, } /// A relay the program is published to. @@ -360,6 +364,39 @@ impl Drop for Output { } } +impl Output { + /// What a replay of the program copies: its video and audio, up to now. + pub fn program(&self) -> moq_scene::replay::Target { + let catalog = self.catalog.snapshot(); + self.target( + catalog.video.renditions.into_iter().next(), + catalog.audio.renditions.into_iter().next(), + ) + } + + /// What a replay of the local source `name` copies: the video of its own + /// track, up to now, or `None` while it publishes no video. + pub fn source(&self, name: &str) -> Option { + let mut catalog = self.catalog.snapshot(); + let listed = catalog.ext.studio.sources.remove(name)?; + let video = listed.video.renditions.into_iter().next()?; + Some(self.target(Some(video), None)) + } + + fn target( + &self, + video: Option<(String, hang::catalog::VideoConfig)>, + audio: Option<(String, hang::catalog::AudioConfig)>, + ) -> moq_scene::replay::Target { + moq_scene::replay::Target { + broadcast: self.publication.broadcast().consume(), + video, + audio, + end: self.clock.now(), + } + } +} + /// The running studio. pub struct Engine { pub state: Mutex, @@ -833,6 +870,7 @@ async fn publish( options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Opus, &input); options.config.bitrate = Some(AUDIO_BITRATE); let graded = catalog.clone(); + let listed = catalog.clone(); let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog, input, &options)?; drop(reserved); @@ -865,6 +903,8 @@ async fn publish( on_demand, _connections: connections, publication, + catalog: listed, + clock: clock.clone(), }; Ok(( output, @@ -1539,6 +1579,15 @@ mod tests { }) .await .expect("health grades the program's tracks"); + + // The program's recent past saves as a clip, from the groups it holds. + let target = engine.target(&crate::replay::Of::Program).expect("the program's past"); + let take = moq_scene::replay::Take::clip(&target, Duration::from_secs(1)).expect("a take"); + let clip = kio::clock::timeout(TIMEOUT, crate::replay::save(take, engine.autosave.path())) + .await + .expect("the clip saves") + .expect("a clip"); + assert_eq!(clip, dir.path().join("show-clip-1.mp4")); } /// With a failover relay, both relays route the program under its one diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 4c649f2..ddadb50 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -14,6 +14,7 @@ mod devices; mod engine; mod guests; mod health; +mod replay; mod setup; mod snap; mod ui; diff --git a/rs/moq-studio/src/replay.rs b/rs/moq-studio/src/replay.rs new file mode 100644 index 0000000..d4e448a --- /dev/null +++ b/rs/moq-studio/src/replay.rs @@ -0,0 +1,219 @@ +//! Instant replays and clips: the recent past of the program or a source, +//! played in a replay source or saved to disk. + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::Context; +use moq_scene::Source; +use moq_scene::replay::{Take, Target}; +use tokio::io::{AsyncSeekExt, AsyncWriteExt}; + +use crate::engine::Engine; + +/// The lengths the Replay panel offers. +pub const LENGTHS: [Duration; 4] = [ + Duration::from_secs(5), + Duration::from_secs(10), + Duration::from_secs(20), + moq_scene::replay::MAX_LENGTH, +]; + +/// The speeds the Replay panel offers: real time, and slow motion. +pub const SPEEDS: [f32; 3] = [1.0, 0.5, 0.25]; + +/// What a replay or clip is of. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum Of { + /// The program, as it was published. + #[default] + Program, + /// A source, by name. + Source(String), +} + +impl std::fmt::Display for Of { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Program => write!(f, "the program"), + Self::Source(name) => write!(f, "`{name}`"), + } + } +} + +impl Engine { + /// What a replay of `of` copies, ending now. + /// + /// The program and the local sources have a past only while they are + /// published, since nothing else encodes them, and a broadcast source once + /// it shows a picture. + pub fn target(&self, of: &Of) -> anyhow::Result { + let output = || { + self.output + .as_ref() + .context("nothing is encoded without --connect, so there is no past to replay") + }; + let name = match of { + Of::Program => return Ok(output()?.program()), + Of::Source(name) => name, + }; + let state = self.state.lock().unwrap(); + let capture = state.collection().capture(name); + match state.collection().sources.get(capture) { + Some(Source::Broadcast { .. }) => state + .sources + .get(capture) + .and_then(|feed| feed.target()) + .with_context(|| format!("`{name}` has shown nothing yet")), + Some(source) if source.has_video() => output()? + .source(capture) + .with_context(|| format!("`{name}` publishes no video to replay")), + Some(_) => anyhow::bail!("`{name}` has no picture to replay"), + None => anyhow::bail!("there is no source named `{name}`"), + } + } +} + +/// Play `take` at `speed` in the replay source `into`. +pub fn play(engine: &Engine, take: Take, speed: f32, into: &str) -> anyhow::Result<()> { + let state = engine.state.lock().unwrap(); + let feed = state + .sources + .get(into) + .with_context(|| format!("the replay source `{into}` is not running"))?; + feed.replay(take, speed)?; + Ok(()) +} + +/// Save `take`, in real time, as a new MP4 file beside `scenes`, the scene +/// file. Returns where. +pub async fn save(take: Take, scenes: &Path) -> anyhow::Result { + let (path, mut file) = create(scenes).await?; + let written = async { + let broadcast = take.consume(); + let catalog = moq_mux::catalog::Consumer::<()>::new(&broadcast, moq_mux::catalog::CatalogFormat::Hang).await?; + // A take refuses renditions served from other broadcasts, so nothing + // resolves through this origin. + let source = moq_mux::Source::new(moq_tokio::origin::spawn().consume(), broadcast); + // A take is history, so nothing in it is late. + let mut export = moq_mux::container::mp4::Export::new(source, catalog).with_max_delay(Duration::MAX); + while let Some(data) = export.next().await? { + file.write_all(&data).await?; + } + let finish = export.finish()?.context("the replay held nothing to save")?; + file.set_len(finish.moov_offset).await?; + file.seek(std::io::SeekFrom::Start(finish.moov_offset)).await?; + file.write_all(&finish.moov).await?; + // The header points readers past the media at the moov, so the moov + // must be on disk first. + file.sync_data().await?; + file.seek(std::io::SeekFrom::Start(finish.mdat_offset)).await?; + file.write_all(&finish.mdat).await?; + file.sync_all().await?; + anyhow::Ok(()) + }; + if let Err(err) = written.await { + // Half a clip is no clip. + drop(file); + let _ = tokio::fs::remove_file(&path).await; + return Err(err.context(format!("cannot save {}", path.display()))); + } + Ok(path) +} + +/// A new file beside the scene file `scenes`: for `show.toml`, +/// `show-clip-1.mp4`, or the first number not taken. +async fn create(scenes: &Path) -> anyhow::Result<(PathBuf, tokio::fs::File)> { + let dir = scenes.parent().unwrap_or(Path::new("")); + let stem = scenes.file_stem().unwrap_or_default().to_string_lossy(); + for n in 1.. { + let path = dir.join(format!("{stem}-clip-{n}.mp4")); + match tokio::fs::File::create_new(&path).await { + Ok(file) => return Ok((path, file)), + Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(err) => return Err(err).with_context(|| format!("cannot create {}", path.display())), + } + } + unreachable!("some clip number is free") +} + +// The software encoder is the one every test machine has. +#[cfg(all(test, feature = "openh264"))] +mod tests { + use std::io::Seek; + + use super::*; + + /// A clip is a finished MP4 of the span, named beside the scene file + /// without overwriting an earlier one, that shows nothing after the span's + /// end even when a picture inside it refers to one shown after. + #[tokio::test] + async fn a_clip_saves_its_span() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + let rendition = config.probe().await.unwrap(); + let reserved = catalog.reserve().with_hls_hint(config.gop.duration(config.framerate)); + let mut producer = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + let mut encoder = moq_video::encode::Encoder::new(&config).unwrap(); + let gray = vec![0x80; 64 * 64 * 4]; + let at = |index: u64| moq_net::Timestamp::from_micros(index * 1_000_000 / 30).unwrap(); + // Each second in the decode order of B-frames: I0 P2 B1 P4 B3 ... B27 P29. + let shown = |index: u64| match index % 30 { + 0 | 29 => index, + odd if odd % 2 == 1 => index + 1, + _ => index - 1, + }; + for index in 0..90u64 { + if index % 30 == 0 { + encoder.cut().unwrap(); + } + let surface = moq_video::Surface::rgba(&gray, moq_video::Size::new(64, 64)).unwrap(); + let mut encoded = encoder.encode(&moq_video::Frame::new(surface, at(index))).unwrap(); + // The software encoder writes each picture as it takes it. + let [unit] = encoded.as_mut_slice() else { + panic!("one access unit per picture"); + }; + unit.timestamp = at(shown(index)); + producer.publish(&encoded).unwrap(); + } + // Between the pictures shown at 2.433 and 2.467 s, which are decoded the + // other way around. + let end = Duration::from_millis(2_450); + let target = Target { + broadcast: broadcast.consume(), + video: catalog.snapshot().video.renditions.into_iter().next(), + audio: None, + end: moq_net::Timestamp::from_micros(end.as_micros() as u64).unwrap(), + }; + + let dir = tempfile::tempdir().unwrap(); + let scenes = dir.path().join("show.toml"); + std::fs::write(dir.path().join("show-clip-1.mp4"), b"an earlier clip").unwrap(); + let take = Take::clip(&target, Duration::from_secs(1)).unwrap(); + let path = save(take, &scenes).await.unwrap(); + assert_eq!(path, dir.path().join("show-clip-2.mp4")); + assert_eq!( + std::fs::read(dir.path().join("show-clip-1.mp4")).unwrap(), + b"an earlier clip" + ); + // A regular MP4 needs no repair, unlike one a crash cut short. + let mut file = std::fs::File::open(&path).unwrap(); + assert!(moq_mux::container::mp4::repair(&mut file).unwrap().is_none()); + + // It shows from the keyframe at 1 s, before the span's start, to its + // end at the latest, and at most a few pictures short of it. + file.rewind().unwrap(); + let moov = std::iter::from_fn(|| mp4_atom::ReadFrom::read_from(&mut file).unwrap()) + .find_map(|atom| match atom { + mp4_atom::Any::Moov(moov) => Some(moov), + _ => None, + }) + .expect("a moov"); + let shows = Duration::from_secs(1) + + Duration::from_secs_f64(moov.mvhd.duration as f64 / f64::from(moov.mvhd.timescale)); + assert!(shows <= end, "shows until {shows:?}, past {end:?}"); + assert!(shows >= end - Duration::from_millis(200), "shows only until {shows:?}"); + } +} diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index bd02ae4..32c42a6 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1,12 +1,13 @@ //! The window: docked panels over the engine's state. -use std::sync::Arc; use std::sync::atomic::Ordering; +use std::sync::{Arc, Mutex}; use std::time::Duration; use anyhow::Context; use egui::{Color32, Ui}; use egui_dock::{DockArea, DockState, NodeIndex, Style, TabViewer}; +use moq_scene::replay::Take; use moq_scene::source::{Feed, Status}; use moq_scene::{ Blend, Canvas, Clip, Collection, Crop, Draft, Filter, Item, Layer, MaskMode, Rect, Scene, Source, Transition, @@ -17,6 +18,7 @@ use crate::autosave::Conflict; use crate::devices::{self, Listing, Pick}; use crate::engine::{Engine, State}; use crate::health::{self, Verdict}; +use crate::replay::{self, Of}; use crate::snap::{self, Grab, Snap}; /// The panels, each a tab the user can move, split, or stack. @@ -34,6 +36,7 @@ enum Tab { Audio, Output, Health, + Replay, } impl Tab { @@ -51,6 +54,7 @@ impl Tab { Self::Audio => "Audio", Self::Output => "Output", Self::Health => "Health", + Self::Replay => "Replay", } } } @@ -80,7 +84,11 @@ impl Studio { tree.split_right(top, 0.5, vec![Tab::Program]); let [_, items] = tree.split_right(bottom, 0.18, vec![Tab::Items]); let [_, sources] = tree.split_right(items, 0.4, vec![Tab::Sources, Tab::Devices, Tab::Guests]); - tree.split_right(sources, 0.5, vec![Tab::Filters, Tab::Health, Tab::Audio, Tab::Output]); + tree.split_right( + sources, + 0.5, + vec![Tab::Filters, Tab::Replay, Tab::Health, Tab::Audio, Tab::Output], + ); Self { dock, @@ -100,6 +108,7 @@ impl Studio { new_item: None, add: AddSource::default(), devices: Devices::default(), + replay: Replay::default(), guest: String::new(), shown: None, message: (dropped > 0).then(|| Err(dropped_message(dropped))), @@ -194,6 +203,7 @@ struct Panels { new_item: Option, add: AddSource, devices: Devices, + replay: Replay, /// The name typed for the next guest. guest: String, /// The join link on screen. @@ -233,6 +243,7 @@ impl TabViewer for Panels { Tab::Audio => self.audio(ui), Tab::Output => self.output(ui), Tab::Health => self.health(ui), + Tab::Replay => self.replay(ui), } } } @@ -1636,6 +1647,138 @@ impl Panels { } } + /// Replay the recent past of the program or a source in a replay source, + /// or save it as a clip. + fn replay(&mut self, ui: &mut Ui) { + let (pictures, replays) = { + let state = self.engine.state.lock().unwrap(); + let named = |kind: fn(&Source) -> bool| -> Vec { + let sources = &state.collection().sources; + sources + .iter() + .filter(|(_, source)| kind(source)) + .map(|(name, _)| name.clone()) + .collect() + }; + ( + named(|source| source.has_video() && !matches!(source, Source::Image { .. } | Source::Color { .. })), + named(|source| matches!(source, Source::Replay)), + ) + }; + let panel = &mut self.replay; + if panel.into.as_ref().is_none_or(|into| !replays.contains(into)) { + panel.into = replays.first().cloned(); + } + + egui::Grid::new("replay").show(ui, |ui| { + ui.label("Replay"); + egui::ComboBox::from_id_salt("replay of") + .selected_text(panel.of.to_string()) + .show_ui(ui, |ui| { + ui.selectable_value(&mut panel.of, Of::Program, Of::Program.to_string()); + for name in pictures { + let of = Of::Source(name); + let label = of.to_string(); + ui.selectable_value(&mut panel.of, of, label); + } + }); + ui.end_row(); + ui.label("Length"); + egui::ComboBox::from_id_salt("replay length") + .selected_text(format!("{} s", panel.length.as_secs())) + .show_ui(ui, |ui| { + for length in replay::LENGTHS { + ui.selectable_value(&mut panel.length, length, format!("{} s", length.as_secs())); + } + }); + ui.end_row(); + ui.label("Speed"); + egui::ComboBox::from_id_salt("replay speed") + .selected_text(format!("{:.0}%", panel.speed * 100.0)) + .show_ui(ui, |ui| { + for speed in replay::SPEEDS { + ui.selectable_value(&mut panel.speed, speed, format!("{:.0}%", speed * 100.0)); + } + }); + ui.end_row(); + ui.label("Play in"); + match &mut panel.into { + Some(into) => { + egui::ComboBox::from_id_salt("replay into") + .selected_text(into.as_str()) + .show_ui(ui, |ui| { + for name in replays { + let label = name.clone(); + ui.selectable_value(into, name, label); + } + }); + } + None => { + ui.weak("no replay source; add one in Sources"); + } + } + ui.end_row(); + }); + + let (play, save) = ui + .horizontal(|ui| { + let play = ui + .add_enabled(panel.into.is_some(), egui::Button::new("Replay")) + .on_hover_text("Play the last moments in the replay source") + .clicked(); + let save = ui + .button("Save clip") + .on_hover_text( + "Save the last moments, in real time, beside the scene file. Copied without re-encoding, a \ + clip starts at the keyframe before them, and video with B-frames can end early.", + ) + .clicked(); + (play, save) + }) + .inner; + if let Some(outcome) = panel.outcome.lock().unwrap().as_ref() { + match outcome { + Ok(message) => ui.label(message), + Err(err) => ui.colored_label(ui.visuals().error_fg_color, err), + }; + } + if !play && !save { + return; + } + + // Copied at the click, without waiting on anything, so the span ends + // there and replays apply in the order they were asked for. + let outcome = panel.outcome.clone(); + let into = panel.into.as_ref().filter(|_| play); + let take = self.engine.target(&panel.of).and_then(|target| { + Ok(match into { + Some(_) => Take::new(&target, panel.length)?, + None => Take::clip(&target, panel.length)?, + }) + }); + let take = match take { + Ok(take) => take, + Err(err) => { + *outcome.lock().unwrap() = Some(Err(format!("{err:#}"))); + return; + } + }; + let of = panel.of.clone(); + if let Some(into) = into { + let result = + replay::play(&self.engine, take, panel.speed, into).map(|()| format!("replaying {of} in `{into}`")); + *outcome.lock().unwrap() = Some(result.map_err(|err| format!("{err:#}"))); + return; + } + let engine = self.engine.clone(); + self.runtime.spawn(async move { + let result = replay::save(take, engine.autosave.path()) + .await + .map(|path| format!("saved {of} to {}", path.display())); + *outcome.lock().unwrap() = Some(result.map_err(|err| format!("{err:#}"))); + }); + } + /// The canvas's size and frame rate, applied at once rather than as each /// field changes, since a new canvas starts a new rendition. fn canvas_form(&mut self, ui: &mut Ui) { @@ -2384,6 +2527,7 @@ fn describe(source: &Source) -> String { Source::Color { color } => format!("color: {color}"), Source::Broadcast { path } => format!("broadcast: {path}"), Source::Clone { source } => format!("clone of {source}"), + Source::Replay => "replay".into(), Source::Lut { path } => format!("LUT: {}", path.display()), } } @@ -2460,11 +2604,12 @@ enum Kind { Window, Microphone, SystemAudio, + Replay, Lut, } impl Kind { - const ALL: [Self; 9] = [ + const ALL: [Self; 10] = [ Self::Color, Self::Image, Self::Broadcast, @@ -2473,6 +2618,7 @@ impl Kind { Self::Window, Self::Microphone, Self::SystemAudio, + Self::Replay, Self::Lut, ]; @@ -2486,6 +2632,7 @@ impl Kind { Self::Window => "Window", Self::Microphone => "Microphone", Self::SystemAudio => "System audio", + Self::Replay => "Replay", Self::Lut => "LUT", } } @@ -2493,7 +2640,7 @@ impl Kind { /// What the value field holds, if this kind takes one. fn value(self) -> Option<&'static str> { match self { - Self::Color | Self::SystemAudio => None, + Self::Color | Self::SystemAudio | Self::Replay => None, Self::Image => Some("file"), Self::Lut => Some(".cube file"), Self::Broadcast => Some("path"), @@ -2551,10 +2698,34 @@ impl AddSource { Kind::Window => Source::Window { id: value }, Kind::Microphone => Source::Microphone { id }, Kind::SystemAudio => Source::SystemAudio, + Kind::Replay => Source::Replay, }) } } +/// The Replay panel's choices, and how the last replay or clip went. +struct Replay { + of: Of, + length: Duration, + speed: f32, + /// The replay source a replay plays in. + into: Option, + /// Set by the replay or save running in the background. + outcome: Arc>>>, +} + +impl Default for Replay { + fn default() -> Self { + Self { + of: Of::Program, + length: replay::LENGTHS[1], + speed: replay::SPEEDS[0], + into: None, + outcome: Arc::default(), + } + } +} + /// The capture devices this machine has, listed in the background. #[derive(Default)] struct Devices {