diff --git a/Cargo.lock b/Cargo.lock index 7e78405..9ec7955 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -117,12 +117,56 @@ dependencies = [ "unicode-width", ] +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + [[package]] name = "anstyle" version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + [[package]] name = "anyhow" version = "1.0.104" @@ -627,6 +671,15 @@ dependencies = [ "bytes", ] +[[package]] +name = "bzip2" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3a53fac24f34a81bc9954b5d6cfce0c21e18ec6959f44f56e8e90e4bb7c346c" +dependencies = [ + "libbz2-rs-sys", +] + [[package]] name = "calloop" version = "0.13.0" @@ -689,6 +742,40 @@ dependencies = [ "shlex 2.0.1", ] +[[package]] +name = "cef" +version = "156.3.0+156.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "630eba7cb50b3cc5ee13ece8c922f36f72ace8a1db8e81b6f694b98c5cf373b1" +dependencies = [ + "ash", + "cef-dll-sys", + "libc", + "libloading 0.9.0", + "objc2 0.6.5", + "objc2-foundation 0.3.2", + "objc2-io-surface", + "objc2-metal 0.3.2", + "thiserror 2.0.21", + "tracing", + "wgpu", + "windows", + "windows-sys 0.61.2", +] + +[[package]] +name = "cef-dll-sys" +version = "156.3.0+156.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02d194589b022eed4f975dc40fc8a6318ea7b42e2ff8b393f880f496790cafc4" +dependencies = [ + "anyhow", + "cc", + "cmake", + "download-cef", + "serde_json", +] + [[package]] name = "cexpr" version = "0.6.0" @@ -763,6 +850,46 @@ dependencies = [ "libloading 0.8.9", ] +[[package]] +name = "clap" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa8876b300ab35ba921adea3dfd70157a46249b33f95c9084ae5709785478946" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0797fb7aeb1406c84efac526901f7ec3ead2124f946b494e72879d4b54704d" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9c751b79415d4e559e3d1fcf128e09e720eb673a06d26cf6f392d37d75b66e0" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 3.0.6", +] + +[[package]] +name = "clap_lex" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" + [[package]] name = "clipboard-win" version = "5.4.3" @@ -816,6 +943,12 @@ dependencies = [ "bytemuck", ] +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + [[package]] name = "combine" version = "4.6.8" @@ -835,6 +968,18 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "console" +version = "0.16.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e96a4956774c13c126a8b5af4daa79384f4d826534c95a02d76afb39e2ab64e3" +dependencies = [ + "encode_unicode", + "libc", + "unicode-width", + "windows-sys 0.61.2", +] + [[package]] name = "const-oid" version = "0.9.6" @@ -856,12 +1001,41 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "cookie" +version = "0.18.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a373e3602691c3cdea496d2f0ee5935151e6168fe87739483c463db1b2f2f87" +dependencies = [ + "percent-encoding", + "time", + "version_check", +] + [[package]] name = "cookie-factory" version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9885fa71e26b8ab7855e2ec7cae6e9b380edff76cd052e07c683a0319d51b3a2" +[[package]] +name = "cookie_store" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15b2c103cf610ec6cae3da84a766285b42fd16aad564758459e6ecf128c75206" +dependencies = [ + "cookie", + "document-features", + "idna", + "indexmap 2.14.2", + "log", + "serde", + "serde_derive", + "serde_json", + "time", + "url", +] + [[package]] name = "core-foundation" version = "0.9.4" @@ -1310,6 +1484,26 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" +[[package]] +name = "download-cef" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05ee509499d14dc47da5f824e4bb1bd558ceb259adc5475956265d48787a80da" +dependencies = [ + "bzip2", + "clap", + "fs-err", + "indicatif", + "regex", + "semver", + "serde", + "serde_json", + "sha1_smol", + "tar", + "thiserror 2.0.21", + "ureq", +] + [[package]] name = "dpi" version = "0.1.2" @@ -1569,6 +1763,12 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" +[[package]] +name = "encode_unicode" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0" + [[package]] name = "endi" version = "1.1.1" @@ -1745,6 +1945,16 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.14" @@ -2653,6 +2863,19 @@ dependencies = [ "serde_core", ] +[[package]] +name = "indicatif" +version = "0.18.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9433806cd6b4ec1aba79c021c7e4c58fb4c3b9977c085062e611ac929998fb0c" +dependencies = [ + "console", + "portable-atomic", + "unicode-width", + "unit-prefix", + "web-time", +] + [[package]] name = "inotify" version = "0.11.5" @@ -2679,6 +2902,12 @@ version = "2.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itertools" version = "0.13.0" @@ -2931,6 +3160,12 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20870f649af7073d53e38067b2a84312175d56ea15217e1b15bc83506ec50afb" +[[package]] +name = "libbz2-rs-sys" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32f454fd6831a29c0e827ed075bca2a15be201bd57f4168f0fd1041124eab17e" + [[package]] name = "libc" version = "0.2.191" @@ -3416,6 +3651,7 @@ version = "0.1.0" dependencies = [ "bytemuck", "bytes", + "cef", "hang", "humantime-serde", "image", @@ -3424,6 +3660,8 @@ dependencies = [ "moq-mux", "moq-net", "moq-video", + "postcard", + "rustix 1.1.5", "serde", "serde_with", "tempfile", @@ -4263,8 +4501,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "180788110936d59bab6bd83b6060ffdfffb3b922ba1396b312ae795e1de9d81d" dependencies = [ "bitflags 2.13.2", + "libc", "objc2 0.6.5", "objc2-core-foundation", + "objc2-foundation 0.3.2", ] [[package]] @@ -4315,6 +4555,7 @@ dependencies = [ "objc2 0.6.5", "objc2-core-foundation", "objc2-foundation 0.3.2", + "objc2-io-surface", ] [[package]] @@ -4472,6 +4713,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "openh264" version = "0.9.9" @@ -5397,6 +5644,7 @@ dependencies = [ "aws-lc-rs", "log", "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -5803,6 +6051,12 @@ dependencies = [ "digest", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + [[package]] name = "sha2" version = "0.11.1" @@ -6240,6 +6494,17 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" +[[package]] +name = "tar" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" +dependencies = [ + "filetime", + "libc", + "xattr", +] + [[package]] name = "target-lexicon" version = "0.13.5" @@ -6695,6 +6960,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "unit-prefix" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81e544489bf3d8ef66c953931f56617f423cd4b5494be343d9b9d3dda037b9a3" + [[package]] name = "unsafe-libopus" version = "0.2.0" @@ -6718,6 +6989,38 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "ureq" +version = "3.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7be76686bdf7198002062423ca0d8cf9b75a27c8e1de804513b97e97b555d9d" +dependencies = [ + "base64 0.23.1", + "cookie_store", + "flate2", + "log", + "percent-encoding", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "ureq-proto", + "utf8-zero", + "webpki-roots", +] + +[[package]] +name = "ureq-proto" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "622dce33c252115466d83f63e9744a18f12b4406aae51a01c15ad8fcf1185eb3" +dependencies = [ + "base64 0.23.1", + "http", + "httparse", + "log", +] + [[package]] name = "url" version = "2.5.8" @@ -6775,12 +7078,24 @@ dependencies = [ "usage-argv", ] +[[package]] +name = "utf8-zero" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" + [[package]] name = "utf8_iter" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + [[package]] name = "uuid" version = "1.28.0" @@ -7813,6 +8128,16 @@ dependencies = [ "time", ] +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix 1.1.5", +] + [[package]] name = "xcursor" version = "0.3.11" diff --git a/README.md b/README.md index e848152..7bba27a 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,8 @@ # MoQ Studio `moq-studio` is a native streaming studio for Linux, macOS, and Windows. It -composes cameras, screens, windows, images, colors, and broadcasts from a relay -into scenes on the GPU, mixes their audio, and publishes the result to a +composes cameras, screens, windows, images, colors, web pages, and broadcasts +from a relay into scenes on the GPU, mixes their audio, and publishes the result to a [MoQ](https://github.com/moq-dev/moq) relay. It is early and not packaged yet, so build it from source. The moq crates come from the `moq` submodule: @@ -31,6 +31,9 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ - `capture` adds camera, display, window, microphone, and system-audio sources. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. +- `cef` adds browser sources, on Linux only for now. The build takes CEF from + `CEF_PATH`, which the dev shell sets, or else downloads it, and the binary + loads `libcef.so` from beside itself or from that `CEF_PATH`. - On Linux, display and window sources capture X11 and XWayland only; a Wayland screen or window cannot be picked yet. - On macOS, the studio is not an app bundle yet, so the camera, microphone, @@ -85,11 +88,11 @@ Every panel is a tab you can drag, split, or stack. | Panel | Use it to | | --- | --- | -| Preview, Program | See the scene you are editing and the one that is live, and drag items in the preview to place them. **Cut** puts the preview on program at once; **Fade** blends over the fade time set beside it. | +| Preview, Program | See the scene you are editing and the one that is live, and drag items in the preview to place them. **Cut** puts the preview on program at once; **Fade** blends over the fade time set beside it. **Interact** passes the mouse over the preview to the browser source under it instead, and the keyboard to the one last clicked. | | Scenes | Pick the scene in preview, and add, rename (double-click), reorder, or remove one. A scene on program, or still fading out, cannot be removed, even by an undo, until program cuts away from it. | | History | See every edit, and undo or redo back to any of them. | | Items | Place, show, hide, and order what the preview scene or the top layer draws, back to front. Select an item to filter or copy it. | -| Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. | +| Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. A browser source shows whether its page is copied on the GPU or, as a fallback, uploaded from memory (CPU), in place of a track. | | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Guests | Invite a guest, show their join link as a QR code, see who joined, and take a guest out of every scene. | | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | @@ -135,6 +138,15 @@ best video and audio its catalog lists, switches when the catalog changes, and comes back on its own when the publisher restarts. Audio from every source is mixed into the program whether or not its scene is live; mute a source to leave it out. +A browser source renders a web page, such as an overlay, alert box, or chat +widget, at the canvas frame rate (CEF caps it at 60) over a transparent +background, and its sound joins the program audio. Each page runs in a +process of its own, so a page or browser crash fails that source, with the +reason, while the stream carries on; edit or re-add the source to restart it. +Every start is a fresh profile: no cookies or logins persist. A page opens +no other windows (`window.open` and new-tab links do nothing), and its +dropdown lists, such as a `, + /// The canvas's frame rate, which the page paints at. + #[cfg(target_os = "linux")] + framerate: u32, + /// Whether the helper has yet to hear `framerate`. + #[cfg(target_os = "linux")] + retime: bool, + paint: Option, +} + +#[cfg(target_os = "linux")] +impl PageState { + /// Queue `input` behind any owed releases. Past the backlog, a release is + /// owed rather than dropped, so a helper that catches up never leaves the + /// page holding a button or key the studio let go of; other input drops. + fn send(&mut self, input: Input) { + use std::sync::mpsc::TrySendError; + + self.flush(); + let Some(commands) = &self.commands else { return }; + let input = match self.owed.is_empty() { + true => match commands.try_send(protocol::Command::Input(input)) { + Ok(()) => { + self.backlogged = false; + return; + } + Err(TrySendError::Full(protocol::Command::Input(input))) => input, + Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), + Err(TrySendError::Disconnected(_)) => return self.disconnect(), + }, + // Newer input waits behind the owed releases. + false => input, + }; + if let Some(release) = Release::of(&input) { + self.owed.retain(|owed| Release::of(owed) != Some(release)); + self.owed.push_back(input); + } + if !std::mem::replace(&mut self.backlogged, true) { + tracing::warn!("a browser source's helper is not reading input; dropping it"); + } + } + + /// Queue a new frame rate, then the owed releases, oldest first, while the + /// queue has room. + fn flush(&mut self) { + use std::sync::mpsc::TrySendError; + + let Some(commands) = &self.commands else { return }; + if self.retime { + match commands.try_send(protocol::Command::FrameRate(self.framerate)) { + Ok(()) => self.retime = false, + Err(TrySendError::Full(_)) => return, + Err(TrySendError::Disconnected(_)) => return self.disconnect(), + } + } + while let Some(input) = self.owed.pop_front() { + match commands.try_send(protocol::Command::Input(input)) { + Ok(()) => {} + Err(TrySendError::Full(protocol::Command::Input(input))) => { + self.owed.push_front(input); + return; + } + Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), + Err(TrySendError::Disconnected(_)) => return self.disconnect(), + } + } + } + + /// The helper is gone, which its source reports. + fn disconnect(&mut self) { + self.commands = None; + self.owed.clear(); + } +} + +/// What an input lets go of, which a later release of the same supersedes. +#[cfg(target_os = "linux")] +#[derive(Clone, Copy, PartialEq, Eq)] +enum Release { + Button(Button), + Key(i32), + Focus, + Pointer, +} + +#[cfg(target_os = "linux")] +impl Release { + fn of(input: &Input) -> Option { + match *input { + Input::Button { + button, down: false, .. + } => Some(Self::Button(button)), + Input::Key { code, down: false, .. } => Some(Self::Key(code)), + Input::Focus(false) => Some(Self::Focus), + Input::Leave { .. } => Some(Self::Pointer), + _ => None, + } + } +} + +impl Page { + /// Whether `other` is this page, rather than one restarted under the same + /// name. + pub fn same(&self, other: &Page) -> bool { + Arc::ptr_eq(&self.state, &other.state) + } + + /// How the newest picture arrived, once one has. + pub fn paint(&self) -> Option { + self.state.lock().unwrap().paint + } + + /// Paint at `rate`, the canvas's frame rate, from now on, without waiting on + /// the helper. One that is behind hears it once it catches up. + pub(crate) fn set_framerate(&self, rate: u32) { + #[cfg(target_os = "linux")] + { + let mut state = self.state.lock().unwrap(); + if state.framerate != rate { + state.framerate = rate; + state.retime = true; + state.flush(); + } + } + #[cfg(not(target_os = "linux"))] + let _ = rate; + } + + /// Interact with the page, without waiting on its helper. Dropped while + /// the page is starting or after it stopped, as a click on a window that is + /// not there yet would be, and while the helper is too far behind to take + /// more, as a hung window's would be, except that a release is delivered + /// once the helper catches up. + pub fn send(&self, input: Input) { + #[cfg(target_os = "linux")] + self.state.lock().unwrap().send(input); + #[cfg(not(target_os = "linux"))] + let _ = input; + } +} + +/// Interaction with a page. Positions are in page pixels from its top-left +/// corner. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub enum Input { + /// The pointer moved over the page. + Move { x: i32, y: i32, modifiers: Modifiers }, + /// The pointer left the page. + Leave { modifiers: Modifiers }, + /// A mouse button went down or up; `clicks` counts a double or triple + /// click. + Button { + x: i32, + y: i32, + button: Button, + down: bool, + clicks: u32, + modifiers: Modifiers, + }, + /// A scroll, in pixels. + Wheel { + x: i32, + y: i32, + dx: i32, + dy: i32, + modifiers: Modifiers, + }, + /// A key went down or up. `code` is the Windows virtual-key code, which is + /// what CEF takes on every platform. + Key { + code: i32, + down: bool, + modifiers: Modifiers, + }, + /// Text typed, after the keyboard layout and any input method. + Text(String), + /// Copy the page's selection to the clipboard. + Copy, + /// Cut the page's selection to the clipboard. + Cut, + /// The page gained or lost keyboard focus. + Focus(bool), +} + +/// A mouse button. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Button { + Left, + Middle, + Right, +} + +/// The keys and buttons held during an [`Input`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Modifiers { + pub shift: bool, + pub ctrl: bool, + pub alt: bool, + /// Command on macOS, the Windows key elsewhere. + pub meta: bool, + pub left: bool, + pub middle: bool, + pub right: bool, +} + +/// The texture a page's pixels are uploaded to: premultiplied BGRA, as CEF +/// paints. +#[cfg_attr( + not(target_os = "linux"), + expect(dead_code, reason = "browser sources are Linux-only") +)] +fn upload( + device: &wgpu::Device, + queue: &wgpu::Queue, + width: u32, + height: u32, + data: &[u8], +) -> Result { + let limit = device.limits().max_texture_dimension_2d; + if width == 0 || height == 0 || width > limit || height > limit { + return Err(format!( + "a {width}x{height} page does not fit this GPU's {limit}px texture limit" + )); + } + if data.len() != width as usize * height as usize * 4 { + return Err(format!("{} bytes is not a {width}x{height} picture", data.len())); + } + let size = wgpu::Extent3d { + width, + height, + depth_or_array_layers: 1, + }; + let texture = device.create_texture(&wgpu::TextureDescriptor { + label: Some("moq-scene page"), + size, + mip_level_count: 1, + sample_count: 1, + dimension: wgpu::TextureDimension::D2, + format: wgpu::TextureFormat::Bgra8Unorm, + usage: wgpu::TextureUsages::TEXTURE_BINDING | wgpu::TextureUsages::COPY_DST, + view_formats: &[], + }); + queue.write_texture( + texture.as_image_copy(), + data, + wgpu::TexelCopyBufferLayout { + offset: 0, + bytes_per_row: Some(width * 4), + rows_per_image: Some(height), + }, + size, + ); + Ok(texture) +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + + fn button(down: bool) -> Input { + Input::Button { + x: 1, + y: 1, + button: Button::Left, + down, + clicks: 1, + modifiers: Modifiers::default(), + } + } + + /// Past the backlog, a release is owed rather than dropped, once per + /// button, and goes ahead of newer input as the queue makes room, so the + /// page never keeps holding what the studio let go of. + #[test] + fn a_release_past_the_backlog_is_owed() { + let (sender, queue) = std::sync::mpsc::sync_channel(4); + let page = Page::default(); + page.state.lock().unwrap().commands = Some(sender); + page.send(button(true)); + while !page.state.lock().unwrap().backlogged { + page.send(Input::Text("x".into())); + } + page.send(button(false)); + page.send(button(true)); + page.send(button(false)); + page.send(Input::Move { + x: 2, + y: 2, + modifiers: Modifiers::default(), + }); + assert_eq!(page.state.lock().unwrap().owed, [button(false)]); + + // The helper catches up, and each write makes room, as the writer's do. + let mut received = Vec::new(); + while let Ok(protocol::Command::Input(input)) = queue.try_recv() { + received.push(input); + page.state.lock().unwrap().flush(); + } + assert_eq!(received.first(), Some(&button(true))); + assert_eq!(received.last(), Some(&button(false))); + assert_eq!(received.iter().filter(|input| **input == button(false)).count(), 1); + assert!(!received.iter().any(|input| matches!(input, Input::Move { .. }))); + } + + /// A new canvas frame rate reaches the page once, even past the backlog. + #[test] + fn a_new_frame_rate_reaches_the_page() { + let (sender, queue) = std::sync::mpsc::sync_channel(1); + let page = Page::default(); + page.state.lock().unwrap().commands = Some(sender); + page.set_framerate(30); + page.set_framerate(30); + assert!(matches!(queue.try_recv(), Ok(protocol::Command::FrameRate(30)))); + assert!(queue.try_recv().is_err(), "an unchanged rate is not sent again"); + + page.send(Input::Text("x".into())); + page.set_framerate(60); + assert!(matches!(queue.try_recv(), Ok(protocol::Command::Input(_)))); + page.state.lock().unwrap().flush(); + assert!(matches!(queue.try_recv(), Ok(protocol::Command::FrameRate(60)))); + } +} diff --git a/rs/moq-scene/src/browser/protocol.rs b/rs/moq-scene/src/browser/protocol.rs new file mode 100644 index 0000000..c84f52f --- /dev/null +++ b/rs/moq-scene/src/browser/protocol.rs @@ -0,0 +1,292 @@ +//! What a browser source says to its helper process, and back. +//! +//! One Unix stream socket carries both directions, each message a +//! little-endian `u32` length and that many bytes of postcard. A shared +//! texture's DMA-BUF descriptors ride with the first byte of its message as +//! `SCM_RIGHTS`, so they arrive with the message that describes them. + +use std::io::{self, IoSlice, IoSliceMut, Read, Write}; +use std::mem::MaybeUninit; +use std::os::fd::{AsFd, BorrowedFd, OwnedFd}; +use std::os::unix::net::UnixStream; + +use rustix::net::{ + RecvAncillaryBuffer, RecvAncillaryMessage, RecvFlags, SendAncillaryBuffer, SendAncillaryMessage, SendFlags, +}; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +use super::Input; + +/// The most planes a shared texture has, and so the most descriptors one +/// message carries: CEF's own limit. +pub(crate) const MAX_PLANES: usize = 4; + +/// The longest message either side accepts: the largest page's pixels, with +/// room for the rest of the message. Bounds what a misbehaving peer can make +/// the other allocate. +const MAX_MESSAGE: usize = (super::MAX_SIDE * super::MAX_SIDE * 4) as usize + 1024; + +/// What the studio asks of the helper. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub(crate) enum Command { + /// The first message: the page to open and how to deliver it. + Open(Open), + /// The last picture is copied, so the helper may let CEF reuse it. + Painted, + /// Interaction with the page. + Input(Input), + /// The canvas's frame rate changed, so the page paints at this one. + FrameRate(u32), +} + +/// The page a helper opens. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub(crate) struct Open { + pub url: String, + /// The page's viewport, in pixels. + pub width: u32, + pub height: u32, + /// How often the page paints: the canvas's frame rate. + pub framerate: u32, + /// Whether to paint into shared textures the studio imports, rather than + /// into memory it uploads. + pub accelerated: bool, + /// The DRM render node to paint on: the studio's GPU. + pub render_node: Option, + /// An empty directory for CEF's profile, since two CEF instances cannot + /// share one. The studio removes it once the helper has exited. + pub profile: String, + /// The sound the page plays is delivered at the program bus's rate and + /// channel count, so it joins the bus without resampling. + pub sample_rate: u32, + pub channels: u32, +} + +/// What the helper tells the studio. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub(crate) enum Event { + /// The page painted into a shared texture, whose plane descriptors ride + /// with this message. Valid until the studio answers + /// [`Command::Painted`]. + Texture(Texture), + /// The page painted into memory: `width * height` premultiplied BGRA + /// pixels, tightly packed. + Pixels { width: u32, height: u32, data: Vec }, + /// Interleaved samples at the [`Open`] rate and channel count. + Audio(Vec), + /// The page cannot be shown, and why. + Failed(String), +} + +/// A shared texture's layout, CEF's `cef_accelerated_paint_info_t` minus the +/// descriptors. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub(crate) struct Texture { + pub width: u32, + pub height: u32, + pub format: Format, + /// The DRM format modifier: how the planes are tiled. + pub modifier: u64, + pub planes: Vec, +} + +/// The pixel layouts CEF paints. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) enum Format { + Rgba, + Bgra, +} + +/// One plane of a shared texture. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct Plane { + pub stride: u32, + pub offset: u64, + pub size: u64, +} + +/// One end of the socket. +/// +/// Sending takes `&self`, so a caller sending from several threads holds a +/// lock around each [`send`](Self::send); receiving is for one thread. +pub(crate) struct Channel { + socket: UnixStream, +} + +impl Channel { + pub fn new(socket: UnixStream) -> Self { + Self { socket } + } + + /// Another handle on the same socket, so one thread can send while + /// another receives. + pub fn try_clone(&self) -> io::Result { + Ok(Self::new(self.socket.try_clone()?)) + } + + /// Send `message` with `fds` attached. + pub fn send(&self, message: &T, fds: &[BorrowedFd<'_>]) -> io::Result<()> { + assert!(fds.len() <= MAX_PLANES, "more descriptors than a message carries"); + let payload = postcard::to_stdvec(message).map_err(io::Error::other)?; + if payload.len() > MAX_MESSAGE { + return Err(io::Error::new(io::ErrorKind::InvalidInput, "message too long")); + } + let mut bytes = Vec::with_capacity(4 + payload.len()); + bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + bytes.extend_from_slice(&payload); + + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(MAX_PLANES))]; + let mut control = SendAncillaryBuffer::new(&mut space); + if !fds.is_empty() { + let pushed = control.push(SendAncillaryMessage::ScmRights(fds)); + assert!(pushed, "the control buffer holds every descriptor"); + } + // The descriptors go with the first byte sent; a short send leaves the + // rest for a plain write. + let sent = rustix::net::sendmsg(&self.socket, &[IoSlice::new(&bytes)], &mut control, SendFlags::NOSIGNAL)?; + (&self.socket).write_all(&bytes[sent..]) + } + + /// The next message and the descriptors that came with it, or `None` once + /// the peer has closed the socket. + pub fn recv(&self) -> io::Result)>> { + let mut length = [0u8; 4]; + let mut read = 0; + let mut fds = Vec::new(); + while read < length.len() { + let mut space = [MaybeUninit::uninit(); rustix::cmsg_space!(ScmRights(MAX_PLANES))]; + let mut control = RecvAncillaryBuffer::new(&mut space); + let received = rustix::net::recvmsg( + &self.socket, + &mut [IoSliceMut::new(&mut length[read..])], + &mut control, + RecvFlags::CMSG_CLOEXEC, + )?; + for message in control.drain() { + if let RecvAncillaryMessage::ScmRights(received) = message { + fds.extend(received); + } + } + if received.flags.contains(rustix::net::ReturnFlags::CTRUNC) { + return Err(invalid("more descriptors than a message carries")); + } + match received.bytes { + 0 if read == 0 => return Ok(None), + 0 => return Err(io::ErrorKind::UnexpectedEof.into()), + bytes => read += bytes, + } + } + + let length = u32::from_le_bytes(length) as usize; + if length > MAX_MESSAGE { + return Err(invalid("message too long")); + } + let mut payload = vec![0u8; length]; + (&self.socket).read_exact(&mut payload)?; + let message = postcard::from_bytes(&payload).map_err(invalid)?; + Ok(Some((message, fds))) + } +} + +impl AsFd for Channel { + fn as_fd(&self) -> BorrowedFd<'_> { + self.socket.as_fd() + } +} + +fn invalid(message: impl std::fmt::Display) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pair() -> (Channel, Channel) { + let (a, b) = UnixStream::pair().unwrap(); + (Channel::new(a), Channel::new(b)) + } + + /// Descriptors arrive with the message they were sent with, not the one + /// before or after, however the stream splits. + #[test] + fn descriptors_ride_with_their_message() { + let (studio, helper) = pair(); + let texture = Texture { + width: 1280, + height: 720, + format: Format::Bgra, + modifier: 0, + planes: vec![Plane { + stride: 1280 * 4, + offset: 0, + size: 1280 * 720 * 4, + }], + }; + let file = tempfile::tempfile().unwrap(); + helper.send(&Event::Audio(vec![0.5; 3]), &[]).unwrap(); + helper.send(&Event::Texture(texture.clone()), &[file.as_fd()]).unwrap(); + helper.send(&Event::Failed("gone".into()), &[]).unwrap(); + drop(helper); + + let (event, fds) = studio.recv::().unwrap().unwrap(); + assert_eq!(event, Event::Audio(vec![0.5; 3])); + assert!(fds.is_empty()); + + let (event, fds) = studio.recv::().unwrap().unwrap(); + assert_eq!(event, Event::Texture(texture)); + assert_eq!(fds.len(), 1); + let ours = rustix::fs::fstat(&fds[0]).unwrap(); + let theirs = rustix::fs::fstat(&file).unwrap(); + assert_eq!( + (ours.st_dev, ours.st_ino), + (theirs.st_dev, theirs.st_ino), + "the same file" + ); + + let (event, fds) = studio.recv::().unwrap().unwrap(); + assert_eq!(event, Event::Failed("gone".into())); + assert!(fds.is_empty()); + + assert!(studio.recv::().unwrap().is_none(), "closed"); + } + + /// A message larger than the socket buffer arrives whole. + #[test] + fn a_large_message_arrives_whole() { + let (studio, helper) = pair(); + let data: Vec = (0..1920 * 1080 * 4).map(|i| i as u8).collect(); + let sent = Event::Pixels { + width: 1920, + height: 1080, + data, + }; + let sender = { + let sent = sent.clone(); + std::thread::spawn(move || helper.send(&sent, &[]).unwrap()) + }; + let (received, _) = studio.recv::().unwrap().unwrap(); + sender.join().unwrap(); + assert_eq!(received, sent); + } + + /// A peer cannot make the other side allocate past the bound. + #[test] + fn refuses_an_oversized_length() { + let (studio, helper) = pair(); + (&helper.socket).write_all(&u32::MAX.to_le_bytes()).unwrap(); + let err = studio.recv::().unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + /// A socket closed partway through a message is an error, not a clean end. + #[test] + fn a_truncated_message_is_an_error() { + let (studio, helper) = pair(); + (&helper.socket).write_all(&[8, 0]).unwrap(); + drop(helper); + let err = studio.recv::().unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); + } +} diff --git a/rs/moq-scene/src/browser/run.rs b/rs/moq-scene/src/browser/run.rs new file mode 100644 index 0000000..175083b --- /dev/null +++ b/rs/moq-scene/src/browser/run.rs @@ -0,0 +1,319 @@ +//! The studio's side of a browser source: start its helper, then deliver what +//! the helper paints and plays until the source stops. + +use std::os::fd::OwnedFd; +use std::os::unix::net::UnixStream; +use std::os::unix::process::CommandExt; +use std::process::{Child, ExitStatus, Stdio}; +use std::sync::mpsc; + +use moq_audio::bus::Channel as Sound; + +use super::protocol::{Channel, Command, Event, Open}; +use super::{ARG, Helper, Page, Paint}; +use crate::Error; +use crate::source::{Mailbox, Picture, Status}; + +/// Commands waiting for the helper to read them. A helper this far behind is +/// hung, and [`Page::send`] drops new input rather than wait for it. +const BACKLOG: usize = 256; + +/// Run the page `open` describes until its helper stops, delivering pictures +/// to `picture` and sound to `sound`. Dropping the future stops the helper. +/// +/// Fills in how the helper paints and the page's frame rate: `open` names +/// only the page and the bus's format. +pub(crate) async fn run( + helper: &Helper, + mut open: Open, + page: &Page, + picture: &Mailbox, + sound: Option, + status: &kio::Producer, +) -> Result<(), Error> { + // Declared first so it is removed last, once the helper is gone. + let profile = tempfile::tempdir().map_err(|err| Error::Browser(format!("cannot create a profile: {err}")))?; + open.profile = profile.path().to_string_lossy().into_owned(); + // A later change waits for the writer, which sends it after `open`. + open.framerate = { + let mut state = page.state.lock().unwrap(); + state.retime = false; + state.framerate + }; + open.accelerated = super::import::supported(helper); + open.render_node = helper + .render_node + .as_ref() + .and_then(|node| node.to_str()) + .map(str::to_string); + let (mut process, events) = + spawn(helper).map_err(|err| Error::Browser(format!("cannot start {}: {err}", helper.program.display())))?; + let commands = events + .try_clone() + .map_err(|err| Error::Browser(format!("socket: {err}")))?; + let end = match commands + .send(&Command::Open(open), &[]) + .map(|()| writer(commands, page)) + { + Ok(Ok(commands)) => { + { + let mut state = page.state.lock().unwrap(); + state.commands = Some(commands.clone()); + state.flush(); + } + let end = reader(helper, events, commands, page, picture, sound, status).await; + page.state.lock().unwrap().commands = None; + end + } + Ok(Err(err)) => End::Failed(format!("cannot start the page's writer: {err}")), + Err(err) => closed(err), + }; + // A stopped page leaves the canvas rather than freezing on its last + // picture, which for an alert or a ticker would show stale news. + *picture.lock().unwrap() = Some(Picture::Rgba { + size: moq_video::Size::new(1, 1), + pixels: vec![0; 4], + }); + match end { + // The helper is exiting or gone, so its status says why. + End::Closed => { + let exit = process + .exited() + .map_err(|err| Error::Browser(format!("the browser process: {err}")))?; + Err(Error::Browser(format!("the browser process exited: {exit}"))) + } + End::Failed(reason) => Err(Error::Browser(reason)), + } +} + +/// Why a page stopped. +#[derive(Clone)] +enum End { + /// The helper closed the socket: it exited, or crashed. + Closed, + /// The helper reported a failure, or broke the protocol. + Failed(String), +} + +/// What a socket error says about the helper. +fn closed(err: std::io::Error) -> End { + use std::io::ErrorKind::*; + match err.kind() { + BrokenPipe | ConnectionReset | UnexpectedEof => End::Closed, + _ => End::Failed(format!("the browser process's socket: {err}")), + } +} + +/// The helper process, killed with every CEF process it started when +/// dropped. +/// +/// The helper leads its own process group, which CEF's subprocesses join, so +/// killing the group leaves none of them behind. The group is killed before +/// the helper is reaped, while its id cannot belong to anything else. +struct Process { + child: Child, + exit: Option, +} + +impl Process { + /// Wait for the helper to exit on its own, then stop the rest of its + /// group, returning how it exited. + fn exited(&mut self) -> std::io::Result { + let pid = rustix::process::Pid::from_child(&self.child); + let options = rustix::process::WaitIdOptions::EXITED | rustix::process::WaitIdOptions::NOWAIT; + rustix::process::waitid(rustix::process::WaitId::Pid(pid), options)?; + self.stop() + } + + /// Kill the helper's group and reap the helper. + fn stop(&mut self) -> std::io::Result { + if let Some(exit) = self.exit { + return Ok(exit); + } + let group = rustix::process::Pid::from_child(&self.child); + let _ = rustix::process::kill_process_group(group, rustix::process::Signal::KILL); + let exit = self.child.wait()?; + self.exit = Some(exit); + Ok(exit) + } +} + +impl Drop for Process { + fn drop(&mut self) { + let _ = self.stop(); + } +} + +/// Write the commands queued on the returned sender to `channel`, on a thread +/// of its own, so a helper that stops reading blocks only that thread, never a +/// sender holding the studio's state. Each write makes room for the releases +/// `page` owes. The thread ends once every sender is dropped or the socket +/// fails, as it does when the helper is killed. +fn writer(channel: Channel, page: &Page) -> std::io::Result> { + let (sender, queue) = mpsc::sync_channel(BACKLOG); + // Weak, since the page holds a sender, which would keep the thread waiting. + let page = std::sync::Arc::downgrade(&page.state); + std::thread::Builder::new() + .name("browser-commands".into()) + .spawn(move || { + for command in queue { + if channel.send(&command, &[]).is_err() { + break; + } + if let Some(page) = page.upgrade() { + page.lock().unwrap().flush(); + } + } + })?; + Ok(sender) +} + +/// Run [`deliver`] on a thread of its own, since it blocks on the socket, and +/// wait for why it ended. A thread that cannot start, or that stops without +/// saying why, fails the page with the reason rather than leaving it live. +async fn reader( + helper: &Helper, + events: Channel, + commands: mpsc::SyncSender, + page: &Page, + picture: &Mailbox, + sound: Option, + status: &kio::Producer, +) -> End { + let done = kio::Producer::new(None); + let ended = done.consume(); + let spawned = { + let (helper, page, picture, status) = (helper.clone(), page.clone(), picture.clone(), status.clone()); + std::thread::Builder::new() + .name("browser-events".into()) + .spawn(move || { + let end = deliver(&helper, &events, &commands, &page, &picture, sound, &status); + if let Ok(mut done) = done.write() { + *done = Some(end); + } + }) + }; + if let Err(err) = spawned { + return End::Failed(format!("cannot start the page's reader: {err}")); + } + let end = ended + .wait(|end: &kio::Ref<'_, Option>| match end.as_ref() { + Some(end) => std::task::Poll::Ready(end.clone()), + None => std::task::Poll::Pending, + }) + .await; + end.unwrap_or_else(|_| End::Failed("the page's reader stopped".into())) +} + +/// Start a helper, handing it one end of a socket as its stdin. +fn spawn(helper: &Helper) -> std::io::Result<(Process, Channel)> { + let (ours, theirs) = UnixStream::pair()?; + let child = std::process::Command::new(&helper.program) + .arg(ARG) + .stdin(OwnedFd::from(theirs)) + .stdout(Stdio::null()) + .process_group(0) + .spawn()?; + Ok((Process { child, exit: None }, Channel::new(ours))) +} + +/// Read the helper's events until it closes the socket or reports a failure. +fn deliver( + helper: &Helper, + events: &Channel, + commands: &mpsc::SyncSender, + page: &Page, + picture: &Mailbox, + mut sound: Option, + status: &kio::Producer, +) -> End { + let mut live = false; + loop { + let (event, fds) = match events.recv::() { + Ok(Some(received)) => received, + Ok(None) => return End::Closed, + Err(err) => return closed(err), + }; + let (texture, paint) = match event { + Event::Texture(texture) => (super::import::texture(helper, &texture, fds), Paint::Gpu), + Event::Pixels { width, height, data } => ( + super::upload(&helper.device, &helper.queue, width, height, &data), + Paint::Cpu, + ), + Event::Audio(samples) => { + if let Some(sound) = &mut sound + && let Err(err) = sound.write(bytemuck::cast_slice(&samples)) + { + return End::Failed(format!("the page's sound: {err}")); + } + continue; + } + Event::Failed(reason) => return End::Failed(reason), + }; + // The copy is done, so the helper may hand the source back to CEF. + // Waits behind queued input, never dropped: the helper waits on it. A + // closed queue means the writer lost the socket, so the helper is gone. + if commands.send(Command::Painted).is_err() { + return End::Closed; + } + let texture = match texture { + Ok(texture) => texture, + Err(reason) => return End::Failed(reason), + }; + page.state.lock().unwrap().paint = Some(paint); + *picture.lock().unwrap() = Some(Picture::Texture(texture)); + if !std::mem::replace(&mut live, true) + && let Ok(mut status) = status.write() + { + *status = Status::Live; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::browser::{Input, Modifiers}; + + /// A helper that stops reading costs dropped input, never a sender stuck + /// on a full socket while it holds the studio's state. + #[test] + fn a_helper_that_stops_reading_never_blocks_input() { + let (ours, theirs) = UnixStream::pair().unwrap(); + let page = Page::default(); + page.state.lock().unwrap().commands = Some(writer(Channel::new(ours), &page).unwrap()); + // Far more than the socket buffers and the backlog hold together. + for _ in 0..4096 { + page.send(Input::Text("x".repeat(1024))); + } + assert!(page.state.lock().unwrap().backlogged); + drop(theirs); + } + + /// Each write makes room for an owed release, so it reaches the page once + /// the helper reads again, with no newer input to carry it. + #[test] + fn the_writer_delivers_owed_releases() { + let (ours, theirs) = UnixStream::pair().unwrap(); + let page = Page::default(); + let commands = writer(Channel::new(ours), &page).unwrap(); + let release = Input::Key { + code: 65, + down: false, + modifiers: Modifiers::default(), + }; + { + let mut state = page.state.lock().unwrap(); + state.commands = Some(commands.clone()); + state.owed.push_back(release.clone()); + } + commands.send(Command::Painted).unwrap(); + + let helper = Channel::new(theirs); + assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); + let Some((Command::Input(input), _)) = helper.recv().unwrap() else { + panic!("the owed release never came"); + }; + assert_eq!(input, release); + } +} diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index a8e98ca..9ab7c88 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -130,6 +130,13 @@ impl Collection { source.unwrap_or_default().iter().chain(&item.filters) } + /// What `item` shows of its source's picture: its source's crop, then its + /// own. + pub fn crop(&self, item: &Item) -> Crop { + let source = self.video.get(&item.source).map(|video| video.crop); + source.unwrap_or_default().then(&item.crop) + } + /// Check what the types cannot: the canvas can be encoded, names are unique, /// every item, clone, audio, and video entry names a source that exists, /// and every crop and filter's settings are in range, each filter naming a @@ -177,6 +184,7 @@ impl Collection { return Err(Error::Invalid(format!("`{name}` names no file"))); } for (name, source) in &self.sources { + source.validate(name)?; let Source::Clone { source } = source else { continue }; match self.sources.get(source) { Some(Source::Clone { .. }) => { @@ -827,6 +835,18 @@ pub enum Source { /// The broadcast path on that relay. path: String, }, + /// A web page, such as an overlay, alert box, or chat widget, with its + /// sound. Desktop only: viewers cannot render one, so it reaches them + /// only as pixels. + Browser { + /// The page's address. + url: String, + /// The page's viewport width in pixels, independent of where it is + /// placed. + width: u32, + /// The page's viewport height in pixels. + height: u32, + }, /// Another source's picture, captured once, with [`Collection::video`] /// filters of its own. Its sound is not mixed a second time. Clone { @@ -877,9 +897,24 @@ impl Source { pub fn has_audio(&self) -> bool { matches!( self, - Self::Microphone { .. } | Self::SystemAudio | Self::Broadcast { .. } | Self::Replay + Self::Microphone { .. } | Self::SystemAudio | Self::Broadcast { .. } | Self::Browser { .. } | Self::Replay ) } + + fn validate(&self, name: &str) -> Result<(), Error> { + if let Self::Browser { url, width, height } = self { + if url.trim().is_empty() { + return Err(Error::Invalid(format!("browser source `{name}` has no url"))); + } + let max = crate::browser::MAX_SIDE; + if *width == 0 || *height == 0 || *width > max || *height > max { + return Err(Error::Invalid(format!( + "browser source `{name}` is {width}x{height}; each side must be 1 to {max} pixels" + ))); + } + } + Ok(()) + } } /// The gain and mute of one source in the program audio. @@ -966,6 +1001,14 @@ mod tests { path: "studio/guest".into(), }, ); + collection.sources.insert( + "chat".into(), + Source::Browser { + url: "https://example.com/chat".into(), + width: 400, + height: 800, + }, + ); collection.audio.insert("mic".into(), Audio { gain: 0.5, muted: true }); collection.sources.insert( "camera faded".into(), @@ -1357,6 +1400,35 @@ rect = { x = 0, y = 680, width = 1280, height = 40 } assert!(matches!(collection.validate(), Err(Error::Invalid(_)))); } + /// A page needs an address and a size the helper can paint and the socket + /// can carry. + #[test] + fn refuses_a_degenerate_page() { + let page = |url: &str, width, height| { + let mut collection = example(); + collection.sources.insert( + "chat".into(), + Source::Browser { + url: url.into(), + width, + height, + }, + ); + collection.validate() + }; + assert!(page("https://example.com", 4096, 1).is_ok()); + for (url, width, height) in [ + (" ", 400, 800), + ("https://example.com", 0, 800), + ("https://example.com", 400, 4097), + ] { + assert!( + matches!(page(url, width, height), Err(Error::Invalid(_))), + "{url:?} {width}x{height}" + ); + } + } + #[test] fn colors_parse_and_print() { assert_eq!("#102030".parse::().unwrap(), Rgba([0x10, 0x20, 0x30, 0xff])); diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 37d2b67..a38f06c 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -8,7 +8,7 @@ use kio::clock::Instant; use moq_video::render::{Renderer, wgpu}; use crate::Error; -use crate::collection::{self, Canvas, Collection, Item, Layer as Place, Source}; +use crate::collection::{Canvas, Collection, Item, Layer as Place, Source}; use crate::effect::{self, Effects, Frame, Lut, Picture, Prepared}; use crate::source::{self, Sources}; @@ -305,6 +305,8 @@ impl Compositor { lut: None, error: None, }); + // A page arrives premultiplied; frames and images are straight. + let premultiplied = matches!(picture, source::Picture::Texture(_)); let texture = match picture { source::Picture::Frame(frame) => { let renderer = match layer.renderer.as_mut() { @@ -317,6 +319,7 @@ impl Compositor { .map_err(Error::from) } source::Picture::Rgba { size, pixels } => upload(&self.device, &self.queue, size, &pixels), + source::Picture::Texture(texture) => Ok(texture), source::Picture::Lut(cube) => { layer.lut = Some(self.effects.lut(&cube)); continue; @@ -326,7 +329,7 @@ impl Compositor { match texture { Ok(texture) => { layer.error = None; - layer.picture = Some(self.effects.picture(&texture)); + layer.picture = Some(self.effects.picture(&texture, premultiplied)); } Err(err) => { if layer.error.is_none() { @@ -400,12 +403,10 @@ impl Draw<'_> { fn prepare(&mut self, item: &Item) -> Option { let collection = self.collection; let picture = self.library.0.get(collection.capture(&item.source))?.picture.as_ref()?; - let video = collection.video.get(&item.source); - let crop = video.map_or(collection::Crop::default(), |video| video.crop); let item = effect::Item { picture, rect: item.rect, - crop: crop.then(&item.crop), + crop: collection.crop(item), filters: collection.filters(item).collect(), blend: item.blend, }; @@ -479,7 +480,7 @@ impl Target { | wgpu::TextureUsages::COPY_SRC, view_formats: &[], }); - let picture = effects.picture(&texture); + let picture = effects.picture(&texture, false); Self { texture, picture } } } diff --git a/rs/moq-scene/src/effect.rs b/rs/moq-scene/src/effect.rs index 9ad98a6..6b8be00 100644 --- a/rs/moq-scene/src/effect.rs +++ b/rs/moq-scene/src/effect.rs @@ -323,6 +323,8 @@ fn reach(sigma: f32) -> f32 { pub(crate) struct Picture { view: wgpu::TextureView, bind: wgpu::BindGroup, + /// Whether its alpha is premultiplied, as a web page's is. + premultiplied: bool, } /// A LUT source's table on the GPU. @@ -659,7 +661,7 @@ impl Effects { } /// `texture`, a source's picture or a canvas-sized target, bound to sample. - pub fn picture(&self, texture: &wgpu::Texture) -> Picture { + pub fn picture(&self, texture: &wgpu::Texture, premultiplied: bool) -> Picture { let view = texture.create_view(&Default::default()); let bind = self.device.create_bind_group(&wgpu::BindGroupDescriptor { label: Some("moq-scene input"), @@ -675,7 +677,11 @@ impl Effects { }, ], }); - Picture { view, bind } + Picture { + view, + bind, + premultiplied, + } } /// Upload a LUT's table. @@ -825,7 +831,7 @@ impl Effects { area, uv: crop, valid: crop, - premultiplied: false, + premultiplied: item.picture.premultiplied, }; let stages: Vec = item.filters.iter().map(|filter| Stage::of(filter)).collect(); @@ -859,12 +865,13 @@ impl Effects { MaskMode::Alpha => 0.0, MaskMode::Luminance => 1.0, }; + let picture = library.picture(source)?; let op = Op { kind: [Op::MASK, 0, 0, 0], - params: [mode, 0.0, 0.0, 0.0], + params: [mode, f32::from(u8::from(picture.premultiplied)), 0.0, 0.0], ..Default::default() }; - (op, None, Some(library.picture(source)?.view)) + (op, None, Some(picture.view)) } filter => (Op::of(filter).expect("a per-pixel filter"), None, None), }; @@ -1172,7 +1179,7 @@ impl Effects { }); Raster { size, - picture: self.picture(&texture), + picture: self.picture(&texture, true), } } }; diff --git a/rs/moq-scene/src/effect.wgsl b/rs/moq-scene/src/effect.wgsl index 043d1ba..40d503a 100644 --- a/rs/moq-scene/src/effect.wgsl +++ b/rs/moq-scene/src/effect.wgsl @@ -19,7 +19,8 @@ const LUMA: vec3 = vec3(0.2126, 0.7152, 0.0722); struct Op { kind: vec4, // MATRIX: unused. CHROMA_KEY: key Cb, key Cr, similarity, smoothness. - // LUMA_KEY: min, max, smoothness. LUT: amount, size. MASK: mode. + // LUMA_KEY: min, max, smoothness. LUT: amount, size. MASK: mode, and 1 if + // the mask is premultiplied. params: vec4, // MATRIX: applied to straight RGBA. LUT: the domain's max in column 0. matrix: mat4x4, @@ -143,7 +144,8 @@ fn coverage(item: vec2, op: Op) -> f32 { } let color = textureSampleLevel(mask, bilinear, item, 0.0); if op.params.x > 0.5 { - return dot(color.rgb, LUMA) * color.a; + let rgb = select(color.rgb * color.a, color.rgb, op.params.y > 0.5); + return dot(rgb, LUMA); } return color.a; } diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 9a18b05..88d6b4d 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -12,7 +12,8 @@ //! scenes and collections as text. //! - [`source::Sources`] runs each source: a camera, display, window, //! microphone, or system-audio capture (with the `capture` feature), a still -//! image, a solid color, or a broadcast subscribed over MoQ. Pictures go to the +//! image, a solid color, a broadcast subscribed over MoQ, or a web page +//! rendered by a browser engine (the [`browser`] module). Pictures go to the //! compositor and sound goes to a `moq_audio::bus::Bus`, the program audio. //! - [`Compositor`] draws the scene in preview and the scene on program into //! textures on the caller's `wgpu` device, cutting or fading between them. @@ -26,6 +27,7 @@ //! source that stalls repeats its last picture rather than holding up the //! canvas. +pub mod browser; pub mod clip; mod collection; mod compositor; @@ -105,4 +107,8 @@ pub enum Error { /// The broadcast subscription failed. #[error(transparent)] Net(#[from] moq_net::Error), + + /// A browser source's page failed, or its helper process stopped. + #[error("browser: {0}")] + Browser(String), } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 74880d4..c6c1773 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -6,7 +6,7 @@ //! dropped, which happens when the source is removed or edited. //! //! With [`Context::publish`] set, each local source that shows a picture also -//! publishes it as its own track; see [`crate::publish`]. +//! publishes it as its own track, web pages aside; see [`crate::publish`]. use std::collections::BTreeMap; use std::future::Future; @@ -22,6 +22,7 @@ use moq_mux::catalog::Stream; use tokio::sync::watch; use crate::Error; +use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::publish::{Claim, Listing, Publish, Still}; @@ -33,13 +34,20 @@ pub(crate) enum Picture { Frame(moq_video::Frame), /// Tightly packed RGBA with straight alpha, such as a still image. Rgba { size: moq_video::Size, pixels: Vec }, + /// A texture on the compositor's device with premultiplied alpha, such as + /// a web page. + #[cfg_attr( + not(target_os = "linux"), + expect(dead_code, reason = "browser sources are Linux-only") + )] + Texture(moq_video::render::wgpu::Texture), /// A LUT source's table, which is not drawn but looked up. Lut(Cube), } /// The one picture a source holds for the compositor. A newer picture replaces /// an undrawn one, so a slow canvas skips pictures instead of queuing them. -type Mailbox = Arc>>; +pub(crate) type Mailbox = Arc>>; /// When a source last delivered a picture, `None` before its first. type Shown = Arc>>; @@ -110,6 +118,8 @@ pub struct Context { /// What cameras, screens, and broadcasts deliver video as: /// `moq_video::render::output` of the compositor's device. pub output: moq_video::Output, + /// What browser sources run in; without it they fail as unsupported. + pub browser: Option, /// Where local sources publish their own tracks, or `None` to publish nothing. pub publish: Option, } @@ -139,8 +149,9 @@ impl Sources { /// clones rather than opening one. Returns a driver for each source started, /// which does nothing until run. /// - /// Also applies the collection's gain and mute to every audio strip, which - /// never restarts a source. Panics on a gain [`Collection::validate`] refuses. + /// Also applies the collection's gain and mute to every audio strip, and its + /// frame rate to every web page, which never restarts a source. Panics on a + /// gain [`Collection::validate`] refuses. pub fn sync(&mut self, collection: &Collection) -> Vec { self.feeds .retain(|name, feed| collection.sources.get(name) == Some(&feed.source)); @@ -152,6 +163,10 @@ impl Sources { self.feeds.insert(name.clone(), feed); drivers.push(driver); } + // A page paints at the canvas's frame rate, live. + if let Some(page) = self.feeds[name].page() { + page.set_framerate(collection.canvas.framerate); + } if let Some(strip) = self.feeds[name].strip() { let audio = collection.audio.get(name).copied().unwrap_or_default(); @@ -180,6 +195,7 @@ pub struct Feed { shown: Shown, strip: Option, status: kio::Consumer, + page: Option, track: Option>, played: Playback, /// A replay source's next replay. @@ -237,7 +253,8 @@ impl Feed { } /// What the source's own track is doing, or `None` when it publishes none: - /// nothing is published, or the source is another broadcast or only sound. + /// nothing is published, or the source is another broadcast, a web page, or + /// only sound. /// /// A failed track leaves the source drawing in the studio's own scenes. pub fn track(&self) -> Option { @@ -264,6 +281,11 @@ impl Feed { self.strip.as_ref() } + /// A browser source's page: how it paints, and interaction with it. + pub fn page(&self) -> Option<&Page> { + self.page.as_ref() + } + /// The newest picture not yet drawn. pub(crate) fn take(&self) -> Option { self.picture.lock().unwrap().take() @@ -280,6 +302,11 @@ pub struct Driver { shown: Shown, strip: Option, status: kio::Producer, + #[cfg_attr( + not(target_os = "linux"), + expect(dead_code, reason = "browser sources are Linux-only") + )] + page: Option, /// The source's own track, when it publishes one. track: Option, played: Playback, @@ -392,6 +419,7 @@ impl Driver { std::future::pending().await } Source::Broadcast { path } => self.broadcast(path).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"); self.replays(plays, listing).await @@ -438,6 +466,7 @@ impl Driver { Source::Color { .. } | Source::Image { .. } | Source::Broadcast { .. } + | Source::Browser { .. } | Source::Clone { .. } | Source::Lut { .. } | Source::Replay => unreachable!("not a device source"), @@ -495,6 +524,37 @@ impl Driver { )) } + /// Render the page at `url` until its helper stops, or fail at once if the + /// studio offers no helper. + #[cfg(target_os = "linux")] + async fn browser(&self, url: &str, width: u32, height: u32) -> Result<(), Error> { + let Some(helper) = &self.context.browser else { + return Err(Error::Unsupported( + "this build cannot render web pages (enable the `cef` feature)".into(), + )); + }; + let page = self.page.as_ref().expect("browser sources have a page"); + let bus = self.context.bus.config().clone(); + let open = crate::browser::Open { + url: url.to_string(), + width, + height, + sample_rate: bus.sample_rate, + channels: bus.layout.channels(), + ..Default::default() + }; + let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); + input.format = moq_audio::Format::F32; + let strip = self.strip.clone().expect("browser sources have a strip"); + let sound = self.context.bus.channel(input, strip)?; + crate::browser::run(helper, open, page, &self.picture, Some(sound), &self.status).await + } + + #[cfg(not(target_os = "linux"))] + async fn browser(&self, _: &str, _: u32, _: u32) -> Result<(), Error> { + Err(Error::Unsupported("browser sources run on Linux only for now".into())) + } + /// 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> { @@ -843,6 +903,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { let shown = Shown::default(); let strip = source.has_audio().then(Strip::default); let status = kio::Producer::new(Status::Starting); + let page = matches!(source, Source::Browser { .. }).then(Page::default); // Claimed here, in the order sources open, so an edited source's // replacement owns the listing however their drivers are scheduled. let track = context @@ -870,6 +931,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { shown: shown.clone(), strip: strip.clone(), status: status.consume(), + page: page.clone(), track: track.as_ref().map(|track| track.status.consume()), played: played.clone(), plays, @@ -882,6 +944,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { shown, strip, status, + page, track, played, plays: replays, @@ -895,9 +958,15 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { /// layout rather than encoded again. /// /// Microphones and system audio do not yet: moq-audio offers no capture that -/// feeds both the program bus and an encoder. +/// feeds both the program bus and an encoder. Nor do web pages, which viewers +/// see only in the program: H.264 would encode a transparent overlay over +/// black. fn publishes(source: &Source) -> bool { - source.has_video() && !matches!(source, Source::Broadcast { .. } | Source::Clone { .. }) + source.has_video() + && !matches!( + source, + Source::Broadcast { .. } | Source::Clone { .. } | Source::Browser { .. } + ) } /// The frame rate a capture is encoded at when the device reports none. @@ -940,6 +1009,7 @@ pub(crate) mod test { origin: origin.consume(), dir: dir.to_path_buf(), output: moq_video::Output::Cpu, + browser: None, publish: None, }; (context, origin) @@ -1363,6 +1433,36 @@ mod tests { .await; } + /// Without a helper to run it in, a browser source fails rather than + /// staying blank. + #[tokio::test] + async fn a_page_without_a_helper_says_so() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let mut collection = Collection::new(); + collection.sources.insert( + "chat".into(), + Source::Browser { + url: "https://example.com".into(), + width: 400, + height: 800, + }, + ); + let (context, _origin) = context(dir.path()); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let chat = sources.get("chat").unwrap(); + let Status::Failed(reason) = chat.status() else { + panic!("a page without a helper fails"); + }; + assert!(reason.contains("web pages") || reason.contains("Linux"), "{reason}"); + assert!(chat.page().is_some_and(|page| page.paint().is_none())); + assert!(chat.strip().is_some(), "a page feeds the bus"); + }) + .await; + } + #[cfg(not(feature = "capture"))] #[tokio::test] async fn a_build_without_capture_says_so() { diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index 3a2b3ce..7e773b1 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -32,6 +32,10 @@ vaapi = ["moq-video/vaapi"] # Wayland screens and windows are not reachable yet: display and window sources # capture X11 only. pipewire = ["capture", "moq-scene/pipewire"] +# Browser sources, rendered by CEF in a helper process (Linux). Off by default: +# the build downloads the CEF runtime (or takes it from `CEF_PATH`) and copies +# it beside the binary, where the binary looks for it when it starts. +cef = ["moq-scene/cef"] [dependencies] anyhow = { workspace = true, features = ["backtrace"] } diff --git a/rs/moq-studio/build.rs b/rs/moq-studio/build.rs new file mode 100644 index 0000000..f64ff44 --- /dev/null +++ b/rs/moq-studio/build.rs @@ -0,0 +1,17 @@ +//! With `cef` on Linux, the binary loads `libcef.so` when it starts: from +//! beside itself, where a package installs the CEF runtime, or else from +//! `CEF_PATH` when the build has one, as the dev shell does. + +use std::path::Path; + +fn main() { + println!("cargo::rerun-if-env-changed=CEF_PATH"); + let cef = std::env::var_os("CARGO_FEATURE_CEF").is_some(); + if !cef || std::env::var("CARGO_CFG_TARGET_OS").as_deref() != Ok("linux") { + return; + } + println!("cargo::rustc-link-arg-bins=-Wl,-rpath,$ORIGIN"); + if let Some(dir) = std::env::var_os("CEF_PATH").filter(|dir| Path::new(dir).join("libcef.so").exists()) { + println!("cargo::rustc-link-arg-bins=-Wl,-rpath,{}", Path::new(&dir).display()); + } +} diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 504b363..fd4d57e 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -490,6 +490,7 @@ impl Engine { dir, output: moq_video::render::output(&gpu.device), publish: publish.as_ref().map(|publish| publish.sources.clone()), + browser: browser(&gpu)?, }); let mut compositor = Compositor::new(&gpu.device, &gpu.queue, collection.canvas)?; compositor.set_preview(collection.scenes.first().map(|scene| scene.name.clone())); @@ -1202,6 +1203,21 @@ fn encode_frames( Ok(()) } +/// What browser sources run in: this executable, which `main` hands to +/// `moq_scene::browser::child` when it is started as their helper. +fn browser(gpu: &Gpu) -> anyhow::Result> { + if !cfg!(feature = "cef") { + return Ok(None); + } + let program = std::env::current_exe().context("cannot find the studio's executable")?; + Ok(Some(moq_scene::browser::Helper::new( + program, + &gpu.adapter, + &gpu.device, + &gpu.queue, + ))) +} + /// Run a source on its own thread until its feed is dropped, or fail the /// source when its thread cannot start. fn spawn(driver: Driver) { diff --git a/rs/moq-studio/src/interact.rs b/rs/moq-studio/src/interact.rs new file mode 100644 index 0000000..5d287b3 --- /dev/null +++ b/rs/moq-studio/src/interact.rs @@ -0,0 +1,996 @@ +//! Interaction with browser sources through the preview: the pointer goes to +//! the page under it, and the keyboard to the page last clicked. + +use std::cell::RefCell; +use std::time::Duration; + +use kio::clock::Instant; +use moq_scene::browser::{Button, Input, Modifiers}; +use moq_scene::{Collection, Crop, Rect, Source}; + +/// How close in time and space a press must follow the last to count toward a +/// double or triple click. +const MULTI_CLICK: Duration = Duration::from_millis(500); +const MULTI_CLICK_SLOP: i32 = 4; + +/// Which page has the pointer and the keyboard. +#[derive(Default)] +pub struct Interact { + /// The page under the pointer, so it hears when the pointer leaves. + hovered: Option, + /// The page last clicked, which receives the keyboard. + focused: Option, + /// The page a button went down on, which gets the pointer wherever it + /// goes until every button it saw go down comes up, so a drag never + /// leaves a page with a button stuck down. + captured: Option, + /// The last press, to count multi-clicks. + last: Option, +} + +struct Press { + page: String, + button: Button, + x: i32, + y: i32, + at: Instant, + clicks: u32, +} + +struct Capture { + page: String, + /// Where the pressed item places the page, since the same page can be in + /// several. + place: Place, + /// The last point the page heard, where a release lands that comes from + /// no pointer. + at: (i32, i32), + /// The buttons the page saw go down and not yet up. + held: Vec