diff --git a/examples/voip-cli/src/main.rs b/examples/voip-cli/src/main.rs index 6914c7945..28e13decf 100644 --- a/examples/voip-cli/src/main.rs +++ b/examples/voip-cli/src/main.rs @@ -3,37 +3,49 @@ //! Audio is bridged to the system through `cpal` (cross-platform; ALSA on Linux, //! CoreAudio on macOS, WASAPI on Windows). `cpal` is a dev-dependency, so it is //! linked only for this example and never reaches consumers of the library. +//! VIDEO is bridged through `ffmpeg`/`ffplay` subprocesses (must be on PATH when +//! `--video` is used): ffmpeg encodes the webcam / a file / a test pattern to H.264 +//! and ffplay renders the peer's stream — the library only transports encoded AUs. //! -//! Subcommands: -//! loopback Mic → Opus → E2E-SRTP protect → unprotect → Opus → speaker. -//! Exercises the whole media stack locally; NO WhatsApp connection. -//! Run it and you should hear yourself, processed by the voip pipeline. -//! listen [accept] Connect to WhatsApp, print incoming calls; reject (default) or accept. -//! call Connect, discover the peer's devices, encrypt the callKey per device, -//! and send a ``; logs the signaling flow via raw nodes. +//! Subcommands (all accept a trailing `--video`): +//! loopback [--video] Mic → Opus → E2E-SRTP protect → unprotect → Opus → speaker. +//! Exercises the whole media stack locally; NO WhatsApp connection. +//! With --video: ffmpeg source → AU splitter → ffplay window instead. +//! listen [accept] [--video] Connect, print incoming calls; reject (default) or accept. +//! With --video an accepted call answers with video media too. +//! call [--video] Place a call; with --video it is a video call from the start. //! -//! cargo run --example voip --features "voip sqlite-storage tokio-transport ureq-client tokio-native" -- loopback +//! cargo run -p whatsapp-rust-voip-cli --release -- loopback +//! +//! During a live call, single-key commands on stdin (terminal only): `v` toggles video +//! (upgrade / accept a pending peer request / downgrade), `q` hangs up. +//! Env: `WA_VIDEO_INPUT` = `testsrc` | file/URL | webcam device (default: OS webcam); +//! `WA_VIDEO_SINK` = `window` (default) | `file` | `none`; optional quality overrides: +//! `WA_VIDEO_SIZE`, `WA_VIDEO_FPS`, `WA_VIDEO_BITRATE_KBPS`, `WA_VIDEO_SINK_FPS`. //! //! The inbound MEDIA path is the library facade: `client.voip().accept(&call).audio(mic, //! speaker).start()` returns a `CallHandle` and the library owns the callKey decrypt, the relay //! socket, the sans-IO engine, and the task lifetime. This example only supplies the cpal audio -//! device and reacts to engine events. +//! device / ffmpeg pipes and reacts to engine events. use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, AtomicI32, Ordering}; use std::sync::{Arc, Mutex}; -use anyhow::{Context, Result, anyhow}; -use log::{error, info, warn}; +use anyhow::{Result, anyhow, bail}; +use log::{debug, error, info, warn}; use portable_atomic::AtomicU64; use wacore::stanza::call::{self as stanza, CAPABILITY_OFFER}; use wacore::types::call::{CallAction, IncomingCall}; use wacore::types::events::{Event, EventHandler}; use wacore::voip::CallEvent; use whatsapp_rust::prelude::*; -use whatsapp_rust::voip::CallHandle; use whatsapp_rust::voip::audio::{WaOpusDecoder, WaOpusEncoder}; use whatsapp_rust::voip::session::{MediaPipeline, MediaPipelineParams}; +use whatsapp_rust::voip::{CallHandle, VideoState}; + +mod video; +use video::VideoOpts; const FRAME_SAMPLES: usize = 960; // 60 ms @ 16 kHz const WA_RATE: u32 = 16_000; @@ -87,29 +99,81 @@ async fn main() -> Result<()> { ); let args: Vec = std::env::args().collect(); - match args.get(1).map(String::as_str) { - Some("loopback") => run_loopback().await, - Some("listen") => { - run_bot(Mode::Listen { - accept: args.get(2).map(String::as_str) == Some("accept"), - }) - .await + let command = parse_cli(&args); + if video_source_is_ignored(&command, std::env::var_os("WA_VIDEO_INPUT").is_some()) { + warn!( + "WA_VIDEO_INPUT is set, but --video is missing; outbound video is disabled. Keep \ + --video on the same shell command line." + ); + } + match command { + CliCommand::Loopback { video: true } => { + video::run_video_loopback(&VideoOpts::from_env().await?).await } - Some("call") => { - let jid = args - .get(2) - .context("usage: voip call ")? - .parse::() - .map_err(|e| anyhow!("bad jid: {e}"))?; - run_bot(Mode::Call(jid)).await + CliCommand::Loopback { video: false } => run_loopback().await, + CliCommand::Listen { accept, video } => run_bot(Mode::Listen { accept, video }).await, + CliCommand::Call { jid, video } => { + let jid = jid.parse::().map_err(|e| anyhow!("bad jid: {e}"))?; + run_bot(Mode::Call { jid, video }).await } - _ => { - eprintln!("usage: voip >"); + CliCommand::Usage => { + eprintln!("usage: voip > [--video]"); Ok(()) } } } +/// A parsed CLI invocation. Kept separate from `Mode` (and pure) so the argument classification — +/// including the `--video`-implies-accept rule that bit a real test run — is unit-testable. +#[derive(Debug, PartialEq, Eq)] +enum CliCommand { + Loopback { video: bool }, + Listen { accept: bool, video: bool }, + Call { jid: String, video: bool }, + Usage, +} + +/// Classify `argv` (including `argv[0]`). `--video` may appear anywhere. On `listen`, `--video` +/// IMPLIES `accept`: there is no reason to request video while rejecting every call, so +/// `listen --video` means "accept video calls" rather than silently rejecting them (the footgun a +/// user hit: the phone showed "no answer" because the reject went out). +fn parse_cli(argv: &[String]) -> CliCommand { + let video = argv.iter().any(|a| a == "--video"); + let pos: Vec<&str> = argv + .iter() + .skip(1) + .map(String::as_str) + .filter(|a| *a != "--video") + .collect(); + match pos.first().copied() { + Some("loopback") => CliCommand::Loopback { video }, + Some("listen") => CliCommand::Listen { + accept: pos.get(1).copied() == Some("accept") || video, + video, + }, + Some("call") => match pos.get(1) { + Some(jid) => CliCommand::Call { + jid: (*jid).to_string(), + video, + }, + None => CliCommand::Usage, + }, + _ => CliCommand::Usage, + } +} + +fn video_source_is_ignored(command: &CliCommand, video_input_is_set: bool) -> bool { + video_input_is_set + && matches!( + command, + CliCommand::Loopback { video: false } + | CliCommand::Listen { + accept: false, + video: false, + } + ) +} + // ===================== cpal audio bridge ===================== // // The engine speaks 16 kHz mono i16 in 960-sample (60 ms) frames; the OS audio device speaks its own @@ -626,25 +690,20 @@ async fn run_loopback() -> Result<()> { // ===================== live call / listen ===================== enum Mode { - Listen { accept: bool }, - Call(Jid), + Listen { accept: bool, video: bool }, + Call { jid: Jid, video: bool }, } /// Drives calls off the typed `Event::IncomingCall` (no raw-node forwarding needed): on an offer it /// answers signaling then hands the MEDIA plane to the library facade -/// (`client.voip().accept(..).audio(..).start()`), on a terminate it hangs the matching call up. The -/// facade owns the relay socket, the callKey decrypt, the engine, and the task lifetime; this only -/// supplies the PipeWire mic/speaker and remembers the `CallHandle` so a `` can stop it. +/// (`client.voip().accept(..).audio(..).start()`). The facade owns the relay socket, callKey decrypt, +/// engine, and termination; this example supplies the mic/speaker and keeps handles for its UI. struct CallObserver { client: Arc, accept: bool, - /// Whether this run ever starts a media call (auto-accept inbound OR an outbound `call`), so a - /// `` racing media startup is worth recording. A pure-reject run starts no media, so it - /// must NOT record (the set would grow unbounded with nothing to consume it). - manages_media: bool, - /// Per-call bookkeeping for the example's terminate-driven hangup. The client's own - /// `CallRegistry` already tears every call down on disconnect; this is only so a `` - /// can stop a specific live call (and so a terminate that races media startup isn't lost). + /// `--video`: start/answer calls with video media and auto-accept peer upgrade requests. + video: bool, + /// Per-call UI bookkeeping. The client's `CallRegistry` owns media termination. state: Arc>, } @@ -652,45 +711,91 @@ struct CallObserver { struct CallState { /// Live calls' handles by call-id. handles: HashMap>, - /// Call-ids that were terminated BEFORE their media handle finished starting, so a late - /// `start_media()` hangs the call up instead of leaving an orphaned live call. - terminated: HashSet, + /// Registration order; the last live entry is the stdin UI target. + call_order: Vec, + /// The stdin `v` toggle's view of each call's video: pending peer request vs our video live. + video_ui: HashMap, + starting: HashSet, + /// Prevents a late startup from resurrecting media after the peer ended the call. + terminated_during_startup: HashSet, +} + +#[derive(Clone, Copy, PartialEq)] +enum VideoUi { + /// The peer asked for video (`UpgradeRequestV2`); `v` accepts it. + PendingPeerRequest, + /// Our video plane is up; `v` downgrades. + Active, } impl CallObserver { - fn new(client: Arc, accept: bool, manages_media: bool) -> Self { + fn new(client: Arc, accept: bool, video: bool) -> Self { Self { client, accept, - manages_media, + video, state: Arc::new(Mutex::new(CallState::default())), } } } -/// Register a freshly-started `CallHandle` for terminate-driven hangup. Returns false (and the caller -/// should hang up) if a `` for this call-id already arrived while media was starting. On -/// success, spawns the wait_ended cleanup that drops the map entry when the call ends on its own. -/// Shared by the inbound (accept) and outbound (call) paths. -fn register_handle(state: &Arc>, cid: String, handle: Arc) -> bool { - let registered = { - let mut st = state.lock().unwrap(); - if st.terminated.remove(&cid) { - false - } else { - st.handles.insert(cid.clone(), handle.clone()); - true - } - }; - if !registered { - return false; +fn lock_call_state(state: &Mutex) -> std::sync::MutexGuard<'_, CallState> { + state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn begin_call_startup(state: &Arc>, call_id: &str) { + let mut st = lock_call_state(state); + st.starting.insert(call_id.to_string()); + st.terminated_during_startup.remove(call_id); +} + +fn complete_call_startup(state: &Arc>, call_id: &str) -> bool { + let mut st = lock_call_state(state); + st.starting.remove(call_id); + st.terminated_during_startup.remove(call_id) +} + +fn record_peer_terminate(state: &Arc>, call_id: &str) { + let mut st = lock_call_state(state); + if st.starting.contains(call_id) { + st.terminated_during_startup.insert(call_id.to_string()); + } + st.handles.remove(call_id); + st.video_ui.remove(call_id); + st.call_order.retain(|id| id != call_id); +} + +fn peer_terminated_during_startup(state: &Arc>, call_id: &str) -> bool { + lock_call_state(state) + .terminated_during_startup + .contains(call_id) +} + +fn mark_call_most_recent(order: &mut Vec, call_id: &str) { + order.retain(|id| id != call_id); + order.push(call_id.to_string()); +} + +/// Register a handle for UI control and remove it when the library-owned call ends. +async fn register_handle(state: &Arc>, cid: String, handle: Arc) { + if complete_call_startup(state, &cid) { + // The library also sees the terminate, but this closes a handle created after that teardown. + handle.hangup().await; + info!("◾ discarded media startup for already-ended call {cid}"); + return; + } + { + let mut st = lock_call_state(state); + st.handles.insert(cid.clone(), handle.clone()); + mark_call_most_recent(&mut st.call_order, &cid); } - // Drop our map entry once the call ends on its own (no terminate). let state = state.clone(); tokio::spawn(async move { handle.wait_ended().await; { - let mut st = state.lock().unwrap(); + let mut st = lock_call_state(&state); // Remove only if it is still OUR handle: a same-call-id replacement may now own this slot // (its own cleanup will remove it), so we must not delete the live call. if st @@ -699,67 +804,68 @@ fn register_handle(state: &Arc>, cid: String, handle: Arc) { if let Event::IncomingCall(call) = &*event { match &call.action { - CallAction::Offer { call_id, .. } => { + CallAction::Offer { + call_id, is_video, .. + } => { let client = self.client.clone(); let call = call.clone(); let accept = self.accept; + // Only answer with video when we're allowed to AND the offer is actually a video + // call; advertising `