From d9e544607a3060668ed6193a6495df57efa57911 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 6 Jun 2026 15:46:34 -0400 Subject: [PATCH] feat: screen sharing via pixelpass (Discord-style, presence-borne ticket) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surface pixelpass screen-sharing from inside a peerspeak room. peerspeak owns voice, pixelpass owns pixels β€” they're never Cargo deps of each other; the contract is pixelpass's CLI flags + its `--output json` stdout stream. Modelled on Discord: multiple simultaneous sharers, a πŸ”΄ Live badge + πŸ‘ Watch on each sharing peer's card, and in-progress shares visible to late joiners. - New `src/screenshare` module: pure `parse_pixelpass_event` seam + `pixelpass_path` discovery (13 unit tests), async `spawn_host` (β†’ ticket) and `spawn_viewer` (β†’ parse connected{url} β†’ open mpv, vlc fallback). No new deps. - Sharing rides presence: `PeerState.sharing: Option` (serde-defaulted), so the existing gossip re-announce delivers the offer to late joiners for free and a PeerUpdated fires on start/stop β€” no separate gossip message needed. - core: Start/Stop/ViewShare commands; host + viewer children tracked in the session, killed on stop/leave (kill_on_drop backstop). Viewer limit left to pixelpass's bandwidth-measured cap. - UI: Share/Stop button (graceful "needs pixelpass" disabled state), Live badge + Watch on peer cards, Sharing badge on the self card. Verified by screenshot. - config: optional `pixelpass_path` override (hand-editable). Tests-green; the 2-machine gossip/remote path is not yet field-verified. Co-Authored-By: Claude Opus 4.8 --- docs/FEATURES.md | 18 +- docs/screenshare-integration.md | 22 +- src/app/mod.rs | 88 +++++++ src/bin/test_net.rs | 2 + src/config.rs | 5 + src/core/messages.rs | 16 ++ src/core/mod.rs | 120 ++++++++- src/lib.rs | 1 + src/network/gossip.rs | 1 + src/network/mod.rs | 8 + src/screenshare/mod.rs | 426 ++++++++++++++++++++++++++++++++ 11 files changed, 703 insertions(+), 4 deletions(-) create mode 100644 src/screenshare/mod.rs diff --git a/docs/FEATURES.md b/docs/FEATURES.md index bcf09ef..acbabd3 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -68,6 +68,19 @@ covers internals). When you ship a feature, add it here. | History cap | βš™οΈ | `CHAT_HISTORY_MAX = 300`. | | Local echo of own messages | βœ… | | +## Screen share (via pixelpass) + +| Feature | Status | Notes | +|---|---|---| +| Share your screen | πŸ§ͺ | Spawns `pixelpass --host --output json`, announces the ticket on presence. `src/screenshare/mod.rs`. Subprocess contract not yet 2-machine-verified. | +| Watch a peer's share | πŸ§ͺ | πŸ‘ Watch spawns a pixelpass viewer β†’ opens the stream in mpv (vlc fallback). | +| Live badge on sharing peers | βœ… | πŸ”΄ Live on the peer card; πŸ”΄ Sharing badge on the self card. | +| Multiple simultaneous sharers | πŸ§ͺ | Discord-style: each sharer is an independent pixelpass host; the room shows one offer per sharer. | +| Late-joiner sees in-progress share | πŸ§ͺ | Sharing rides presence, so the existing gossip re-announce delivers the offer to late joiners. | +| Graceful "needs pixelpass" state | βœ… | Share button disabled + labelled when the binary isn't on `$PATH`. Config override: `pixelpass_path`. | +| Viewer limit | βš™οΈ | Not overridden β€” pixelpass bandwidth-measures its own safe cap (protects the sharer's uplink); refusals surface as an error. | +| Audio of the share | βš™οΈ | Video only on the room path (so it can't echo/double with the voice mix). | + ## Recording | Feature | Status | Notes | @@ -122,11 +135,14 @@ Re-run on a real desktop ↔ dopedart call before calling these done: 4. **In-room text chat** β€” 2-machine delivery. 5. **Adaptive jitter buffer** β€” behaviour under real packet loss/jitter. 6. **Chat drawer** β€” toggle + persistence across restart. +7. **Screen share** β€” 2-machine: sharer's ticket reaches the room over gossip, + a peer's Watch opens the stream, late joiner sees an in-progress share, stop + clears the badge. (Subprocess contract is solo-verifiable; the gossip/remote + half needs dopedart.) Requires `pixelpass` + `mpv` on both ends. ## Not built (candidate features) Things that came up but **do not** exist yet: - Soundboard (play short clips into the call mix). -- Video / screen share. - Invite links beyond the raw ticket / room persistence. diff --git a/docs/screenshare-integration.md b/docs/screenshare-integration.md index 86a1f86..4baee0e 100644 --- a/docs/screenshare-integration.md +++ b/docs/screenshare-integration.md @@ -1,8 +1,26 @@ # Screen-share integration (peerspeak ↔ pixelpass) Design decisions for surfacing pixelpass screen-sharing from inside a peerspeak -room. Locked 2026-06-06; **not yet implemented.** peerspeak owns voice, -pixelpass owns pixels β€” this doc is the seam between them. +room. Locked 2026-06-06; **implemented 2026-06-06** (`src/screenshare/mod.rs` + +presence-borne ticket; tests-green, the 2-machine gossip path not yet +field-verified). peerspeak owns voice, pixelpass owns pixels β€” this doc is the +seam between them. + +## Open questions β€” resolved (Discord-modelled) + +- **Concurrency:** multiple simultaneous sharers, Discord-style. Sharing rides + *presence* (`PeerState.sharing: Option`), so each sharer shows one + offer; no artificial one-at-a-time cap. +- **Late joiners:** yes β€” because sharing is presence, the existing gossip + re-announce delivers an in-progress offer to a new joiner for free (no + separate `ScreenShareOffer` message was needed; the doc's sketch is superseded + by the presence-field approach). +- **Viewer limit:** not overridden β€” pixelpass bandwidth-measures its own safe + cap; a full host's `viewer_refused` surfaces as a UI error. +- **UI placement:** per-peer card (πŸ”΄ Live + πŸ‘ Watch on the sharer's card; + πŸ”΄ Sharing badge + Stop on the self card) β€” the Discord "Live badge" model. +- **Player:** viewer parses `connected{url}` and opens mpv (vlc fallback), + mirroring pixelpass's own low-latency invocation. ## Core principle: mutually optional, runtime-only coupling diff --git a/src/app/mod.rs b/src/app/mod.rs index 8e9e326..5f638be 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -152,6 +152,10 @@ pub enum AppMessage { SelectRoomLayout(RoomLayout), /// Toggle the Chat drawer open/closed (drawer layout). ToggleDrawerChat, + /// Start/stop sharing our own screen (spawns/kills a pixelpass host). + ToggleScreenShare, + /// Watch a peer's screen share, identified by their pixelpass ticket. + WatchShare(String), } fn core_subscription() -> impl iced::futures::Stream { @@ -217,6 +221,10 @@ pub struct AppState { ever_connected: HashSet, controller: Arc, current_screen: Screen, + /// Whether we're currently sharing our own screen (confirmed by the core). + self_sharing: bool, + /// Whether the `pixelpass` binary is available, gating the Share controls. + pixelpass_available: bool, } impl AppState { @@ -254,6 +262,9 @@ impl Default for AppState { let _ = controller.send(CoreCommand::SetInputVolume(config.input_volume)); let _ = controller.send(CoreCommand::SetOutputVolume(config.output_volume)); let _ = controller.send(CoreCommand::SetNetworkMode(config.network_mode)); + let _ = controller.send(CoreCommand::SetPixelpassPath(config.pixelpass_path.clone())); + let pixelpass_available = + crate::screenshare::is_available(config.pixelpass_path.as_deref()); let all_devices = enumerate_audio_devices(); let input_devices: Vec<_> = all_devices.iter().filter(|d| d.is_input).cloned().collect(); let output_devices: Vec<_> = all_devices.iter().filter(|d| !d.is_input).cloned().collect(); @@ -298,6 +309,8 @@ impl Default for AppState { ever_connected: HashSet::new(), controller, current_screen: Screen::Home, + self_sharing: false, + pixelpass_available, } } } @@ -407,6 +420,18 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { AppMessage::LeavePressed => { let _ = state.controller.send(CoreCommand::Leave); } + AppMessage::ToggleScreenShare => { + if state.self_sharing { + let _ = state.controller.send(CoreCommand::StopScreenShare); + } else { + let _ = state.controller.send(CoreCommand::StartScreenShare); + state.status_message = "Starting screen share…".to_string(); + } + } + AppMessage::WatchShare(ticket) => { + let _ = state.controller.send(CoreCommand::ViewShare(ticket)); + state.status_message = "Opening screen share…".to_string(); + } AppMessage::ToggleMutePressed => { let _ = state.controller.send(CoreCommand::ToggleMute); state.is_muted = !state.is_muted; @@ -445,6 +470,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { state.status_message = "Ready to connect".to_string(); state.current_screen = Screen::Home; state.mic_level = 0.0; + state.self_sharing = false; notify::play(Sound::SelfLeave, state.config.custom_sound_self_leave.as_deref()); } UiEvent::PeerJoined { id, state: peer_state } => { @@ -510,6 +536,14 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { push_chat(&mut state.chat_messages, ChatEntry { name, text, mine: false }); } } + UiEvent::ScreenShareStarted => { + state.self_sharing = true; + state.status_message = "Sharing your screen".to_string(); + } + UiEvent::ScreenShareStopped => { + state.self_sharing = false; + state.status_message = "Screen share stopped".to_string(); + } UiEvent::Error(err) => { state.status_message = format!("Error: {}", err); } @@ -1270,6 +1304,15 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { } ] .align_y(iced::alignment::Vertical::Center), + // Live "you're sharing" badge β€” only present while sharing. + { + let el: Element<'_, AppMessage> = if state.self_sharing { + text("πŸ”΄ Sharing your screen").size(13).color(color_red).into() + } else { + iced::widget::Space::new().width(0.0).height(0.0).into() + }; + el + }, progress_bar(0.0..=0.3, state.mic_level) .girth(8.0) .style(move |_t: &Theme| iced::widget::progress_bar::Style { @@ -1317,6 +1360,26 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { .style(b_style(mute_bg, color_blue, mute_fg, 6.0)) .padding(6); + // Screen-share "Live" badge + Watch button when this peer is sharing. + // Watch is enabled only if pixelpass is installed locally. + let share_el: Element<'_, AppMessage> = if let Some(ticket) = peer.sharing.clone() { + let mut watch_btn = button(text("πŸ‘ Watch").size(13)) + .style(b_style(color_blue, color_lavender, color_crust, 6.0)) + .padding(6); + if state.pixelpass_available { + watch_btn = watch_btn.on_press(AppMessage::WatchShare(ticket)); + } + row![ + text("πŸ”΄ Live").size(13).color(color_red), + watch_btn, + ] + .spacing(6) + .align_y(iced::alignment::Vertical::Center) + .into() + } else { + iced::widget::Space::new().width(0.0).height(0.0).into() + }; + // VU meter colour: dim when locally muted (you don't hear them), // green while speaking, faint otherwise. let vu_color = if is_locally_muted { @@ -1334,6 +1397,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { text(format!("ID: {}", &peer_id.to_string()[..8])).size(11).color(color_subtext) ], horizontal_space(), + share_el, mute_btn, indicator ] @@ -1444,6 +1508,30 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { .padding(14) .width(iced::Length::Fill) }, + vertical_space(20.0), + { + // Screen share. Disabled (no on_press) when pixelpass is absent, + // with the label saying so β€” a normal, handled state. + let (share_label, share_bg, share_hover, share_fg) = if !state.pixelpass_available { + ("πŸ–₯ Needs pixelpass", color_surface, color_surface, color_subtext) + } else if state.self_sharing { + ("πŸ›‘ Stop Sharing", color_red, color_maroon, color_crust) + } else { + ("πŸ–₯ Share Screen", color_surface, color_blue, color_text) + }; + let mut share_btn = button( + text(share_label) + .size(16) + .align_x(iced::alignment::Horizontal::Center), + ) + .style(b_style(share_bg, share_hover, share_fg, 8.0)) + .padding(14) + .width(iced::Length::Fill); + if state.pixelpass_available { + share_btn = share_btn.on_press(AppMessage::ToggleScreenShare); + } + share_btn + }, vertical_space(30.0), button( text("Leave Room") diff --git a/src/bin/test_net.rs b/src/bin/test_net.rs index f4945e2..11eac9e 100644 --- a/src/bin/test_net.rs +++ b/src/bin/test_net.rs @@ -61,6 +61,7 @@ async fn main() -> Result<(), Box> { name: "Alice".to_string(), is_muted: false, addr: endpoint_a.addr(), + sharing: None, }; room_a.join(&ticket_str, state_a).await?; println!("Node A joined topic."); @@ -78,6 +79,7 @@ async fn main() -> Result<(), Box> { name: "Bob".to_string(), is_muted: false, addr: endpoint_b.addr(), + sharing: None, }; room_b.join(&ticket_str, state_b).await?; println!("Node B joined topic."); diff --git a/src/config.rs b/src/config.rs index 80d78be..bc86f2a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -145,6 +145,10 @@ pub struct AppConfig { pub custom_sound_mic_toggle: Option, #[serde(default)] pub custom_sound_reconnect_failed: Option, + /// Optional override for the `pixelpass` binary location (screen share). + /// Empty / unset = look it up on `$PATH`. Hand-editable; no Settings UI yet. + #[serde(default)] + pub pixelpass_path: Option, } impl Default for AppConfig { @@ -172,6 +176,7 @@ impl Default for AppConfig { custom_sound_self_leave: None, custom_sound_mic_toggle: None, custom_sound_reconnect_failed: None, + pixelpass_path: None, } } } diff --git a/src/core/messages.rs b/src/core/messages.rs index fa09f25..3f9c1d9 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -31,6 +31,18 @@ pub enum CoreCommand { SetRecording(bool), /// Broadcast a room text-chat message. No-op when not in a call. SendChat(String), + /// Set the pixelpass binary location (config override, empty = use `$PATH`). + /// Sent at startup so screen-share can resolve the binary. + SetPixelpassPath(Option), + /// Start sharing our screen: spawn a pixelpass host and announce its ticket + /// on our presence so the room can watch. No-op when not in a call. + StartScreenShare, + /// Stop sharing our screen: kill the pixelpass host and clear the presence + /// ticket. No-op when not sharing. + StopScreenShare, + /// Watch a peer's screen share: spawn a pixelpass viewer for `ticket` and + /// open it in a local player. + ViewShare(String), } #[derive(Debug, Clone)] @@ -56,5 +68,9 @@ pub enum UiEvent { /// A room text-chat message arrived from a peer (never our own β€” local /// messages are echoed by the UI on send). ChatMessage { name: String, text: String }, + /// Our own screen share started; the UI flips the Share button to "Stop". + ScreenShareStarted, + /// Our own screen share stopped (or failed to start). + ScreenShareStopped, Error(String), } diff --git a/src/core/mod.rs b/src/core/mod.rs index b525973..96426a2 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -313,11 +313,26 @@ struct ActiveSession { transport: Arc, /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop. echo_cancel: Option, + /// Our pixelpass screen-share host child while sharing (`kill_on_drop`, so it + /// also dies if the session is dropped without an explicit stop). + screenshare_host: Option, + /// pixelpass viewer children we spawned to watch peers' shares; killed on + /// session teardown (each also self-exits when its player window closes). + screenshare_viewers: Vec, } impl ActiveSession { - async fn shutdown(self, audio_backend: Arc) { + async fn shutdown(mut self, audio_backend: Arc) { crate::log_msg("ActiveSession::shutdown started"); + // Tear down any screen-share children first so the host stops streaming + // promptly (kill_on_drop is the backstop, but kill explicitly so viewers + // see the stream end without waiting on drop ordering). + if let Some(mut host) = self.screenshare_host.take() { + let _ = host.kill().await; + } + for mut viewer in self.screenshare_viewers.drain(..) { + let _ = viewer.kill().await; + } self.datagram_task.abort(); self.mixer_task.abort(); self.event_task.abort(); @@ -407,6 +422,10 @@ async fn run_core_loop( let locally_muted = Arc::new(Mutex::new(HashSet::::new())); let mut current_name = "Anonymous".to_string(); let mut network_mode = NetworkMode::default(); + // Pixelpass binary override (config), and the ticket of our own active screen + // share (rides our presence so the room β€” incl. late joiners β€” can watch). + let mut pixelpass_override: Option = None; + let mut current_sharing: Option = None; let mut active_session: Option = None; // Standalone capture-only mic meter, live only when no session exists. @@ -501,10 +520,13 @@ async fn run_core_loop( memory_lookup.clone(), )); + // Fresh join starts not sharing; clear any stale share ticket. + current_sharing = None; let self_state = PeerState { name: current_name.clone(), is_muted: is_muted.load(Ordering::Relaxed), addr: endpoint.addr(), + sharing: None, }; crate::log_msg(&format!("Attempting room_state.join with self_state={:?}", self_state)); @@ -910,6 +932,8 @@ async fn run_core_loop( grace_timers, transport: transport.clone(), echo_cancel: echo_cancel_guard, + screenshare_host: None, + screenshare_viewers: Vec::new(), }; let self_id = endpoint.id().to_string(); @@ -920,6 +944,7 @@ async fn run_core_loop( CoreCommand::Leave => { // Finalize any recording first, while the audio feeders are alive. stop_recording(&recorder, &is_recording, &ui_tx).await; + current_sharing = None; if let Some(session) = active_session.take() { session.shutdown(audio_backend.clone()).await; let _ = ui_tx.send(UiEvent::RoomLeft).await; @@ -936,6 +961,7 @@ async fn run_core_loop( name: current_name.clone(), is_muted: new_state, addr: session.endpoint.addr(), + sharing: current_sharing.clone(), }; let _ = session.room_state.update_self_state(self_state).await; } @@ -1069,6 +1095,98 @@ async fn run_core_loop( crate::log_msg(&format!("Failed to send chat: {e}")); } } + + CoreCommand::SetPixelpassPath(path) => { + pixelpass_override = path.filter(|p| !p.trim().is_empty()); + } + + CoreCommand::StartScreenShare => { + let Some(session) = &mut active_session else { + let _ = ui_tx + .send(UiEvent::Error("Join a call before sharing your screen".into())) + .await; + continue; + }; + if session.screenshare_host.is_some() { + continue; // already sharing + } + let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) { + Some(b) => b, + None => { + let _ = ui_tx + .send(UiEvent::Error( + "pixelpass not found β€” install it to share your screen".into(), + )) + .await; + continue; + } + }; + match crate::screenshare::spawn_host(&bin).await { + Ok((child, ticket)) => { + crate::log_msg("Screen share host started"); + session.screenshare_host = Some(child); + current_sharing = Some(ticket.clone()); + let self_state = PeerState { + name: current_name.clone(), + is_muted: is_muted.load(Ordering::Relaxed), + addr: session.endpoint.addr(), + sharing: Some(ticket), + }; + let _ = session.room_state.update_self_state(self_state).await; + let _ = ui_tx.send(UiEvent::ScreenShareStarted).await; + } + Err(e) => { + let _ = ui_tx + .send(UiEvent::Error(format!("Couldn't start screen share: {e}"))) + .await; + } + } + } + + CoreCommand::StopScreenShare => { + current_sharing = None; + if let Some(session) = &mut active_session { + if let Some(mut child) = session.screenshare_host.take() { + let _ = child.kill().await; + crate::log_msg("Screen share host stopped"); + } + let self_state = PeerState { + name: current_name.clone(), + is_muted: is_muted.load(Ordering::Relaxed), + addr: session.endpoint.addr(), + sharing: None, + }; + let _ = session.room_state.update_self_state(self_state).await; + } + let _ = ui_tx.send(UiEvent::ScreenShareStopped).await; + } + + CoreCommand::ViewShare(ticket) => { + let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) { + Some(b) => b, + None => { + let _ = ui_tx + .send(UiEvent::Error( + "pixelpass not found β€” install it to watch screen shares".into(), + )) + .await; + continue; + } + }; + match crate::screenshare::spawn_viewer(&bin, &ticket).await { + Ok(child) => { + crate::log_msg("Screen share viewer started"); + if let Some(session) = &mut active_session { + session.screenshare_viewers.push(child); + } + } + Err(e) => { + let _ = ui_tx + .send(UiEvent::Error(format!("Couldn't watch screen share: {e}"))) + .await; + } + } + } } } diff --git a/src/lib.rs b/src/lib.rs index a8f7210..8fbd1bf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ pub mod core; pub mod app; pub mod config; pub mod notify; +pub mod screenshare; use std::path::PathBuf; use std::sync::OnceLock; diff --git a/src/network/gossip.rs b/src/network/gossip.rs index 730525c..faab78c 100644 --- a/src/network/gossip.rs +++ b/src/network/gossip.rs @@ -339,6 +339,7 @@ mod tests { name: "TestPeerGossip".to_string(), is_muted: true, addr, + sharing: None, } } diff --git a/src/network/mod.rs b/src/network/mod.rs index 6750c69..bbf4515 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -27,6 +27,13 @@ pub struct PeerState { pub name: String, pub is_muted: bool, pub addr: iroh::EndpointAddr, + /// When this peer is sharing their screen, the pixelpass relay ticket a + /// viewer needs to watch it; `None` when not sharing. Riding presence means + /// the existing gossip re-announce delivers it to late joiners for free, and + /// a `PeerUpdated` fires automatically on share start/stop. Defaulted so + /// older configs / peers that predate the field still deserialize. + #[serde(default)] + pub sharing: Option, } #[derive(Debug, Clone)] @@ -161,6 +168,7 @@ mod tests { name: "TestPeer".to_string(), is_muted: false, addr, + sharing: None, } } diff --git a/src/screenshare/mod.rs b/src/screenshare/mod.rs new file mode 100644 index 0000000..a30bf37 --- /dev/null +++ b/src/screenshare/mod.rs @@ -0,0 +1,426 @@ +//! Screen-share integration: drive `pixelpass` as a child process. +//! +//! peerspeak owns voice; pixelpass owns pixels. The two are **never** Cargo +//! dependencies of each other β€” the contract is pixelpass's CLI flags plus its +//! `--output json` stdout event stream, treated as a stable public API. This +//! module spawns a pixelpass *host* (to share our screen) or a *viewer* (to +//! watch a peer's share), scrapes the JSON it needs, and otherwise stays out of +//! the way. Absence of the `pixelpass` binary is a normal, handled state β€” the +//! UI degrades to a disabled "install pixelpass" control rather than erroring. +//! +//! The only piece that travels between peers is the host's relay **ticket**: +//! peerspeak puts it on the sharer's presence (see `PeerState.sharing`) so the +//! room's existing gossip plane distributes it, and each viewer one-clicks it +//! into a local pixelpass viewer. The ticket *is* the capability, so this is +//! "click to grant the room access" with no ACL to fight. + +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::time::Duration; + +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::process::{Child, Command}; + +/// The binary we shell out to. Looked up on `$PATH` unless a config override +/// points elsewhere. +const PIXELPASS_BIN: &str = "pixelpass"; + +/// How long to wait for the host to emit its ticket / the viewer to connect +/// before giving up and killing the child. Startup is normally sub-second; this +/// is only a safety net so a hung pixelpass can't wedge the caller forever. +const STARTUP_TIMEOUT: Duration = Duration::from_secs(20); + +/// One parsed line from pixelpass's `--output json` stdout stream. Mirrors the +/// `event` tags in pixelpass's `src/common/output.rs`. Recognized-but-unused +/// events collapse to [`PixelpassEvent::Other`]; blank or non-JSON lines parse +/// to `None`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PixelpassEvent { + /// Host: the relay ticket a viewer needs. Emitted once at host startup. + Ticket(String), + /// Viewer: the local player URL is ready to open. + Connected(String), + /// Host: a viewer joined; carries the new active count and the cap. + ViewerJoined { active: u32, max: u32 }, + /// Host: a viewer left. + ViewerLeft { active: u32, max: u32 }, + /// Host: a viewer was turned away (host full or capture spawn failed). + Refused(String), + /// Host: capture pipeline spawned (on first viewer). + CaptureStarted, + /// Host: capture pipeline torn down (on last viewer). + CaptureStopped, + /// A recognized event we don't act on (e.g. `host_info`). + Other, +} + +/// Parse a single stdout line from pixelpass `--output json`. Pure: no I/O. +pub fn parse_pixelpass_event(line: &str) -> Option { + let line = line.trim(); + if line.is_empty() { + return None; + } + let v: serde_json::Value = serde_json::from_str(line).ok()?; + let event = v.get("event")?.as_str()?; + let ev = match event { + "ticket" => PixelpassEvent::Ticket(v.get("value")?.as_str()?.to_string()), + "connected" => PixelpassEvent::Connected(v.get("url")?.as_str()?.to_string()), + "viewer_joined" => PixelpassEvent::ViewerJoined { + active: json_u32(&v, "active"), + max: json_u32(&v, "max"), + }, + "viewer_left" => PixelpassEvent::ViewerLeft { + active: json_u32(&v, "active"), + max: json_u32(&v, "max"), + }, + "viewer_refused" => PixelpassEvent::Refused( + v.get("reason") + .and_then(|r| r.as_str()) + .unwrap_or("") + .to_string(), + ), + "capture" => match v.get("state").and_then(|s| s.as_str()) { + Some("started") => PixelpassEvent::CaptureStarted, + Some("stopped") => PixelpassEvent::CaptureStopped, + _ => PixelpassEvent::Other, + }, + _ => PixelpassEvent::Other, + }; + Some(ev) +} + +fn json_u32(v: &serde_json::Value, key: &str) -> u32 { + v.get(key).and_then(|x| x.as_u64()).unwrap_or(0) as u32 +} + +/// Resolve the pixelpass binary: an explicit config override (used only if it +/// points at an existing file), otherwise the first `pixelpass` found on +/// `$PATH`. `None` means it isn't installed β€” a normal, handled state. An +/// override that doesn't resolve falls through to the `$PATH` search rather than +/// failing outright. +pub fn pixelpass_path(config_override: Option<&str>) -> Option { + if let Some(p) = config_override { + let p = p.trim(); + if !p.is_empty() { + let pb = PathBuf::from(p); + if pb.is_file() { + return Some(pb); + } + // Override set but missing β€” fall through to the $PATH search. + } + } + let path_var = std::env::var_os("PATH")?; + std::env::split_paths(&path_var) + .map(|dir| dir.join(PIXELPASS_BIN)) + .find(|c| c.is_file()) +} + +/// Whether pixelpass is available to shell out to. +pub fn is_available(config_override: Option<&str>) -> bool { + pixelpass_path(config_override).is_some() +} + +/// Spawn a pixelpass host (`pixelpass --host --output json`), wait for its +/// startup ticket, and return the live child plus the ticket. The child keeps +/// running (streaming to viewers) until killed or dropped; remaining stdout is +/// drained in a background task so a full pipe can't stall the host. We do +/// **not** pass `--max-viewers`: pixelpass bandwidth-measures its own safe cap, +/// protecting the sharer's uplink, and refuses extras with `viewer_refused`. +pub async fn spawn_host(bin: &Path) -> std::io::Result<(Child, String)> { + let mut child = Command::new(bin) + .arg("--host") + .arg("--output") + .arg("json") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn()?; + + let stdout = child + .stdout + .take() + .ok_or_else(|| std::io::Error::other("pixelpass host stdout missing"))?; + let mut lines = BufReader::new(stdout).lines(); + + let ticket = match read_until(&mut lines, |e| match e { + PixelpassEvent::Ticket(t) => Some(t), + _ => None, + }) + .await + { + Ok(Some(t)) => t, + Ok(None) => { + let _ = child.kill().await; + return Err(std::io::Error::other( + "pixelpass host exited before emitting a ticket", + )); + } + Err(e) => { + let _ = child.kill().await; + return Err(e); + } + }; + + drain_in_background(lines, "host"); + Ok((child, ticket)) +} + +/// Spawn a pixelpass viewer for `ticket`, wait for it to connect, and open the +/// stream in a local player (mpv, falling back to vlc). Returns the live viewer +/// child so the caller can kill it on room-leave; it also self-exits when the +/// player window closes (its tunnel ends). +pub async fn spawn_viewer(bin: &Path, ticket: &str) -> std::io::Result { + let mut child = Command::new(bin) + .arg(ticket) + .arg("--output") + .arg("json") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn()?; + + let stdout = child + .stdout + .take() + .ok_or_else(|| std::io::Error::other("pixelpass viewer stdout missing"))?; + let mut lines = BufReader::new(stdout).lines(); + + let url = match read_until(&mut lines, |e| match e { + PixelpassEvent::Connected(u) => Some(u), + _ => None, + }) + .await + { + Ok(Some(u)) => u, + Ok(None) => { + let _ = child.kill().await; + return Err(std::io::Error::other( + "pixelpass viewer exited before connecting", + )); + } + Err(e) => { + let _ = child.kill().await; + return Err(e); + } + }; + + if let Err(e) = launch_player(&url) { + let _ = child.kill().await; + return Err(e); + } + + drain_in_background(lines, "viewer"); + Ok(child) +} + +/// Read JSON event lines until `pick` returns `Some(value)`. Returns `Ok(None)` +/// on EOF (child exited first) and `Err` on an I/O error or the startup timeout. +async fn read_until( + lines: &mut tokio::io::Lines>, + mut pick: impl FnMut(PixelpassEvent) -> Option, +) -> std::io::Result> +where + R: tokio::io::AsyncRead + Unpin, +{ + loop { + match tokio::time::timeout(STARTUP_TIMEOUT, lines.next_line()).await { + Ok(Ok(Some(line))) => { + if let Some(ev) = parse_pixelpass_event(&line) + && let Some(v) = pick(ev) + { + return Ok(Some(v)); + } + } + Ok(Ok(None)) => return Ok(None), + Ok(Err(e)) => return Err(e), + Err(_) => { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "timed out waiting for pixelpass startup event", + )); + } + } + } +} + +/// Keep reading the child's stdout to EOF in the background so a full pipe can't +/// stall it; log notable events for diagnostics. +fn drain_in_background(mut lines: tokio::io::Lines>, role: &'static str) +where + R: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + tokio::spawn(async move { + while let Ok(Some(line)) = lines.next_line().await { + if let Some(ev) = parse_pixelpass_event(&line) { + crate::log_msg(&format!("pixelpass {role}: {ev:?}")); + } + } + }); +} + +/// Open the viewer stream URL in a media player. Mirrors pixelpass's own +/// low-latency mpv invocation; falls back to vlc. The player is reaped in a +/// background task so it doesn't linger as a zombie when its window closes. +fn launch_player(url: &str) -> std::io::Result<()> { + const MPV_ARGS: &[&str] = &[ + "--profile=low-latency", + "--untimed", + "--hwdec=auto", + "--audio-buffer=0.2", + "--demuxer-max-bytes=2M", + "--demuxer-readahead-secs=0.5", + ]; + const VLC_ARGS: &[&str] = &["--network-caching=200", "--live-caching=200"]; + + let child = match spawn_player("mpv", MPV_ARGS, url) { + Ok(c) => c, + Err(_) => spawn_player("vlc", VLC_ARGS, url).map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::NotFound, + "no media player found β€” install mpv or vlc to watch screen shares", + ) + })?, + }; + tokio::spawn(async move { + let mut child = child; + let _ = child.wait().await; + }); + Ok(()) +} + +fn spawn_player(bin: &str, args: &[&str], url: &str) -> std::io::Result { + Command::new(bin) + .args(args) + .arg(url) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .kill_on_drop(false) + .spawn() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_ticket() { + assert_eq!( + parse_pixelpass_event(r#"{"event":"ticket","value":"abc123"}"#), + Some(PixelpassEvent::Ticket("abc123".to_string())) + ); + } + + #[test] + fn parses_connected_url() { + assert_eq!( + parse_pixelpass_event(r#"{"event":"connected","url":"http://127.0.0.1:5500"}"#), + Some(PixelpassEvent::Connected("http://127.0.0.1:5500".to_string())) + ); + } + + #[test] + fn parses_viewer_joined_and_left() { + assert_eq!( + parse_pixelpass_event(r#"{"event":"viewer_joined","id":"x","active":2,"max":4}"#), + Some(PixelpassEvent::ViewerJoined { active: 2, max: 4 }) + ); + assert_eq!( + parse_pixelpass_event(r#"{"event":"viewer_left","id":"x","active":1,"max":4}"#), + Some(PixelpassEvent::ViewerLeft { active: 1, max: 4 }) + ); + } + + #[test] + fn viewer_counts_default_to_zero_when_absent() { + assert_eq!( + parse_pixelpass_event(r#"{"event":"viewer_joined"}"#), + Some(PixelpassEvent::ViewerJoined { active: 0, max: 0 }) + ); + } + + #[test] + fn parses_refused_with_and_without_reason() { + assert_eq!( + parse_pixelpass_event(r#"{"event":"viewer_refused","reason":"host is full"}"#), + Some(PixelpassEvent::Refused("host is full".to_string())) + ); + assert_eq!( + parse_pixelpass_event(r#"{"event":"viewer_refused"}"#), + Some(PixelpassEvent::Refused("".to_string())) + ); + } + + #[test] + fn parses_capture_states() { + assert_eq!( + parse_pixelpass_event(r#"{"event":"capture","state":"started"}"#), + Some(PixelpassEvent::CaptureStarted) + ); + assert_eq!( + parse_pixelpass_event(r#"{"event":"capture","state":"stopped"}"#), + Some(PixelpassEvent::CaptureStopped) + ); + // Unknown capture state is recognized-but-unused, not a parse failure. + assert_eq!( + parse_pixelpass_event(r#"{"event":"capture","state":"paused"}"#), + Some(PixelpassEvent::Other) + ); + } + + #[test] + fn recognized_but_unused_event_is_other() { + assert_eq!( + parse_pixelpass_event( + r#"{"event":"host_info","display_server":"wayland","max_viewers":4}"# + ), + Some(PixelpassEvent::Other) + ); + assert_eq!( + parse_pixelpass_event(r#"{"event":"some_future_event"}"#), + Some(PixelpassEvent::Other) + ); + } + + #[test] + fn blank_and_non_json_lines_are_none() { + assert_eq!(parse_pixelpass_event(""), None); + assert_eq!(parse_pixelpass_event(" "), None); + assert_eq!(parse_pixelpass_event("not json at all"), None); + // Valid JSON but not an event object. + assert_eq!(parse_pixelpass_event("[1,2,3]"), None); + assert_eq!(parse_pixelpass_event(r#"{"no_event":"here"}"#), None); + } + + #[test] + fn ticket_without_value_is_none() { + // A malformed ticket event (missing `value`) must not panic. + assert_eq!(parse_pixelpass_event(r#"{"event":"ticket"}"#), None); + } + + #[test] + fn surrounding_whitespace_is_tolerated() { + assert_eq!( + parse_pixelpass_event(" {\"event\":\"ticket\",\"value\":\"t\"}\n"), + Some(PixelpassEvent::Ticket("t".to_string())) + ); + } + + #[test] + fn path_override_to_existing_file_is_used() { + // A real file (this source file) stands in for a custom binary location. + let this_file = file!(); + let resolved = pixelpass_path(Some(this_file)); + assert_eq!(resolved.as_deref(), Some(Path::new(this_file))); + assert!(is_available(Some(this_file))); + } + + #[test] + fn empty_override_falls_through() { + // An empty/whitespace override is ignored (falls through to $PATH); we + // only assert it doesn't return the empty path as a match. + assert_ne!(pixelpass_path(Some(" ")).as_deref(), Some(Path::new(""))); + } +}