diff --git a/CHANGELOG.md b/CHANGELOG.md index b433800..7ff6aa7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,8 +12,85 @@ for a staging or local environment must set both fields. Code that builds `..ClientConfig::default()`. `DEFAULT_COORDINATOR_WS_URL` moved from `rtc::coordinator::ws` to the crate root. +### Call events come in one stream for each source + +`Call`, `RtcCall` and `RtcCore` replace `subscribe()`, `on()`, `off()` and the +`CallEvent` enum with three streams: + +- `sfu_events()` gives `SfuCallEvent`: the events from the SFU, including + `CallEnded { reason }` for the SFU `call_ended`. +- `coordinator_events()` gives `CoordinatorEvent`: the call-scoped coordinator + events, including `call.ended`. +- `client_events()` gives `ClientCallEvent`: `CallingStateChanged`. + +The SDK leaves the call on the SFU `call_ended` or the coordinator +`call.ended`; the other one may then not arrive. `CallingStateChanged(Left)` is +the reliable end of the call. Each stream has its own buffer, so a lagging +receiver loses events only from its own stream. + +### Track events carry the SFU data + +`SfuCallEvent::TrackPublished` and `SfuCallEvent::TrackUnpublished` give +`track_type` as a `TrackType`, not an `i32`, and add `participant`. +`TrackUnpublished` also adds `cause`. Patterns that match these variants must +use the new fields or `..`. + +### Decoded audio frames carry their RTP timestamp + +`PcmFrame` adds `pts: Option`: the RTP timestamp of the first sample, in +units of 1/48000 s, wrapping like RTP. `RemoteTrack::next_pcm` sets it; a frame +rebuilt for a lost packet continues from the frame before it. Frames that the +application or a conversion builds have `None`, and `write_pcm` ignores the +field. Code that builds `PcmFrame` with a struct literal must set `pts` or use +`PcmFrame::new` / `PcmFrame::mono`. + +### Subscription config has the stream-py shape + +`SubscriptionConfig` replaces `audio`, `video`, `screen_share` and +`video_dimension` with the fields of the stream-py `SubscriptionConfig`: + +- `default: TrackSubscriptionConfig` gives `track_types`, `video_dimension` + and `screenshare_dimension`. Screen-share video and screen-share audio are + now separate track types, and screen share has its own dimension. +- `role_filters` gives a rule by participant role. The first role of the + participant that has a rule selects it; other participants use `default`. +- `max_subscriptions` limits the number of tracks. The tracks of the + participants that the call learned about first are kept. + +`SubscriptionConfig::default()` now subscribes to nothing, and +`SubscriptionConfig::matches` is removed. The presets `audio_all`, +`audio_video`, `all` and `none` stay. The default video and screen-share +dimension is now 1920×1080 (it was 1280×720), also for a `SubscriptionTarget` +without a dimension. `Call::participants` gives the participants in the order +the call learned about them. `set_incoming_video_enabled` keeps the configured +video dimension. + ## New Features +### A token-only client can prepare a call before the join + +`RtcClient::call` returns an `RtcCall` that is not joined yet, and +`RtcCall::join` joins it. Register `on_track` and subscribe before the join to +get the join events and tracks. `Call::rtc` gives the same `RtcCall` type for a +client with an API secret; both handles share one session. `RtcCall` also adds +`update_publish_options` and `set_disconnection_timeout`. + +### Stable call event names + +`SfuCallEvent::name` gives the stable `SfuEvent` field name of the source event +(for example `participant_joined` or `call_ended`), and +`participant_count_changed`. `ClientCallEvent::name` gives +`calling_state_changed`. A `CoordinatorEvent` has its coordinator `event_type` +(for example `call.created`). + +### Configurable call event buffer + +`ClientConfig::call_event_capacity` sets how many events each call event +stream keeps for a slow receiver. The default stays 256. A larger value makes a +lag less likely, but each call allocates all slots of its three streams. Code +that builds `ClientConfig` with a struct literal must set the new field or use +`..ClientConfig::default()`. + ### Video REST: advanced call statistics and reporting Application-level stats on `VideoClient` (`get_active_calls_status`, diff --git a/Cargo.lock b/Cargo.lock index 57d9f64..e8690ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1149,6 +1149,7 @@ dependencies = [ "flate2", "futures-util", "hmac 0.13.0", + "indexmap", "openh264", "opus", "prost", diff --git a/Cargo.toml b/Cargo.toml index 66c9285..7a89f67 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ base64 = "0.23.1" bytes = "1.12.1" futures-util = "0.3.34" hmac = "0.13.0" +indexmap = "2.14.2" prost = "0.14.4" prost-types = "0.14.4" reqwest = { version = "0.13.4", default-features = false, features = ["rustls", "json", "charset", "http2", "stream"] } diff --git a/README.md b/README.md index 88eae92..7d9d77a 100644 --- a/README.md +++ b/README.md @@ -31,8 +31,8 @@ remote audio and video, transform it, and publish media back into the call. feedback, and client call-event reporting. - Join a call as a server-side SFU participant with retry, reconnect, and migration handling. -- Subscribe globally or by participant session to remote audio, video, and - screen-share tracks. +- Subscribe globally, by participant role, or by participant session to remote + audio, video, and screen-share tracks. - Observe typed participant, connection-quality, pin, grant, and inbound-pause state from the SFU. - Read Opus audio as PCM, decode VP8/VP9/H264 video as I420, or work with raw @@ -252,8 +252,9 @@ async fn run_media_bridge(call: &Call) -> Result<(), Box> For a complete bridge with cancellation, barge-in, audio and video processing, and deterministic cleanup, see [`gpt_realtime_bot`](https://github.com/GetStream/stream-video-rust/blob/main/examples/gpt_realtime_bot.rs). -For selective agents, use `Call::update_subscription_targets` with -`SubscriptionTarget` values instead of subscribing to every participant. A +For selective agents, set `role_filters` and `max_subscriptions` in +`SubscriptionConfig`, or use `Call::update_subscription_targets` with +`SubscriptionTarget` values for an exact list of participant sessions. A temporary `mute_track` / `unmute_track` preserves the same local track and sender; `stop_publish` remains terminal for that local track handle. The latest SFU view is available synchronously through `Call::call_state`. diff --git a/examples/gpt_realtime_bot.rs b/examples/gpt_realtime_bot.rs index 2168a2b..c6019eb 100644 --- a/examples/gpt_realtime_bot.rs +++ b/examples/gpt_realtime_bot.rs @@ -52,7 +52,7 @@ use getstream::models::{CallRequest, GetOrCreateCallRequest, MemberRequest, User use getstream::rtc::proto::models::TrackType; use getstream::rtc::{ JoinCallData, LocalAudioTrack, LocalVideoTrack, RemoteTrack, RtcError, SubscriptionConfig, - VideoFrame, + TrackSubscriptionConfig, VideoFrame, }; use getstream::video::Call; use getstream::{Stream, TokenOptions}; @@ -328,6 +328,17 @@ async fn configure_openai( .await .context("add audio track to the OpenAI PeerConnection")?, ); + let paced_mic = mic.clone(); + pc.on_peer_connection_state_change(Box::new(move |state| { + let paced_mic = paced_mic.clone(); + Box::pin(async move { + if state == RTCPeerConnectionState::Connected { + paced_mic.start_pacing().await; + } else { + paced_mic.pause_pacing(); + } + }) + })); let camera = LocalVideoTrack::h264().context("H264 track for OpenAI")?; spawn_rtcp_drain( @@ -733,8 +744,12 @@ pub async fn start_bot( if let Err(error) = call .update_subscriptions(SubscriptionConfig { - video_dimension: Some((640, 360)), - ..SubscriptionConfig::audio_video() + default: TrackSubscriptionConfig { + track_types: vec![TrackType::Audio, TrackType::Video], + video_dimension: (640, 360), + ..Default::default() + }, + ..Default::default() }) .await .context("update_subscriptions") diff --git a/src/client.rs b/src/client.rs index 39ca7ac..9ceca59 100644 --- a/src/client.rs +++ b/src/client.rs @@ -32,6 +32,7 @@ const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(55); const DEFAULT_MAX_CONNS_PER_HOST: usize = 5; +const DEFAULT_CALL_EVENT_CAPACITY: usize = 256; /// Default maximum body accepted from coordinator HTTP endpoints (16 MiB). pub const DEFAULT_MAX_RESPONSE_BODY_BYTES: usize = 16 * 1024 * 1024; /// Default maximum inbound SFU/coordinator WebSocket frame and message size (4 MiB). @@ -120,6 +121,11 @@ pub struct ClientConfig { /// and known secret keys are redacted regardless. Other fields, including /// PII and secrets stored under custom keys, remain visible when enabled. pub log_bodies: bool, + /// Events that each call event stream (`sfu_events`, `coordinator_events`, + /// `client_events`) keeps for a slow receiver, rounded up to a power of two. + /// A receiver that falls further behind loses the oldest events. Each call + /// allocates all slots. Must be at least 1. Default 256. + pub call_event_capacity: usize, } impl Default for ClientConfig { @@ -133,6 +139,7 @@ impl Default for ClientConfig { max_conns_per_host: DEFAULT_MAX_CONNS_PER_HOST, retry: RetryConfig::default(), log_bodies: false, + call_event_capacity: DEFAULT_CALL_EVENT_CAPACITY, } } } @@ -148,6 +155,7 @@ pub(crate) struct Client { http: reqwest::Client, retry: RetryConfig, log_bodies: bool, + call_event_capacity: usize, max_response_body_bytes: usize, max_websocket_message_bytes: usize, stream_client_header: String, @@ -206,6 +214,14 @@ impl Client { config.coordinator_ws_url ))); } + // tokio broadcast channels panic outside this range. + if config.call_event_capacity == 0 || config.call_event_capacity > usize::MAX >> 1 { + return Err(Error::Config(format!( + "call event capacity {} must be between 1 and {}", + config.call_event_capacity, + usize::MAX >> 1 + ))); + } let http = reqwest::Client::builder() .pool_max_idle_per_host(config.max_conns_per_host) @@ -227,6 +243,7 @@ impl Client { http, retry: config.retry, log_bodies: config.log_bodies, + call_event_capacity: config.call_event_capacity, max_response_body_bytes: limits.max_response_body_bytes, max_websocket_message_bytes: limits.max_websocket_message_bytes, stream_client_header: format!("stream-rust-{}", env!("CARGO_PKG_VERSION")), @@ -245,6 +262,10 @@ impl Client { &self.coordinator_ws_url } + pub(crate) fn call_event_capacity(&self) -> usize { + self.call_event_capacity + } + /// The shared `reqwest` client (connection pool). Used by the RTC layer to /// reuse the pool for coordinator join + SFU Twirp calls. pub(crate) fn http(&self) -> &reqwest::Client { @@ -652,6 +673,17 @@ mod tests { assert!(matches!(error, Error::Config(_))); } + #[test] + fn call_event_capacity_must_be_at_least_one() { + let config = ClientConfig { + call_event_capacity: 0, + ..ClientConfig::default() + }; + let error = Client::new("key".to_owned(), "secret".to_owned(), config) + .expect_err("zero call event capacity"); + assert!(matches!(error, Error::Config(_))); + } + #[test] fn compatibility_limits_are_conservative_and_configurable() { let limits = NetworkLimits::default(); diff --git a/src/rtc/client.rs b/src/rtc/client.rs index 1d73aab..e238e26 100644 --- a/src/rtc/client.rs +++ b/src/rtc/client.rs @@ -14,9 +14,13 @@ use crate::client::{Client, ClientConfig, NetworkLimits}; use crate::error::Result as CrateResult; use crate::token::{self, TokenOptions}; +use super::coordinator::ws::CoordinatorEvent; use super::error::{Result, RtcError}; -use super::join::{CallEvent, CallStateSnapshot, CallingState, JoinCallData, RtcCore}; +use super::join::{ + CallStateSnapshot, CallingState, ClientCallEvent, JoinCallData, RtcCore, SfuCallEvent, +}; use super::proto::models::TrackType; +use super::publish_options::ClientPublishOptions; use super::subscriptions::{SubscriptionConfig, SubscriptionTarget}; use super::tracks::{LocalAudioTrack, LocalTrack, LocalVideoTrack, RemoteParticipant, RemoteTrack}; @@ -47,9 +51,9 @@ where pub(crate) enum UserTokenSource { Static(String), Provider(Arc), + /// Mints a call-scoped token for the joining user from the API secret. ServerMinted { client: Arc, - user_id: String, call_cid: String, expiration: Duration, }, @@ -61,13 +65,11 @@ impl std::fmt::Debug for UserTokenSource { Self::Static(_) => f.debug_tuple("Static").field(&"").finish(), Self::Provider(_) => f.debug_tuple("Provider").field(&"").finish(), Self::ServerMinted { - user_id, call_cid, expiration, .. } => f .debug_struct("ServerMinted") - .field("user_id", user_id) .field("call_cid", call_cid) .field("expiration", expiration) .finish_non_exhaustive(), @@ -82,12 +84,11 @@ impl UserTokenSource { Self::Provider(provider) => provider.load_token().await.map_err(RtcError::from)?, Self::ServerMinted { client, - user_id, call_cid, expiration, } => token::create_user_token( client.api_secret(), - user_id, + expected_user_id, &TokenOptions { expiration: Some(*expiration), call_cids: Some(vec![call_cid.clone()]), @@ -221,6 +222,18 @@ impl RtcClient { self } + /// A handle for `:` that is not joined yet. Register + /// callbacks and subscribe before [`RtcCall::join`] to get the join events + /// and tracks. + pub fn call(&self, call_type: impl Into, call_id: impl Into) -> RtcCall { + let core = RtcCore::new(self.client.clone(), call_type.into(), call_id.into()); + core.set_disconnection_timeout(self.disconnection_timeout); + RtcCall { + core, + token_source: self.token_source.clone(), + } + } + /// Join `:` and return a live [`RtcCall`] handle. pub async fn join( &self, @@ -228,37 +241,63 @@ impl RtcClient { call_id: impl Into, data: JoinCallData, ) -> Result { - let core = RtcCore::new(self.client.clone(), call_type.into(), call_id.into()); - core.set_disconnection_timeout(self.disconnection_timeout); - core.join_with_token_source(self.token_source.clone(), data) - .await?; - Ok(RtcCall { core }) + let call = self.call(call_type, call_id); + call.join(data).await?; + Ok(call) } } -/// A joined call handle from [`RtcClient::join`]. +/// A call handle from [`RtcClient::call`], [`RtcClient::join`], or +/// [`Call::rtc`](crate::Call::rtc). #[derive(Clone)] pub struct RtcCall { core: Arc, + token_source: UserTokenSource, } impl RtcCall { - /// Subscribe to the typed SFU event stream. - pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver { - self.core.subscribe() + pub(crate) fn new(core: Arc, token_source: UserTokenSource) -> Self { + Self { core, token_source } } - /// Register a callback for typed call events. - pub fn on(&self, callback: F) -> tokio::task::AbortHandle - where - F: Fn(CallEvent) + Send + 'static, - { - self.core.on(callback) + /// Join the call as an SFU participant with the user token of its client. + /// Illegal (typed error) if already `JOINING`/`JOINED`. + pub async fn join(&self, data: JoinCallData) -> Result<()> { + self.core + .join_with_token_source(self.token_source.clone(), data) + .await + } + + /// Set the maximum reconnect duration. Zero keeps reconnecting indefinitely. + pub fn set_disconnection_timeout(&self, timeout: Duration) { + self.core.set_disconnection_timeout(timeout); + } + + /// Update publishing preferences for the next join generation. + /// + /// Call this before [`Self::join`]. Updates after joining starts emit a + /// warning and cannot affect the active join generation. + pub fn update_publish_options(&self, options: ClientPublishOptions) { + self.core.update_publish_options(options); + } + + /// Subscribe to the events from the SFU. A receiver gets only events sent + /// after it subscribes. Subscribe before [`Self::join`] to get the join + /// events, or read [`Self::participants`] and [`Self::call_state`]. + pub fn sfu_events(&self) -> tokio::sync::broadcast::Receiver { + self.core.sfu_events() + } + + /// Subscribe to the call-scoped coordinator events. A receiver gets only + /// events sent after it subscribes. + pub fn coordinator_events(&self) -> tokio::sync::broadcast::Receiver { + self.core.coordinator_events() } - /// Remove a callback registered with [`Self::on`]. - pub fn off(&self, handler: &tokio::task::AbortHandle) { - self.core.off(handler); + /// Subscribe to the events that the SDK itself produces. See + /// [`RtcCore::client_events`]. + pub fn client_events(&self) -> tokio::sync::broadcast::Receiver { + self.core.client_events() } /// The current calling state. @@ -460,13 +499,13 @@ mod tests { ); let source = UserTokenSource::ServerMinted { client: client.clone(), - user_id: "user".to_owned(), call_cid: "default:call".to_owned(), expiration: Duration::from_secs(600), }; let token = source.load("user").await.expect("mint token"); let claims = token::decode_token(client.api_secret(), &token).expect("verify token"); + assert_eq!(claims.user_id, "user"); assert_eq!(claims.call_cids, Some(vec!["default:call".to_owned()])); assert_eq!( claims.exp.zip(claims.iat).map(|(exp, iat)| exp - iat), diff --git a/src/rtc/error.rs b/src/rtc/error.rs index fa53994..1b966e7 100644 --- a/src/rtc/error.rs +++ b/src/rtc/error.rs @@ -132,7 +132,7 @@ pub enum RtcError { track_type: models::TrackType, }, - /// A PCM write exceeded the track's low-latency queue. The newest samples + /// A PCM write exceeded the track's PCM queue capacity. The newest samples /// were retained and this many oldest samples were discarded. #[error( "pcm queue overflow: dropped {dropped_samples} oldest samples \ diff --git a/src/rtc/join/connection.rs b/src/rtc/join/connection.rs index eca997f..680dcf4 100644 --- a/src/rtc/join/connection.rs +++ b/src/rtc/join/connection.rs @@ -2,6 +2,44 @@ use super::*; +impl SfuCallEvent { + /// The stable `SfuEvent` field name of the source event, for example + /// `participant_joined` or `change_publish_quality`. The names are public + /// API and do not change. [`SfuCallEvent::ParticipantCountChanged`] is + /// `participant_count_changed`. + pub fn name(&self) -> &'static str { + match self { + Self::ParticipantJoined(_) => "participant_joined", + Self::ParticipantLeft(_) => "participant_left", + Self::ParticipantUpdated(_) => "participant_updated", + Self::TrackPublished { .. } => "track_published", + Self::TrackUnpublished { .. } => "track_unpublished", + Self::DominantSpeakerChanged { .. } => "dominant_speaker_changed", + Self::AudioLevelChanged(_) => "audio_level_changed", + Self::ConnectionQualityChanged(_) => "connection_quality_changed", + Self::ParticipantCountChanged(_) => "participant_count_changed", + Self::PinsUpdated(_) => "pins_updated", + Self::InboundStateChanged(_) => "inbound_state_notification", + Self::PublishOptionsChanged { .. } => "change_publish_options", + Self::PublishQualityChanged(_) => "change_publish_quality", + Self::CallGrantsUpdated(_) => "call_grants_updated", + Self::IceRestarted(_) => "ice_restart", + Self::Error(_) => "error", + Self::CallEnded { .. } => "call_ended", + } + } +} + +impl ClientCallEvent { + /// The stable name of this event: `calling_state_changed`. The names are + /// public API and do not change. + pub fn name(&self) -> &'static str { + match self { + Self::CallingStateChanged(_) => "calling_state_changed", + } + } +} + /// Build the SFU signaling WebSocket URL from `ws_endpoint`, appending the /// informational query params JS attaches (`attempt`, `user_id`, `api_key`, /// `user_session_id`, `cid`). Ported from JS `StreamSfuClient.createWebSocket`. @@ -127,6 +165,15 @@ pub(super) fn register_connection_state( Box::pin(async move { tracing::debug!(label, ?state, "stream.rtc.pc.state"); tracer.trace("connectionstatechange", json!(state.to_string())); + // Not awaited: waiting for the media lock here can block ICE + // state delivery. The sync reads the current publisher state, so + // a subscriber change applies the same state again. + if core.is_generation_current(generation) { + let sync_core = core.clone(); + std::mem::drop(core.spawn_generation_task(generation, async move { + sync_core.sync_audio_pacing().await; + })); + } if state == RTCPeerConnectionState::Connected { ever_connected.store(true, Ordering::SeqCst); let mut lifecycle = core.lifecycle.lock().unwrap_or_else(|e| e.into_inner()); @@ -151,7 +198,7 @@ pub(super) fn register_connection_state( } /// The SFU WebSocket event loop: negotiate subscriber offers, add remote ICE -/// candidates, and fan out typed [`CallEvent`]s. +/// candidates, and fan out typed [`SfuCallEvent`]s. pub(super) async fn event_loop(mut receiver: SfuReceiver, context: EventLoopContext) { loop { let event = match receiver.recv().await { @@ -253,16 +300,18 @@ pub(super) async fn handle_event( } E::ConnectionQualityChanged(event) => { core.update_connection_quality(&event.connection_quality_updates); - let _ = core.events_tx.send(CallEvent::ConnectionQualityChanged( - event.connection_quality_updates, - )); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::ConnectionQualityChanged( + event.connection_quality_updates, + )); } E::ParticipantJoined(ev) => { if let Some(p) = ev.participant { core.upsert_participant(&p); core.recompute_subscriptions_for_generation(context.generation) .await?; - let _ = core.events_tx.send(CallEvent::ParticipantJoined(p)); + let _ = core.sfu_events_tx.send(SfuCallEvent::ParticipantJoined(p)); } } E::ParticipantLeft(ev) => { @@ -270,7 +319,7 @@ pub(super) async fn handle_event( core.remove_participant(&p.session_id); core.recompute_subscriptions_for_generation(context.generation) .await?; - let _ = core.events_tx.send(CallEvent::ParticipantLeft(p)); + let _ = core.sfu_events_tx.send(SfuCallEvent::ParticipantLeft(p)); } } E::ParticipantUpdated(ev) => { @@ -278,7 +327,7 @@ pub(super) async fn handle_event( core.upsert_participant(&p); core.recompute_subscriptions_for_generation(context.generation) .await?; - let _ = core.events_tx.send(CallEvent::ParticipantUpdated(p)); + let _ = core.sfu_events_tx.send(SfuCallEvent::ParticipantUpdated(p)); } } E::TrackPublished(ev) => { @@ -290,64 +339,77 @@ pub(super) async fn handle_event( ); core.recompute_subscriptions_for_generation(context.generation) .await?; - let _ = core.events_tx.send(CallEvent::TrackPublished { + let _ = core.sfu_events_tx.send(SfuCallEvent::TrackPublished { user_id: ev.user_id, session_id: ev.session_id, - track_type: ev.r#type, + track_type: TrackType::try_from(ev.r#type).unwrap_or(TrackType::Unspecified), + participant: ev.participant, }); } E::TrackUnpublished(ev) => { core.remove_published_track(&ev.session_id, ev.r#type); core.recompute_subscriptions_for_generation(context.generation) .await?; - let _ = core.events_tx.send(CallEvent::TrackUnpublished { + let _ = core.sfu_events_tx.send(SfuCallEvent::TrackUnpublished { user_id: ev.user_id, session_id: ev.session_id, - track_type: ev.r#type, + track_type: TrackType::try_from(ev.r#type).unwrap_or(TrackType::Unspecified), + cause: models::TrackUnpublishReason::try_from(ev.cause) + .unwrap_or(models::TrackUnpublishReason::Unspecified), + participant: ev.participant, }); } E::DominantSpeakerChanged(ev) => { core.update_dominant_speaker(&ev.session_id); - let _ = core.events_tx.send(CallEvent::DominantSpeakerChanged { - user_id: ev.user_id, - session_id: ev.session_id, - }); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::DominantSpeakerChanged { + user_id: ev.user_id, + session_id: ev.session_id, + }); } E::AudioLevelChanged(ev) => { core.update_audio_levels(&ev.audio_levels); let _ = core - .events_tx - .send(CallEvent::AudioLevelChanged(ev.audio_levels)); + .sfu_events_tx + .send(SfuCallEvent::AudioLevelChanged(ev.audio_levels)); } E::HealthCheckResponse(event) => { - if let Some(participant_count) = event.participant_count { - core.update_participant_count(participant_count); + if let Some(participant_count) = event.participant_count + && core.update_participant_count(participant_count) + { let _ = core - .events_tx - .send(CallEvent::ParticipantCountChanged(participant_count)); + .sfu_events_tx + .send(SfuCallEvent::ParticipantCountChanged(participant_count)); } } E::PinsUpdated(event) => { core.update_pins(event.pins.clone()); - let _ = core.events_tx.send(CallEvent::PinsUpdated(event.pins)); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::PinsUpdated(event.pins)); } E::InboundStateNotification(event) => { core.update_inbound_state(&event.inbound_video_states); - let _ = core - .events_tx - .send(CallEvent::InboundStateChanged(event.inbound_video_states)); + let _ = core.sfu_events_tx.send(SfuCallEvent::InboundStateChanged( + event.inbound_video_states, + )); } E::ChangePublishOptions(event) => { core.apply_publish_options(context.generation, event.publish_options.clone()) .await?; - let _ = core.events_tx.send(CallEvent::PublishOptionsChanged { - publish_options: event.publish_options, - reason: event.reason, - }); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::PublishOptionsChanged { + publish_options: event.publish_options, + reason: event.reason, + }); } E::ChangePublishQuality(event) => { core.apply_publish_quality(&event).await; - let _ = core.events_tx.send(CallEvent::PublishQualityChanged(event)); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::PublishQualityChanged(event)); } E::CallGrantsUpdated(event) => { core.update_call_grants(event.current_grants); @@ -363,7 +425,9 @@ pub(super) async fn handle_event( } } } - let _ = core.events_tx.send(CallEvent::CallGrantsUpdated(event)); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::CallGrantsUpdated(event)); } E::IceRestart(event) => { let peer_type = @@ -371,12 +435,16 @@ pub(super) async fn handle_event( if peer_type == PeerType::PublisherUnspecified { core.restart_publisher_ice().await?; } - let _ = core.events_tx.send(CallEvent::IceRestarted(peer_type)); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::IceRestarted(peer_type)); } E::Error(err) => { let join_err = SfuJoinError::from_event(err.error, err.reconnect_strategy); let strategy = ReconnectStrategy::from_proto(err.reconnect_strategy); - let _ = core.events_tx.send(CallEvent::Error(join_err.clone())); + let _ = core + .sfu_events_tx + .send(SfuCallEvent::Error(join_err.clone())); if let Some(strategy) = strategy { core.trigger_reconnect(context.generation, strategy, join_err.message.clone()); } @@ -391,8 +459,12 @@ pub(super) async fn handle_event( E::ParticipantMigrationComplete(_) => { core.complete_migration(context.generation); } - E::CallEnded(_) => { - let _ = core.events_tx.send(CallEvent::CallEnded); + E::CallEnded(event) => { + let _ = core.sfu_events_tx.send(SfuCallEvent::CallEnded { + reason: models::CallEndedReason::try_from(event.reason) + .unwrap_or(models::CallEndedReason::Unspecified), + }); + core.end_call(context.generation); } E::PublisherAnswer(_) | E::JoinResponse(_) => { tracing::debug!("stream.rtc.unexpected_handshake_event"); diff --git a/src/rtc/join/lifecycle.rs b/src/rtc/join/lifecycle.rs index 0a485c6..2780189 100644 --- a/src/rtc/join/lifecycle.rs +++ b/src/rtc/join/lifecycle.rs @@ -17,6 +17,10 @@ impl RtcCore { ) -> Result<()> { ensure_crypto_provider(); let generation = self.begin_join()?; + let _attempt = JoinAttempt { + core: self, + generation, + }; // A fresh unified session id for this join lifecycle; reused across // reconnects so the dashboard correlates the participant end-to-end. { @@ -640,26 +644,78 @@ impl RtcCore { impl RtcCore { /// Leave the call: send `leave`, close the PeerConnections and WebSocket, /// abort background tasks. Succeeds from any state, including `Joining` - /// (JS: force to a leaving state rather than waiting for `JOINED`). - pub async fn leave(&self, reason: impl Into) -> Result<()> { + /// (JS: force to a leaving state rather than waiting for `JOINED`). The + /// teardown runs in a runtime task, so it finishes when this future is + /// dropped. + pub async fn leave(self: &Arc, reason: impl Into) -> Result<()> { let reason = reason.into(); let generation = self.cancel_generation(); + let left = LeftCall { + core: self, + generation, + }; - let connection = self.connection.lock().await.take(); - if let Some(connection) = connection { - let session_id = connection.session_id.clone(); - // Record the leave reason so the final `SendStats` (drained by - // `teardown`) carries the end-of-call event (JS `call.leaveReason`). - connection.signal.trace("call.leaveReason", json!(reason)); - { - let mut sender = connection.sfu_sender.lock().await; - let _ = sender.send_leave(session_id, &reason).await; - let _ = sender.close().await; + let core = self.clone(); + let teardown = self.spawn_runtime_task(async move { + // A join that starts after a dropped leave owns a later connection. + let connection = core + .connection + .lock() + .await + .take_if(|connection| connection.generation < generation); + if let Some(connection) = connection { + let session_id = connection.session_id.clone(); + // Record the leave reason so the final `SendStats` (drained by + // `teardown`) carries the end-of-call event (JS `call.leaveReason`). + connection.signal.trace("call.leaveReason", json!(reason)); + { + let mut sender = connection.sfu_sender.lock().await; + let _ = sender.send_leave(session_id, &reason).await; + let _ = sender.close().await; + } + connection.teardown().await; } - connection.teardown().await; + }); + if let Err(error) = teardown.await + && error.is_panic() + { + std::panic::resume_unwind(error.into_panic()); + } + drop(left); + self.stop_coordinator_events(generation).await; + Ok(()) + } + + /// The SFU or the coordinator reported the end of the call: leave the call + /// once. + pub(super) fn end_call(self: &Arc, generation: u64) { + if !self.is_generation_current(generation) { + return; + } + let already_ended = std::mem::replace( + &mut self + .call_state + .lock() + .unwrap_or_else(|error| error.into_inner()) + .ended, + true, + ); + if already_ended { + return; } + let this = self.clone(); + // Not a generation task: `leave` ends the generation. + std::mem::drop(self.spawn_runtime_task(async move { + if this.is_generation_current(generation) { + let _ = this.leave("call ended").await; + } + })); + } + + /// Clear the call state of `generation` and set `Left`. + fn finish_leave(&self, generation: u64) { { - // A join that started during the awaits above owns these fields. + // A join that started during the leave owns these fields. let lifecycle = self.lifecycle.lock().unwrap_or_else(|e| e.into_inner()); if lifecycle.generation == generation { self.participants @@ -674,6 +730,10 @@ impl RtcCore { .lock() .unwrap_or_else(|e| e.into_inner()) .clear(); + self.delivered_tracks + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clear(); self.own_capabilities .lock() .unwrap_or_else(|e| e.into_inner()) @@ -685,8 +745,45 @@ impl RtcCore { } } self.set_state_if_current(generation, CallingState::Left); - self.stop_coordinator_events(generation).await; - Ok(()) + } +} + +/// Abandons a join whose future is dropped while it is still joining. The join +/// cannot await its cleanup there, so a new generation stops its tasks, and +/// `Idle` allows the next join. +struct JoinAttempt<'a> { + core: &'a RtcCore, + generation: u64, +} + +impl Drop for JoinAttempt<'_> { + fn drop(&mut self) { + { + let mut lifecycle = self + .core + .lifecycle + .lock() + .unwrap_or_else(|e| e.into_inner()); + if lifecycle.generation != self.generation || lifecycle.state != CallingState::Joining { + return; + } + lifecycle.generation = lifecycle.generation.wrapping_add(1); + lifecycle.set_state(CallingState::Idle, &self.core.client_events_tx); + } + self.core.lifecycle_changed.notify_waiters(); + } +} + +/// Finishes a `leave` on drop, so a `leave` future that is dropped early still +/// leaves the call. The coordinator tasks stop on the generation change. +struct LeftCall<'a> { + core: &'a RtcCore, + generation: u64, +} + +impl Drop for LeftCall<'_> { + fn drop(&mut self) { + self.core.finish_leave(self.generation); } } @@ -725,7 +822,7 @@ impl RtcCore { let cid = self.cid(); let local_user_id = user_id.to_owned(); - let sender = self.events_tx.clone(); + let sender = self.coordinator_events_tx.clone(); let event_core = self.clone(); let event_task = self.spawn_generation_task(generation, async move { loop { @@ -737,7 +834,11 @@ impl RtcCore { event_core .apply_permissions_updated(&event, &local_user_id) .await; - let _ = sender.send(CallEvent::Coordinator(event)); + let ended = event.event_type == "call.ended"; + let _ = sender.send(event); + if ended { + event_core.end_call(generation); + } } Ok(Some(_)) => {} Ok(None) => { diff --git a/src/rtc/join/mod.rs b/src/rtc/join/mod.rs index a14d292..5cdd164 100644 --- a/src/rtc/join/mod.rs +++ b/src/rtc/join/mod.rs @@ -8,7 +8,8 @@ //! backoff, unrecoverable abort, and SFU switching via `migrating_from`; //! - the SFU WebSocket handshake (`JoinRequest` → `JoinResponse`), subscriber //! answer negotiation, and ICE trickle; -//! - a typed [`CallEvent`] broadcast stream (participant joined/left, tracks, …); +//! - typed broadcast streams by source: [`SfuCallEvent`], coordinator events, +//! and [`ClientCallEvent`]; //! - the reconnect state machine (`RtcCore::run_reconnect`) driven by the pure //! decision logic in [`super::reconnect`], with dedup, the rejoin rate limiter, //! the ICE / negotiation limits, the disconnection timeout, and the @@ -35,6 +36,7 @@ use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering} use std::sync::{Arc, Mutex as StdMutex, Weak}; use std::time::{Duration, Instant}; +use indexmap::IndexMap; use tokio::sync::{Mutex as TokioMutex, Notify, broadcast}; use tokio::task::JoinHandle; use url::Url; @@ -65,7 +67,9 @@ use super::reconnect::{ use super::sfu::signal::SignalClient; use super::sfu::ws::{self, SfuReceiver, SfuSender}; use super::stats::{self, StatsReporter, StatsReporterParts}; -use super::subscriptions::{SubscriptionConfig, SubscriptionTarget, TrackKey}; +use super::subscriptions::{ + DEFAULT_VIDEO_DIMENSION, SubscriptionConfig, SubscriptionTarget, TrackKey, +}; use super::tracer::Tracer; use super::tracks::{LocalTrack, RemoteParticipant, RemoteTrack}; @@ -174,26 +178,26 @@ pub enum CallingState { Left, } -/// A typed SFU event delivered on the [`Call`](crate::Call) event stream. +/// An event from the SFU, delivered by [`RtcCore::sfu_events`]. #[derive(Debug, Clone)] #[non_exhaustive] -pub enum CallEvent { +pub enum SfuCallEvent { /// A participant joined the call. ParticipantJoined(models::Participant), /// A participant left the call. ParticipantLeft(models::Participant), /// A participant's user data changed. ParticipantUpdated(models::Participant), - /// A call-scoped coordinator WebSocket event. - Coordinator(CoordinatorEvent), /// A track was published (audio/video/screenshare). TrackPublished { /// The publisher's user id. user_id: String, /// The publisher's session id. session_id: String, - /// The `TrackType` value. - track_type: i32, + /// The published track type. + track_type: TrackType, + /// The publisher, sent by the SFU only in large calls. + participant: Option, }, /// A track was unpublished. TrackUnpublished { @@ -201,8 +205,12 @@ pub enum CallEvent { user_id: String, /// The publisher's session id. session_id: String, - /// The `TrackType` value. - track_type: i32, + /// The unpublished track type. + track_type: TrackType, + /// Why the track was unpublished. + cause: models::TrackUnpublishReason, + /// The publisher, sent by the SFU only in large calls. + participant: Option, }, /// The dominant speaker changed. DominantSpeakerChanged { @@ -236,12 +244,22 @@ pub enum CallEvent { PublishQualityChanged(event::ChangePublishQuality), /// The current participant's publishing grants changed. CallGrantsUpdated(event::CallGrantsUpdated), - /// An SFU-directed ICE restart completed for a peer connection. + /// The SFU requested an ICE restart. The SDK restarts only the publisher. IceRestarted(PeerType), /// The SFU reported an error for this participant. Error(SfuJoinError), - /// The call ended. - CallEnded, + /// The SFU ended the call. The SDK then leaves the call. + CallEnded { + /// Why the call ended. + reason: models::CallEndedReason, + }, +} + +/// An event that the SDK itself produces, delivered by +/// [`RtcCore::client_events`]. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub enum ClientCallEvent { /// The connection state changed. CallingStateChanged(CallingState), } @@ -279,10 +297,10 @@ struct Lifecycle { impl Lifecycle { /// Call with the lifecycle lock held, so events arrive in the order of the /// state changes. - fn set_state(&mut self, next: CallingState, events: &broadcast::Sender) { + fn set_state(&mut self, next: CallingState, events: &broadcast::Sender) { if self.state != next { self.state = next; - let _ = events.send(CallEvent::CallingStateChanged(next)); + let _ = events.send(ClientCallEvent::CallingStateChanged(next)); } } } @@ -488,7 +506,9 @@ pub struct RtcCore { api_key: String, call_type: String, call_id: String, - events_tx: broadcast::Sender, + sfu_events_tx: broadcast::Sender, + coordinator_events_tx: broadcast::Sender, + client_events_tx: broadcast::Sender, lifecycle: StdMutex, lifecycle_changed: Notify, connection: TokioMutex>, @@ -514,10 +534,15 @@ pub struct RtcCore { /// Tracks the caller explicitly dropped (unsubscribed); never re-subscribed /// until the publisher republishes them. manual_unsub: StdMutex>, + /// The id of the latest [`RemoteTrack`] delivered for each track. Only a + /// drop of that track unsubscribes. + delivered_tracks: StdMutex>, + next_remote_track_id: AtomicU64, /// Exact per-session subscriptions, or `None` while using the coarse policy. manual_subscriptions: StdMutex>>, - /// Known participants keyed by session id (correlation + subscription build). - participants: StdMutex>, + /// Known participants keyed by session id (correlation + subscription build), + /// in the order the call learned about them. + participants: StdMutex>, /// Call-level state supplied by join and incremental SFU events. call_state: StdMutex, /// Serialized publisher negotiation and retryable local publication state. @@ -545,7 +570,7 @@ impl Drop for ReconnectClaim { impl RtcCore { /// Build a fresh (idle) core for a call handle. pub(crate) fn new(client: Arc, call_type: String, call_id: String) -> Arc { - let (events_tx, _rx) = broadcast::channel(256); + let event_capacity = client.call_event_capacity(); Arc::new(Self { api_key: client.api_key().to_owned(), client, @@ -554,7 +579,9 @@ impl RtcCore { token_refresh: TokioMutex::new(()), call_type, call_id, - events_tx, + sfu_events_tx: broadcast::channel(event_capacity).0, + coordinator_events_tx: broadcast::channel(event_capacity).0, + client_events_tx: broadcast::channel(event_capacity).0, lifecycle: StdMutex::new(Lifecycle { state: CallingState::Idle, generation: 0, @@ -578,8 +605,10 @@ impl RtcCore { sub_config: StdMutex::new(SubscriptionConfig::default()), subs_active: AtomicBool::new(false), manual_unsub: StdMutex::new(HashSet::new()), + delivered_tracks: StdMutex::new(HashMap::new()), + next_remote_track_id: AtomicU64::new(0), manual_subscriptions: StdMutex::new(None), - participants: StdMutex::new(HashMap::new()), + participants: StdMutex::new(IndexMap::new()), call_state: StdMutex::new(CallStateCache::default()), media: TokioMutex::new(MediaState::default()), active_subs: StdMutex::new(Vec::new()), @@ -665,7 +694,7 @@ impl RtcCore { if guard.generation != generation { return false; } - guard.set_state(next, &self.events_tx); + guard.set_state(next, &self.client_events_tx); true } @@ -724,7 +753,7 @@ impl RtcCore { match guard.state { CallingState::Idle | CallingState::Left => { guard.generation = guard.generation.wrapping_add(1); - guard.set_state(CallingState::Joining, &self.events_tx); + guard.set_state(CallingState::Joining, &self.client_events_tx); guard.generation_publish_options = guard.publish_options; guard.failure_limits = FailureLimits::default(); guard.rate_limiter = SlidingWindowRateLimiter::rejoin_default(); @@ -873,9 +902,26 @@ impl RtcCore { Ok(()) } - /// Subscribe to the typed event stream. - pub fn subscribe(&self) -> broadcast::Receiver { - self.events_tx.subscribe() + /// Subscribe to the events from the SFU. A receiver gets only events sent + /// after it subscribes; subscribe before the join to get the join events. + pub fn sfu_events(&self) -> broadcast::Receiver { + self.sfu_events_tx.subscribe() + } + + /// Subscribe to the call-scoped coordinator events. A receiver gets only + /// events sent after it subscribes. + pub fn coordinator_events(&self) -> broadcast::Receiver { + self.coordinator_events_tx.subscribe() + } + + /// Subscribe to the events that the SDK itself produces. A receiver gets + /// only events sent after it subscribes. + /// + /// `CallingStateChanged(Left)` is the reliable end of the call: the SDK + /// leaves on the SFU `call_ended` or the coordinator `call.ended`, and the + /// other one may then not arrive. + pub fn client_events(&self) -> broadcast::Receiver { + self.client_events_tx.subscribe() } fn user_request_query(&self) -> Option> { @@ -912,34 +958,6 @@ impl RtcCore { (!token.is_empty()).then(|| (token.clone(), query)) } - /// Register a callback for typed call events. - /// - /// Rust callers receive the full [`CallEvent`] enum and can pattern-match - /// the variants they need. Pass the returned handle to [`Self::off`]. - pub fn on(&self, callback: F) -> tokio::task::AbortHandle - where - F: Fn(CallEvent) + Send + 'static, - { - let mut events = self.subscribe(); - tokio::spawn(async move { - loop { - match events.recv().await { - Ok(event) => callback(event), - Err(broadcast::error::RecvError::Lagged(skipped)) => { - tracing::warn!(skipped, "stream.rtc.event_handler_lagged"); - } - Err(broadcast::error::RecvError::Closed) => break, - } - } - }) - .abort_handle() - } - - /// Remove an event callback registered with [`Self::on`]. - pub fn off(&self, handler: &tokio::task::AbortHandle) { - handler.abort(); - } - /// The cached stats options from the last coordinator join. pub fn stats_options(&self) -> StatsOptions { self.stats_options diff --git a/src/rtc/join/participants.rs b/src/rtc/join/participants.rs index 537e7d7..54f5045 100644 --- a/src/rtc/join/participants.rs +++ b/src/rtc/join/participants.rs @@ -23,12 +23,17 @@ pub(super) struct CallStateCache { pub(super) started_at: Option, pub(super) e2ee_enabled: bool, pub(super) current_grants: Option, + /// Set by the first report of the call end. + pub(super) ended: bool, + /// The local session id of the last join response. + pub(super) local_session_id: String, } impl RtcCore { /// A snapshot of the participants currently known in the call (including this - /// session), built from the SFU participant state. Updated as - /// `ParticipantJoined` / `ParticipantLeft` events arrive. + /// session), built from the SFU participant state, in the order the call + /// learned about them. Updated as `ParticipantJoined` / `ParticipantLeft` + /// events arrive. pub fn participants(&self) -> Vec { let participants = self.participants.lock().unwrap_or_else(|e| e.into_inner()); participants @@ -84,7 +89,8 @@ impl RtcCore { } /// Replace the participants from an authoritative SFU join response when - /// its lifecycle generation is still active. + /// its lifecycle generation is still active, and report the participants + /// that joined, changed, or left since the previous join response. pub(super) fn apply_join_call_state_if_current( &self, generation: u64, @@ -98,19 +104,36 @@ impl RtcCore { } let state = call_state.unwrap_or_default(); let joined = state.participants.clone(); - *self - .call_state - .lock() - .unwrap_or_else(|error| error.into_inner()) = CallStateCache { - participant_count: state.participant_count.unwrap_or_default(), - pins: state.pins, - started_at: state.started_at, - e2ee_enabled: state.e2ee_enabled, - current_grants: None, - }; - { + let previous_session_id = std::mem::replace( + &mut *self + .call_state + .lock() + .unwrap_or_else(|error| error.into_inner()), + CallStateCache { + participant_count: state.participant_count.unwrap_or_default(), + pins: state.pins, + started_at: state.started_at, + e2ee_enabled: state.e2ee_enabled, + current_grants: None, + ended: false, + local_session_id: session_id.to_owned(), + }, + ) + .local_session_id; + let mut previous = { let mut participants = self.participants.lock().unwrap_or_else(|e| e.into_inner()); - participants.clear(); + let previous = std::mem::take(&mut *participants); + // Known participants keep the order in which the call learned them. + let current: HashSet<&str> = joined + .iter() + .map(|participant| participant.session_id.as_str()) + .chain([session_id]) + .collect(); + for id in previous.keys() { + if current.contains(id.as_str()) { + participants.insert(id.clone(), ParticipantState::default()); + } + } let me = participants.entry(session_id.to_owned()).or_default(); me.user_id = user_id.to_owned(); me.session_id = session_id.to_owned(); @@ -136,12 +159,29 @@ impl RtcCore { .published .extend(participant.published_tracks.iter().copied()); } - } + previous + }; + // The local sessions, before and after a REJOIN, produce no events. + let is_local = |id: &str| id == session_id || id == previous_session_id; for participant in joined { - if participant.session_id != session_id { + let known = previous.shift_remove(&participant.session_id); + if is_local(&participant.session_id) { + continue; + } + let event = match known { + None => SfuCallEvent::ParticipantJoined(participant), + Some(entry) if entry.participant != participant => { + SfuCallEvent::ParticipantUpdated(participant) + } + Some(_) => continue, + }; + let _ = self.sfu_events_tx.send(event); + } + for (id, entry) in previous { + if !is_local(&id) { let _ = self - .events_tx - .send(CallEvent::ParticipantJoined(participant)); + .sfu_events_tx + .send(SfuCallEvent::ParticipantLeft(entry.participant)); } } true @@ -168,7 +208,7 @@ impl RtcCore { self.participants .lock() .unwrap_or_else(|e| e.into_inner()) - .remove(session_id); + .shift_remove(session_id); } /// Record a newly-published track for a participant, learning the @@ -262,11 +302,16 @@ impl RtcCore { } } - pub(super) fn update_participant_count(&self, participant_count: models::ParticipantCount) { - self.call_state + /// Returns whether the stored count changed. + pub(super) fn update_participant_count( + &self, + participant_count: models::ParticipantCount, + ) -> bool { + let mut state = self + .call_state .lock() - .unwrap_or_else(|error| error.into_inner()) - .participant_count = participant_count; + .unwrap_or_else(|error| error.into_inner()); + std::mem::replace(&mut state.participant_count, participant_count) != participant_count } pub(super) fn update_pins(&self, pins: Vec) { diff --git a/src/rtc/join/publication.rs b/src/rtc/join/publication.rs index c50e424..2b43bc5 100644 --- a/src/rtc/join/publication.rs +++ b/src/rtc/join/publication.rs @@ -27,6 +27,9 @@ pub(super) struct LocalPublication { pub(super) struct MediaState { pub(super) publications: Vec, pub(super) publish_quality: HashMap<(i32, i32), Vec>, + /// The publish option id and stopped track of each sender that + /// `stop_publish` kept in the publisher envelope. + retired: Vec<(i32, LocalTrack)>, } impl MediaState { @@ -92,6 +95,28 @@ impl MediaState { Some(publication.track) } + /// Remove the publication of a stopped track whose sender stays in the + /// publisher envelope, and keep that sender for a later publish. + pub(super) fn retire(&mut self, track_id: &str) -> Option { + let publish_option_id = self.publications[self.position(track_id)?].publish_option_id; + let track = self.remove(track_id)?; + self.retired.push((publish_option_id, track.clone())); + Some(track) + } + + /// Take the stopped track of the latest sender that [`Self::retire`] kept + /// for this kind of track. + pub(super) fn take_retired( + &mut self, + track_type: TrackType, + publish_option_id: i32, + ) -> Option { + let position = self.retired.iter().rposition(|(option_id, retired)| { + retired.track_type() == track_type && *option_id == publish_option_id + })?; + Some(self.retired.remove(position).1) + } + pub(super) fn refresh_publish_options( &mut self, publish_options: &[models::PublishOption], diff --git a/src/rtc/join/publish.rs b/src/rtc/join/publish.rs index f4221a8..94cdd04 100644 --- a/src/rtc/join/publish.rs +++ b/src/rtc/join/publish.rs @@ -63,9 +63,31 @@ impl RtcCore { }; let mut publisher_rtcp_tasks = Vec::new(); if status.is_none() { - publisher_rtcp_tasks = - match publisher::add_transceiver_for_track(&publisher, &track, &publish_options) - .await + // Reuse the audio sender that `stop_publish` kept, as JS `replaceTrack` + // does: a second audio m-line fails the SFU negotiation. Video RTCP + // readers keep the old track, and simulcast cannot be replaced. + let retired = if matches!( + track.track_type(), + TrackType::Audio | TrackType::ScreenShareAudio + ) { + media.take_retired(track.track_type(), publish_option_id) + } else { + None + }; + let reused = match retired { + Some(retired) => { + publisher::replace_retired_track(&publisher, &retired, &track, &publish_options) + .await? + } + None => false, + }; + if !reused { + publisher_rtcp_tasks = match publisher::add_transceiver_for_track( + &publisher, + &track, + &publish_options, + ) + .await { Ok(tasks) => tasks, Err(error) => { @@ -77,7 +99,11 @@ impl RtcCore { return Err(error); } }; + } media.begin_publish(track.clone(), publish_option_id); + if reused { + media.set_status(&track_id, PublicationStatus::PendingPublishMute); + } if let Some(layers) = media .publish_quality .get(&(publish_option_id, track.track_type() as i32)) @@ -123,7 +149,9 @@ impl RtcCore { .user_id .clone(); self.add_published_track(&user_id, &session_id, track.track_type() as i32, None); - track.start_media(); + if publisher.connection_state() == RTCPeerConnectionState::Connected { + track.start_audio_pacing().await; + } signal .update_mute_states(signal::UpdateMuteStatesRequest { session_id: session_id.clone(), @@ -138,6 +166,24 @@ impl RtcCore { Ok(()) } + /// Pace the published audio only while the current publisher is connected. + pub(super) async fn sync_audio_pacing(&self) { + let media = self.media.lock().await; + let connected = self + .publisher_handles() + .await + .is_some_and(|(publisher, ..)| { + publisher.connection_state() == RTCPeerConnectionState::Connected + }); + for track in media.active_tracks() { + if connected { + track.start_audio_pacing().await; + } else { + track.pause_audio_pacing(); + } + } + } + pub(super) async fn register_publisher_tasks(&self, tasks: Vec>) { if tasks.is_empty() { return; @@ -198,7 +244,7 @@ impl RtcCore { if muted { self.remove_published_track(&session_id, track_type as i32); } - if let Some(removed) = media.remove(&track_id) { + if let Some(removed) = media.retire(&track_id) { removed.stop(); } Ok(()) diff --git a/src/rtc/join/reconnect_runtime.rs b/src/rtc/join/reconnect_runtime.rs index 03cff94..74773aa 100644 --- a/src/rtc/join/reconnect_runtime.rs +++ b/src/rtc/join/reconnect_runtime.rs @@ -4,8 +4,8 @@ use super::*; impl RtcCore { /// Re-publish every local track on the fresh publisher PC after a - /// REJOIN/MIGRATE. The tracks are the same `Arc`-backed handles, so the pacer - /// keeps running and simply binds to the new PeerConnection. + /// REJOIN/MIGRATE. The tracks are the same `Arc`-backed handles, so they bind + /// to the new PeerConnection and pacing resumes when it connects. pub(super) async fn restore_published_tracks(&self) -> Result<()> { let mut media = self.media.lock().await; let capabilities = self @@ -103,7 +103,6 @@ impl RtcCore { } else { self.add_published_track(&user_id, session_id, track.track_type() as i32, None); } - track.start_media(); } Ok(()) } diff --git a/src/rtc/join/subscriptions_runtime.rs b/src/rtc/join/subscriptions_runtime.rs index 1be0f62..d3ee936 100644 --- a/src/rtc/join/subscriptions_runtime.rs +++ b/src/rtc/join/subscriptions_runtime.rs @@ -43,19 +43,25 @@ impl RtcCore { /// Enable or disable incoming video for every remote participant. pub async fn set_incoming_video_enabled(&self, enabled: bool) -> Result<()> { - let config = { - let mut config = self.sub_config.lock().unwrap_or_else(|e| e.into_inner()); - config.video = enabled; - config.video_dimension = None; - *config - }; + { + let mut guard = self.sub_config.lock().unwrap_or_else(|e| e.into_inner()); + let config = &mut *guard; + for rule in std::iter::once(&mut config.default).chain(config.role_filters.values_mut()) + { + rule.track_types + .retain(|track_type| *track_type != TrackType::Video); + if enabled { + rule.track_types.push(TrackType::Video); + } + } + } *self .manual_subscriptions .lock() .unwrap_or_else(|error| error.into_inner()) = None; self.subs_active.store(true, Ordering::SeqCst); self.recompute_subscriptions().await?; - tracing::debug!(enabled = config.video, "stream.rtc.incoming_video_updated"); + tracing::debug!(enabled, "stream.rtc.incoming_video_updated"); Ok(()) } @@ -85,7 +91,11 @@ impl RtcCore { } }; - let config = *self.sub_config.lock().unwrap_or_else(|e| e.into_inner()); + let config = self + .sub_config + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); let targets = self .manual_subscriptions .lock() @@ -116,41 +126,21 @@ impl RtcCore { user_id: entry.user_id.clone(), session_id: entry.session_id.clone(), track_type: target.track_type as i32, - dimension: target.dimension.and_then(|(width, height)| { - is_video_type(target.track_type) - .then_some(models::VideoDimension { width, height }) + dimension: is_video_type(target.track_type).then(|| { + let (width, height) = + target.dimension.unwrap_or(DEFAULT_VIDEO_DIMENSION); + models::VideoDimension { width, height } }), }); } } else { - for entry in participants.values() { - if entry.session_id == session_id { - continue; - } - for &tt_i in &entry.published { - let Ok(track_type) = TrackType::try_from(tt_i) else { - continue; - }; - if !config.matches(track_type) - || manual.contains(&TrackKey::new(entry.session_id.clone(), track_type)) - { - continue; - } - let dimension = if is_video_type(track_type) { - config - .video_dimension - .map(|(width, height)| models::VideoDimension { width, height }) - } else { - None - }; - tracks.push(signal::TrackSubscriptionDetails { - user_id: entry.user_id.clone(), - session_id: entry.session_id.clone(), - track_type: tt_i, - dimension, - }); - } - } + tracks = config.track_subscriptions( + participants + .values() + .map(|entry| &entry.participant) + .filter(|participant| participant.session_id != session_id), + &manual, + ); } } tracks.sort_by(|a, b| { @@ -230,13 +220,21 @@ impl RtcCore { return; } let key = TrackKey::new(participant.session_id.clone(), track_type); + let track_id = self.next_remote_track_id.fetch_add(1, Ordering::SeqCst); + self.delivered_tracks + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(key.clone(), track_id); let weak = Arc::downgrade(&self); + // The caller can drop the track on a thread without a runtime. + let runtime = tokio::runtime::Handle::current(); let on_drop = Box::new(move || { if let Some(core) = weak.upgrade() { + let _runtime = runtime.enter(); let task_core = core.clone(); std::mem::drop(core.spawn_generation_task(generation, async move { task_core - .on_remote_track_dropped(generation, connection_epoch, key) + .on_remote_track_dropped(generation, connection_epoch, key, track_id) .await; })); } @@ -251,6 +249,7 @@ impl RtcCore { generation: u64, connection_epoch: u64, key: TrackKey, + track_id: u64, ) { { let connection = self.connection.lock().await; @@ -259,6 +258,15 @@ impl RtcCore { }) { return; } + let mut delivered = self + .delivered_tracks + .lock() + .unwrap_or_else(|e| e.into_inner()); + if delivered.get(&key) != Some(&track_id) { + return; + } + delivered.remove(&key); + drop(delivered); self.manual_unsub .lock() .unwrap_or_else(|e| e.into_inner()) diff --git a/src/rtc/join/tests.rs b/src/rtc/join/tests.rs index b508b4d..4de2bcb 100644 --- a/src/rtc/join/tests.rs +++ b/src/rtc/join/tests.rs @@ -3,7 +3,7 @@ use super::*; use crate::client::ClientConfig; use crate::rtc::{ - ClientPublishOptions, LocalAudioTrack, LocalVideoTrack, LocalVideoTrackConfig, + ClientPublishOptions, LocalAudioTrack, LocalVideoTrack, LocalVideoTrackConfig, PcmFrame, PreferredVideoCodec, publish_options::H264_FMTP, }; use std::io::{Read, Write}; @@ -162,6 +162,13 @@ async fn fake_sfu( /// A local coordinator WebSocket that sends `connection.ok`. It returns the /// REST base URL and a task that ends when the client socket closes. async fn fake_coordinator() -> (String, tokio::task::JoinHandle<()>) { + fake_coordinator_sending(Vec::new()).await +} + +/// [`fake_coordinator`] that also sends `events` after `connection.ok`. +async fn fake_coordinator_sending( + events: Vec, +) -> (String, tokio::task::JoinHandle<()>) { use futures_util::{SinkExt, StreamExt}; use tokio_tungstenite::tungstenite::Message; @@ -187,6 +194,12 @@ async fn fake_coordinator() -> (String, tokio::task::JoinHandle<()>) { )) .await .expect("send connection.ok"); + for event in events { + socket + .send(Message::Text(event.to_string().into())) + .await + .expect("send coordinator event"); + } while let Some(Ok(_)) = socket.next().await {} }); (format!("ws://{address}"), server) @@ -236,6 +249,21 @@ async fn establish_fake( (connection, sfu) } +/// The event loop context of the current `connection`. +fn event_context(core: &Arc, connection: &Connection) -> EventLoopContext { + EventLoopContext { + core: core.clone(), + subscriber: connection.subscriber.clone(), + publisher: connection.publisher.clone(), + signal: connection.signal.clone(), + session_id: connection.session_id.clone(), + pending_ice: connection.pending_ice.clone(), + generation: connection.generation, + ws_healthy: connection.ws_healthy.clone(), + reconnect_enabled: connection.reconnect_enabled.clone(), + } +} + /// The event loop context of `connection` after a migration detached it. fn detached_context(core: &Arc, connection: &Connection) -> EventLoopContext { connection.reconnect_enabled.store(false, Ordering::SeqCst); @@ -368,6 +396,335 @@ async fn leave_tears_down_the_stored_connection() { assert_eq!(spawned, completed); } +#[tokio::test] +async fn call_ended_from_the_coordinator_is_forwarded_and_leaves_the_call() { + let (coordinator_ws_url, _coordinator) = fake_coordinator_sending(vec![ + json!({ "type": "call.ended", "call_cid": "default:test-call" }), + ]) + .await; + let core = test_core_with_config(ClientConfig { + coordinator_ws_url, + ..ClientConfig::default() + }); + let generation = prepare_joined_core(&core, "alice"); + let (connection, sfu) = establish_fake(&core, generation).await; + *core.connection.lock().await = Some(connection); + let mut events = core.coordinator_events(); + let token = core.current_user_token().expect("user token"); + + core.connect_coordinator_events(generation, &token, "alice") + .await + .expect("coordinator events"); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Ok(event) = events.recv().await + && event.event_type == "call.ended" + { + return; + } + } + }) + .await + .expect("call.ended event"); + + wait_for( + Duration::from_secs(2), + || core.state() == CallingState::Left, + "the ended call is left", + ) + .await; + let requests = requests_until_close(sfu).await; + assert!(requests.iter().any(|request| matches!( + request.request_payload, + Some(event::sfu_request::RequestPayload::LeaveCallRequest(_)) + ))); +} + +#[tokio::test] +async fn sfu_call_ended_is_forwarded_with_its_reason_and_leaves_once() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, sfu) = establish_fake(&core, generation).await; + let context = event_context(&core, &connection); + *core.connection.lock().await = Some(connection); + let mut events = core.sfu_events(); + + // Neither call yields, so the spawned leave cannot run between them. + for _ in 0..2 { + connection::handle_event( + &context, + sfu_event::EventPayload::CallEnded(event::CallEnded { + reason: models::CallEndedReason::Kicked as i32, + }), + ) + .await + .expect("handle SFU call ended"); + } + + wait_for( + Duration::from_secs(2), + || core.state() == CallingState::Left, + "the ended call is left", + ) + .await; + let mut forwarded = 0; + while let Ok(event) = events.try_recv() { + if matches!( + event, + SfuCallEvent::CallEnded { + reason: models::CallEndedReason::Kicked + } + ) { + forwarded += 1; + } + } + assert_eq!(forwarded, 2, "each SFU call_ended is forwarded"); + let leaves = requests_until_close(sfu) + .await + .into_iter() + .filter(|request| { + matches!( + request.request_payload, + Some(event::sfu_request::RequestPayload::LeaveCallRequest(_)) + ) + }) + .count(); + assert_eq!(leaves, 1, "the call is left once"); +} + +#[tokio::test] +async fn each_event_goes_only_to_the_stream_of_its_source() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, _sfu) = establish_fake(&core, generation).await; + let context = event_context(&core, &connection); + *core.connection.lock().await = Some(connection); + let mut sfu = core.sfu_events(); + let mut client = core.client_events(); + let mut coordinator = core.coordinator_events(); + + connection::handle_event( + &context, + sfu_event::EventPayload::PinsUpdated(event::PinsChanged::default()), + ) + .await + .expect("handle pins"); + assert!(core.set_state_if_current(generation, CallingState::Reconnecting)); + + assert!(matches!(sfu.try_recv(), Ok(SfuCallEvent::PinsUpdated(_)))); + assert!(sfu.try_recv().is_err()); + assert!(matches!( + client.try_recv(), + Ok(ClientCallEvent::CallingStateChanged( + CallingState::Reconnecting + )) + )); + assert!(client.try_recv().is_err()); + assert!(coordinator.try_recv().is_err()); +} + +#[test] +fn a_call_event_stream_keeps_the_configured_number_of_events() { + let core = test_core_with_config(ClientConfig { + call_event_capacity: 2, + ..ClientConfig::default() + }); + let generation = core.begin_join().expect("test generation"); + let mut events = core.client_events(); + for state in [ + CallingState::Joined, + CallingState::Reconnecting, + CallingState::Joined, + ] { + assert!(core.set_state_if_current(generation, state)); + } + + assert!(matches!( + events.try_recv(), + Err(broadcast::error::TryRecvError::Lagged(1)) + )); +} + +#[tokio::test] +async fn track_events_report_the_track_type_cause_and_participant() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, _sfu) = establish_fake(&core, generation).await; + let context = event_context(&core, &connection); + *core.connection.lock().await = Some(connection); + let mut events = core.sfu_events(); + let bob = models::Participant { + user_id: "bob".to_owned(), + session_id: "bob-session".to_owned(), + ..Default::default() + }; + + connection::handle_event( + &context, + sfu_event::EventPayload::TrackPublished(event::TrackPublished { + user_id: "bob".to_owned(), + session_id: "bob-session".to_owned(), + r#type: TrackType::Audio as i32, + participant: Some(bob.clone()), + }), + ) + .await + .expect("handle track published"); + + let Ok(SfuCallEvent::TrackPublished { + track_type, + participant, + .. + }) = events.try_recv() + else { + panic!("expected a track published event"); + }; + assert_eq!(track_type, TrackType::Audio); + assert_eq!(participant.as_ref(), Some(&bob)); + + connection::handle_event( + &context, + sfu_event::EventPayload::TrackUnpublished(event::TrackUnpublished { + user_id: "bob".to_owned(), + session_id: "bob-session".to_owned(), + r#type: TrackType::Audio as i32, + cause: models::TrackUnpublishReason::Moderation as i32, + participant: Some(bob.clone()), + }), + ) + .await + .expect("handle track unpublished"); + + let Ok(SfuCallEvent::TrackUnpublished { + track_type, + cause, + participant, + .. + }) = events.try_recv() + else { + panic!("expected a track unpublished event"); + }; + assert_eq!(track_type, TrackType::Audio); + assert_eq!(cause, models::TrackUnpublishReason::Moderation); + assert_eq!(participant, Some(bob)); +} + +#[tokio::test] +async fn participant_count_event_is_sent_only_when_the_count_changes() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, _sfu) = establish_fake(&core, generation).await; + let context = event_context(&core, &connection); + *core.connection.lock().await = Some(connection); + let mut events = core.sfu_events(); + + for total in [2, 2, 3] { + connection::handle_event( + &context, + sfu_event::EventPayload::HealthCheckResponse(event::HealthCheckResponse { + participant_count: Some(models::ParticipantCount { + total, + anonymous: 0, + }), + }), + ) + .await + .expect("handle health check response"); + } + + let mut totals = Vec::new(); + while let Ok(event) = events.try_recv() { + if let SfuCallEvent::ParticipantCountChanged(count) = event { + totals.push(count.total); + } + } + assert_eq!(totals, vec![2, 3]); +} + +#[tokio::test] +async fn dropped_join_allows_a_new_join() { + let (coordinator_ws_url, coordinator) = fake_coordinator().await; + // Accepts connections but never answers, so the coordinator join call waits. + let silent = TcpListener::bind("127.0.0.1:0").expect("bind silent server"); + let core = test_core_with_config(ClientConfig { + coordinator_ws_url, + base_url: format!("http://{}", silent.local_addr().expect("silent address")), + ..ClientConfig::default() + }); + let token = crate::token::create_user_token( + b"test-secret", + "alice", + &crate::token::TokenOptions::default(), + ) + .expect("test user token"); + let mut data = JoinCallData::new("alice"); + data.location = Some("test-location".to_owned()); + + let join = tokio::time::timeout(Duration::from_millis(500), core.join(token, data)).await; + assert!(join.is_err(), "join waits for the coordinator"); + + assert_eq!(core.state(), CallingState::Idle); + tokio::time::timeout(Duration::from_secs(2), coordinator) + .await + .expect("coordinator socket closed") + .expect("fake coordinator task"); + core.begin_join().expect("a new join can start"); +} + +#[tokio::test] +async fn dropped_leave_still_leaves_the_call() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, sfu) = establish_fake(&core, generation).await; + let (subscriber, publisher) = (connection.subscriber.clone(), connection.publisher.clone()); + *core.connection.lock().await = Some(connection); + let connection_slot = core.connection.lock().await; + + let leave = tokio::time::timeout(Duration::from_millis(50), core.leave("dropped leave")).await; + assert!(leave.is_err(), "leave waits for the connection lock"); + drop(connection_slot); + + assert_eq!(core.state(), CallingState::Left); + let requests = requests_until_close(sfu).await; + assert!(requests.iter().any(|request| matches!( + request.request_payload, + Some(event::sfu_request::RequestPayload::LeaveCallRequest(_)) + ))); + wait_for( + Duration::from_secs(2), + || { + subscriber.connection_state() == RTCPeerConnectionState::Closed + && publisher.connection_state() == RTCPeerConnectionState::Closed + }, + "closed peer connections", + ) + .await; + core.begin_join().expect("a new join can start"); +} + +#[tokio::test] +async fn a_dropped_leave_keeps_the_connection_of_a_later_join() { + let core = test_core(); + prepare_joined_core(&core, "alice"); + let mut connection_slot = core.connection.lock().await; + let leave = tokio::time::timeout(Duration::from_millis(50), core.leave("dropped leave")).await; + assert!(leave.is_err(), "leave waits for the connection lock"); + + let generation = core.begin_join().expect("a new join can start"); + let (connection, _sfu) = establish_fake(&core, generation).await; + *connection_slot = Some(connection); + let (_, _, completed) = core.runtime_task_snapshot(); + drop(connection_slot); + wait_for( + Duration::from_secs(2), + || core.runtime_task_snapshot().2 > completed, + "the leave task", + ) + .await; + + assert!(core.connection.lock().await.is_some()); +} + #[tokio::test] async fn leave_closes_a_connection_owned_by_a_cancelled_join() { let core = test_core(); @@ -576,6 +933,131 @@ async fn stale_coordinator_stop_keeps_the_current_coordinator() { .expect("fake coordinator task"); } +#[tokio::test] +async fn published_audio_is_paced_only_while_the_publisher_is_connected() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, _sfu) = establish_fake(&core, generation).await; + let publisher = connection.publisher.clone(); + *core.connection.lock().await = Some(connection); + let audio = LocalAudioTrack::opus().expect("opus track"); + let tone = (0..960_u32) + .map(|n| { + (12_000.0 * (std::f64::consts::TAU * 440.0 * f64::from(n) / 48_000.0).sin()) as i16 + }) + .collect::>() + .repeat(50); + audio + .write_pcm(PcmFrame::mono(tone, 48_000)) + .await + .expect("one second fits the queue"); + publisher + .add_track(audio.webrtc_track()) + .await + .expect("add track"); + core.media + .lock() + .await + .begin_publish(LocalTrack::Audio(audio.clone()), 0); + + let (receiver, mut remote) = peer::connect_audio_receiver(&publisher).await; + + tokio::time::timeout(Duration::from_secs(5), remote.recv()) + .await + .expect("paced audio reaches the receiver") + .expect("remote track"); + publisher.close().await.expect("close publisher"); + wait_for( + Duration::from_secs(2), + || !audio.is_pacing(), + "closed publisher pauses pacing", + ) + .await; + let _ = receiver.close().await; + core.leave("test cleanup").await.expect("leave"); +} + +#[tokio::test] +async fn only_the_latest_remote_track_unsubscribes_when_dropped_without_a_runtime() { + let core = test_core(); + let generation = prepare_joined_core(&core, "alice"); + let (connection, _sfu) = establish_fake(&core, generation).await; + let (epoch, reconnect_enabled) = (connection.epoch, connection.reconnect_enabled.clone()); + *core.connection.lock().await = Some(connection); + let (remote_tx, remote_rx) = std::sync::mpsc::channel(); + *core + .on_track_cb + .lock() + .unwrap_or_else(|error| error.into_inner()) = Some(Arc::new(move |remote| { + let _ = remote_tx.send(remote); + })); + let audio = LocalAudioTrack::opus().expect("opus track"); + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(audio.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut inbound) = peer::connect_audio_receiver(&sender).await; + audio.start_pacing().await; + let inbound = tokio::time::timeout(Duration::from_secs(5), inbound.recv()) + .await + .expect("inbound track") + .expect("inbound track channel"); + for _ in 0..2 { + core.clone() + .handle_incoming_track( + generation, + epoch, + reconnect_enabled.clone(), + inbound.clone(), + Arc::downgrade(&receiver), + ) + .await; + } + let stale = remote_rx.recv().expect("stale remote track"); + let remote = remote_rx.recv().expect("latest remote track"); + let key = TrackKey::new(remote.participant().session_id.clone(), remote.track_type()); + let baseline = alive_tasks(); + + thread::spawn(move || drop(stale)) + .join() + .expect("a drop without a runtime does not panic"); + wait_for( + Duration::from_secs(2), + || alive_tasks() == baseline, + "the stale drop is handled", + ) + .await; + assert!( + !core + .manual_unsub + .lock() + .unwrap_or_else(|error| error.into_inner()) + .contains(&key), + "a stale drop must not unsubscribe the latest track" + ); + + thread::spawn(move || drop(remote)) + .join() + .expect("a drop without a runtime does not panic"); + + wait_for( + Duration::from_secs(2), + || { + core.manual_unsub + .lock() + .unwrap_or_else(|error| error.into_inner()) + .contains(&key) + }, + "unsubscribe after the drop", + ) + .await; + audio.stop(); + let _ = sender.close().await; + let _ = receiver.close().await; + core.leave("test cleanup").await.expect("leave"); +} + #[tokio::test] async fn detached_connection_ignores_publish_options_from_its_sfu() { let core = test_core(); @@ -731,7 +1213,7 @@ async fn leave_cancels_reconnect_task_before_next_generation() { fn state_events_arrive_in_the_order_of_the_state_changes() { let core = test_core(); let generation = core.begin_join().expect("join generation"); - let mut events = core.subscribe(); + let mut events = core.client_events(); let rounds = 20_000; let barrier = Arc::new(std::sync::Barrier::new(3)); let workers = [CallingState::Joined, CallingState::Reconnecting].map(|state| { @@ -750,10 +1232,8 @@ fn state_events_arrive_in_the_order_of_the_state_changes() { barrier.wait(); barrier.wait(); let mut last = None; - while let Ok(event) = events.try_recv() { - if let CallEvent::CallingStateChanged(state) = event { - last = Some(state); - } + while let Ok(ClientCallEvent::CallingStateChanged(state)) = events.try_recv() { + last = Some(state); } assert_eq!(last, Some(core.state()), "round {round}"); } @@ -766,14 +1246,16 @@ fn state_events_arrive_in_the_order_of_the_state_changes() { fn setting_the_same_state_again_sends_no_event() { let core = test_core(); let generation = core.begin_join().expect("join generation"); - let mut events = core.subscribe(); + let mut events = core.client_events(); assert!(core.set_state_if_current(generation, CallingState::Reconnecting)); assert!(core.set_state_if_current(generation, CallingState::Reconnecting)); assert!(matches!( events.try_recv(), - Ok(CallEvent::CallingStateChanged(CallingState::Reconnecting)) + Ok(ClientCallEvent::CallingStateChanged( + CallingState::Reconnecting + )) )); assert!(events.try_recv().is_err()); } @@ -781,20 +1263,20 @@ fn setting_the_same_state_again_sends_no_event() { #[test] fn join_start_sends_joining() { let core = test_core(); - let mut events = core.subscribe(); + let mut events = core.client_events(); core.begin_join().expect("join generation"); assert!(matches!( events.try_recv(), - Ok(CallEvent::CallingStateChanged(CallingState::Joining)) + Ok(ClientCallEvent::CallingStateChanged(CallingState::Joining)) )); } #[tokio::test] async fn state_during_leave_matches_the_last_state_event() { let core = test_core(); - let mut events = core.subscribe(); + let mut events = core.client_events(); core.begin_join().expect("join generation"); let connection_slot = core.connection.lock().await; let generation = core.generation(); @@ -808,10 +1290,8 @@ async fn state_during_leave_matches_the_last_state_event() { .await; let mut last = None; - while let Ok(event) = events.try_recv() { - if let CallEvent::CallingStateChanged(state) = event { - last = Some(state); - } + while let Ok(ClientCallEvent::CallingStateChanged(state)) = events.try_recv() { + last = Some(state); } assert_eq!(last, Some(core.state())); drop(connection_slot); @@ -1199,6 +1679,7 @@ async fn stale_remote_track_drop_does_not_change_new_generation_subscriptions() first, 0, TrackKey::new("remote-session".to_owned(), TrackType::Audio), + 0, ) .await; @@ -1229,6 +1710,93 @@ fn participant_refresh_replaces_published_track_state() { assert!(entry.published.contains(&(TrackType::Audio as i32))); } +#[tokio::test] +async fn turning_incoming_video_off_and_on_keeps_the_video_dimension() { + let core = test_core(); + let config = SubscriptionConfig { + default: crate::rtc::TrackSubscriptionConfig { + track_types: vec![TrackType::Video], + video_dimension: (640, 360), + ..Default::default() + }, + ..Default::default() + }; + core.update_subscriptions(config.clone()) + .await + .expect("update subscriptions"); + core.set_incoming_video_enabled(false) + .await + .expect("video off"); + core.set_incoming_video_enabled(true) + .await + .expect("video on"); + + assert_eq!( + *core.sub_config.lock().unwrap_or_else(|e| e.into_inner()), + config + ); +} + +#[test] +fn a_join_response_keeps_the_order_of_known_participants() { + let core = test_core(); + let generation = core.begin_join().expect("test generation"); + let call_state = |session_ids: &[&str]| { + Some(models::CallState { + participants: session_ids + .iter() + .map(|session_id| models::Participant { + user_id: format!("user-{session_id}"), + session_id: (*session_id).to_owned(), + ..Default::default() + }) + .collect(), + ..Default::default() + }) + }; + assert!(core.apply_join_call_state_if_current( + generation, + "me", + "me", + call_state(&["c", "a", "b"]) + )); + assert!(core.apply_join_call_state_if_current( + generation, + "me", + "me", + call_state(&["d", "b", "a"]) + )); + + let order: Vec<_> = core + .participants() + .into_iter() + .map(|participant| participant.session_id) + .collect(); + assert_eq!(order, ["me", "a", "b", "d"]); +} + +#[test] +fn participants_keep_the_order_in_which_the_call_learned_them() { + let core = test_core(); + let participant = |session_id: &str, user_id: &str| models::Participant { + user_id: user_id.to_owned(), + session_id: session_id.to_owned(), + ..Default::default() + }; + for session_id in ["h", "c", "f", "a", "g", "b", "e", "d"] { + core.upsert_participant(&participant(session_id, "user")); + } + core.remove_participant("f"); + core.upsert_participant(&participant("c", "changed")); + + let order: Vec<_> = core + .participants() + .into_iter() + .map(|participant| participant.session_id) + .collect(); + assert_eq!(order, ["h", "c", "a", "g", "b", "e", "d"]); +} + #[test] fn call_state_snapshot_combines_join_state_and_incremental_sfu_updates() { let core = test_core(); @@ -1290,6 +1858,168 @@ fn call_state_snapshot_combines_join_state_and_incremental_sfu_updates() { assert!(state.current_grants.expect("grants").can_publish_audio); } +#[test] +fn every_call_event_has_its_stable_name() { + let sfu = [ + ( + SfuCallEvent::ParticipantJoined(models::Participant::default()), + "participant_joined", + ), + ( + SfuCallEvent::ParticipantLeft(models::Participant::default()), + "participant_left", + ), + ( + SfuCallEvent::ParticipantUpdated(models::Participant::default()), + "participant_updated", + ), + ( + SfuCallEvent::TrackPublished { + user_id: String::new(), + session_id: String::new(), + track_type: TrackType::Audio, + participant: None, + }, + "track_published", + ), + ( + SfuCallEvent::TrackUnpublished { + user_id: String::new(), + session_id: String::new(), + track_type: TrackType::Audio, + cause: models::TrackUnpublishReason::UserMuted, + participant: None, + }, + "track_unpublished", + ), + ( + SfuCallEvent::DominantSpeakerChanged { + user_id: String::new(), + session_id: String::new(), + }, + "dominant_speaker_changed", + ), + ( + SfuCallEvent::AudioLevelChanged(Vec::new()), + "audio_level_changed", + ), + ( + SfuCallEvent::ConnectionQualityChanged(Vec::new()), + "connection_quality_changed", + ), + ( + SfuCallEvent::ParticipantCountChanged(models::ParticipantCount::default()), + "participant_count_changed", + ), + (SfuCallEvent::PinsUpdated(Vec::new()), "pins_updated"), + ( + SfuCallEvent::InboundStateChanged(Vec::new()), + "inbound_state_notification", + ), + ( + SfuCallEvent::PublishOptionsChanged { + publish_options: Vec::new(), + reason: String::new(), + }, + "change_publish_options", + ), + ( + SfuCallEvent::PublishQualityChanged(event::ChangePublishQuality::default()), + "change_publish_quality", + ), + ( + SfuCallEvent::CallGrantsUpdated(event::CallGrantsUpdated::default()), + "call_grants_updated", + ), + ( + SfuCallEvent::IceRestarted(PeerType::Subscriber), + "ice_restart", + ), + ( + SfuCallEvent::Error(SfuJoinError::from_event(None, 0)), + "error", + ), + ( + SfuCallEvent::CallEnded { + reason: models::CallEndedReason::Ended, + }, + "call_ended", + ), + ]; + + for (event, name) in sfu { + assert_eq!(event.name(), name, "{event:?}"); + } + assert_eq!( + ClientCallEvent::CallingStateChanged(CallingState::Joined).name(), + "calling_state_changed" + ); +} + +#[test] +fn join_state_reports_only_the_participant_changes_since_the_last_join() { + let core = test_core(); + let generation = core.begin_join().expect("join generation"); + let mut events = core.sfu_events(); + let participant = |user_id: &str, session_id: &str| models::Participant { + user_id: user_id.to_owned(), + session_id: session_id.to_owned(), + ..Default::default() + }; + let call_state = |participants| { + Some(models::CallState { + participants, + ..Default::default() + }) + }; + let bob = participant("bob", "session-b"); + + assert!(core.apply_join_call_state_if_current( + generation, + "local-1", + "agent", + call_state(vec![ + participant("agent", "local-1"), + participant("alice", "session-a"), + bob.clone(), + ]), + )); + // A REJOIN: the local session changes, alice left, bob changed, carol joined. + assert!(core.apply_join_call_state_if_current( + generation, + "local-2", + "agent", + call_state(vec![ + participant("agent", "local-2"), + models::Participant { + name: "Bob".to_owned(), + ..bob + }, + participant("carol", "session-c"), + ]), + )); + + let mut reported = Vec::new(); + while let Ok(event) = events.try_recv() { + match event { + SfuCallEvent::ParticipantJoined(p) => reported.push(("joined", p.session_id)), + SfuCallEvent::ParticipantUpdated(p) => reported.push(("updated", p.session_id)), + SfuCallEvent::ParticipantLeft(p) => reported.push(("left", p.session_id)), + _ => {} + } + } + assert_eq!( + reported, + [ + ("joined", "session-a".to_owned()), + ("joined", "session-b".to_owned()), + ("updated", "session-b".to_owned()), + ("joined", "session-c".to_owned()), + ("left", "session-a".to_owned()), + ] + ); +} + #[test] fn mute_state_builder_deduplicates_track_types() { let first = LocalTrack::Audio(LocalAudioTrack::opus().expect("first audio track")); diff --git a/src/rtc/mod.rs b/src/rtc/mod.rs index 929df56..c467c40 100644 --- a/src/rtc/mod.rs +++ b/src/rtc/mod.rs @@ -8,7 +8,8 @@ //! The participant layer sits on top: the [`coordinator`] join REST, dual //! publisher/subscriber PeerConnections ([`peer`]), and the [`join`] state //! machine ([`join::RtcCore`]) with `max_join_retries`, Stream's reconnect -//! strategies, and typed [`join::CallEvent`]s. [`crate::Call::join`] and +//! strategies, and typed event streams by source ([`SfuCallEvent`], +//! [`CoordinatorEvent`], [`ClientCallEvent`]). [`crate::Call::join`] and //! [`crate::Call::leave`] are the high-level entry points; [`RtcClient`] is the //! lower-level user-token client. //! @@ -50,7 +51,9 @@ pub use error::{ SfuTimeoutError, TwirpError, WsConnectionError, is_join_error_code, }; pub use identity::{CLIENT_TYPE, SDK_TYPE, client_details, client_header}; -pub use join::{CallEvent, CallStateSnapshot, CallingState, JoinCallData, RtcCore}; +pub use join::{ + CallStateSnapshot, CallingState, ClientCallEvent, JoinCallData, RtcCore, SfuCallEvent, +}; pub use pcm::chunk::Pad; pub use pcm::convert::G711_SAMPLE_RATE; pub use pcm::{ @@ -63,7 +66,7 @@ pub use reconnect::{ pub use sfu::signal::SignalClient; pub use sfu::ws::{SfuReceiver, SfuSender}; pub use stats::{DEFAULT_REPORTING_INTERVAL_MS, reporting_interval}; -pub use subscriptions::{SubscriptionConfig, SubscriptionTarget}; +pub use subscriptions::{SubscriptionConfig, SubscriptionTarget, TrackSubscriptionConfig}; pub use tracer::{TraceRecord, Tracer}; pub use tracks::{ Codec, LocalAudioTrack, LocalAudioTrackConfig, LocalTrack, LocalVideoTrack, diff --git a/src/rtc/pcm/frame.rs b/src/rtc/pcm/frame.rs index 774281f..40df27a 100644 --- a/src/rtc/pcm/frame.rs +++ b/src/rtc/pcm/frame.rs @@ -20,6 +20,11 @@ pub struct PcmFrame { pub sample_rate: u32, /// Channel count (1 = mono, 2 = stereo). pub channels: u16, + /// The RTP timestamp of the first sample, in units of 1/48000 s (the Opus + /// RTP clock), wrapping like RTP. Set on frames from + /// [`RemoteTrack::next_pcm`](crate::rtc::RemoteTrack::next_pcm); `None` on + /// frames that the application or a conversion builds. + pub pts: Option, } impl PcmFrame { @@ -29,6 +34,7 @@ impl PcmFrame { samples, sample_rate, channels: channels.max(1), + pts: None, } } diff --git a/src/rtc/peer/connection.rs b/src/rtc/peer/connection.rs index f4a8936..d3fda04 100644 --- a/src/rtc/peer/connection.rs +++ b/src/rtc/peer/connection.rs @@ -256,6 +256,74 @@ async fn build_generic_offer( Ok(offer.sdp) } +/// Connect a receive-only audio peer to `sender` in this process, and return it +/// with the channel of its inbound tracks. Returns when `sender` is connected. +#[cfg(test)] +pub(crate) async fn connect_audio_receiver( + sender: &RTCPeerConnection, +) -> ( + Arc, + tokio::sync::mpsc::Receiver>, +) { + use std::time::Duration; + use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState; + use webrtc::rtp_transceiver::RTCRtpTransceiverInit; + + // DTLS needs a process-default provider; `Call::join` installs it too. + let _ = rustls::crypto::ring::default_provider().install_default(); + let receiver = new_peer_connection(&[]).await.expect("receiver"); + receiver + .add_transceiver_from_kind( + RTPCodecType::Audio, + Some(RTCRtpTransceiverInit { + direction: RTCRtpTransceiverDirection::Recvonly, + send_encodings: vec![], + }), + ) + .await + .expect("receive transceiver"); + let (track_tx, track_rx) = tokio::sync::mpsc::channel(1); + receiver.on_track(Box::new(move |track, _, _| { + let track_tx = track_tx.clone(); + Box::pin(async move { + let _ = track_tx.send(track).await; + }) + })); + + let offer = sender.create_offer(None).await.expect("offer"); + let mut gathered = sender.gathering_complete_promise().await; + sender + .set_local_description(offer) + .await + .expect("set offer"); + let _ = gathered.recv().await; + let offer = sender.local_description().await.expect("offer"); + receiver + .set_remote_description(offer) + .await + .expect("apply offer"); + let answer = receiver.create_answer(None).await.expect("answer"); + let mut gathered = receiver.gathering_complete_promise().await; + receiver + .set_local_description(answer) + .await + .expect("set answer"); + let _ = gathered.recv().await; + let answer = receiver.local_description().await.expect("answer"); + sender + .set_remote_description(answer) + .await + .expect("apply answer"); + tokio::time::timeout(Duration::from_secs(10), async { + while sender.connection_state() != RTCPeerConnectionState::Connected { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("sender connects"); + (receiver, track_rx) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/rtc/peer/mod.rs b/src/rtc/peer/mod.rs index 0c8307a..70b4f30 100644 --- a/src/rtc/peer/mod.rs +++ b/src/rtc/peer/mod.rs @@ -12,6 +12,8 @@ mod ice; pub mod publisher; mod subscriber; +#[cfg(test)] +pub(crate) use connection::connect_audio_receiver; pub use connection::{generic_sdp, new_peer_connection, to_rtc_ice_servers, trace_peer_events}; pub(super) use ice::{PendingIce, register_ice_trickle}; pub(super) use subscriber::negotiate_subscriber; diff --git a/src/rtc/peer/publisher.rs b/src/rtc/peer/publisher.rs index 587257b..25c48a2 100644 --- a/src/rtc/peer/publisher.rs +++ b/src/rtc/peer/publisher.rs @@ -101,6 +101,11 @@ pub(crate) async fn restart_ice( } } } + // Media is lost until ICE connects again; the publisher `Connected` state + // resumes pacing. + for track in tracks { + track.pause_audio_pacing(); + } negotiate_publish(publisher, signal, session_id, tracks, publish_options).await } @@ -227,6 +232,39 @@ pub(crate) async fn add_transceiver_for_track( Ok(tasks) } +/// Put `track` on the sender that still carries the stopped track `retired`, +/// as JS `replaceTrack` does. Returns `false` if `publisher` has no such sender. +pub(crate) async fn replace_retired_track( + publisher: &Arc, + retired: &LocalTrack, + track: &LocalTrack, + publish_options: &[PublishOption], +) -> Result { + let retired_track_id = retired.track_id(); + for transceiver in publisher.get_transceivers().await { + let sender = transceiver.sender().await; + if !sender + .track() + .await + .is_some_and(|bound| bound.id() == retired_track_id) + { + continue; + } + let option = publish_option(track, publish_options)?; + track.configure_for_publish(option)?; + track.continue_rtp_from(retired); + let physical = track.webrtc_tracks().into_iter().next().ok_or_else(|| { + RtcError::Media("local publication has no physical encodings".to_owned()) + })?; + sender + .replace_track(Some(physical)) + .await + .map_err(RtcError::from)?; + return Ok(true); + } + Ok(false) +} + fn spawn_rtcp_reader( sender: Arc, rid: Option, @@ -322,7 +360,63 @@ mod tests { use crate::rtc::peer; use crate::rtc::proto::event::VideoLayerSetting; use crate::rtc::proto::models::{Codec, VideoDimension}; - use crate::rtc::tracks::{LocalVideoTrack, LocalVideoTrackConfig}; + use crate::rtc::tracks::{LocalAudioTrack, LocalVideoTrack, LocalVideoTrackConfig}; + use std::time::Duration; + + #[tokio::test] + async fn a_track_on_a_retired_sender_continues_its_rtp_timeline() { + let opus = [PublishOption { + track_type: TrackType::Audio as i32, + codec: Some(Codec { + name: "opus".to_owned(), + ..Default::default() + }), + ..Default::default() + }]; + let retired = LocalTrack::Audio(LocalAudioTrack::opus().expect("first track")); + let publisher = peer::new_peer_connection(&[]).await.expect("publisher"); + let rtcp_tasks = add_transceiver_for_track(&publisher, &retired, &opus) + .await + .expect("first transceiver"); + let (receiver, mut remote_rx) = peer::connect_audio_receiver(&publisher).await; + retired.start_audio_pacing().await; + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let (mut last, _) = remote.read_rtp().await.expect("first packet"); + retired.pause_audio_pacing(); + while let Ok(Ok((packet, _))) = + tokio::time::timeout(Duration::from_millis(200), remote.read_rtp()).await + { + last = packet; + } + retired.stop(); + + let track = LocalTrack::Audio(LocalAudioTrack::opus().expect("second track")); + assert!( + replace_retired_track(&publisher, &retired, &track, &opus) + .await + .expect("replace the retired track") + ); + track.start_audio_pacing().await; + let (next, _) = remote.read_rtp().await.expect("second packet"); + + assert_eq!( + next.header.sequence_number, + last.header.sequence_number.wrapping_add(1) + ); + assert_eq!( + next.header.timestamp, + last.header.timestamp.wrapping_add(960) + ); + track.stop(); + for task in rtcp_tasks { + task.abort(); + } + let _ = publisher.close().await; + let _ = receiver.close().await; + } fn video_option(name: &str) -> PublishOption { PublishOption { diff --git a/src/rtc/reconnect.rs b/src/rtc/reconnect.rs index cc11aa0..c9d73d2 100644 --- a/src/rtc/reconnect.rs +++ b/src/rtc/reconnect.rs @@ -143,7 +143,7 @@ pub fn evaluate_join_failure( return JoinAttemptOutcome::Exhausted; } JoinAttemptOutcome::Retry { - delay: retry_interval(attempt + 1), + delay: retry_interval(attempt), switch_sfu: is_join_error_code || edge_failures >= 2, } } @@ -374,6 +374,21 @@ mod tests { } } + #[test] + fn join_retry_delays_stay_within_backoff_bounds() { + for (attempt, bounds) in [(0, 250..=500), (1, 250..=2500)] { + for _ in 0..100 { + match evaluate_join_failure(false, false, 1, attempt, 3) { + JoinAttemptOutcome::Retry { delay, .. } => { + let d = delay.as_millis(); + assert!(bounds.contains(&d), "attempt {attempt} out of range: {d}"); + } + other => panic!("expected retry, got {other:?}"), + } + } + } + } + #[test] fn rate_limiter_allows_max_then_blocks_within_window() { let mut rl = SlidingWindowRateLimiter::new(10, Duration::from_secs(120)); diff --git a/src/rtc/subscriptions.rs b/src/rtc/subscriptions.rs index 44ad069..6beaba3 100644 --- a/src/rtc/subscriptions.rs +++ b/src/rtc/subscriptions.rs @@ -8,10 +8,17 @@ //! the concrete `TrackSubscriptionDetails` list and (re)sends it whenever the //! participants change. //! -//! The default policy subscribes to remote **audio** only (the backend-bot -//! default); video and screen-share are opt-in. +//! The policy has the shape of the stream-py `SubscriptionConfig`: a default +//! rule, rules by participant role, and a limit on the number of tracks. -use super::proto::models::TrackType; +use std::collections::{HashMap, HashSet}; + +use super::proto::models::{self, TrackType}; +use super::proto::signal; + +/// Video dimension requested when a subscription gives none. The SFU rejects a +/// video or screen-share subscription without a dimension. +pub(crate) const DEFAULT_VIDEO_DIMENSION: (u32, u32) = (1920, 1080); /// A precise subscription to one participant session and track kind. #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -21,12 +28,13 @@ pub struct SubscriptionTarget { pub session_id: String, /// The remote track kind to receive. pub track_type: TrackType, - /// Optional preferred video dimensions sent as an SFU adaptation hint. + /// Preferred video dimensions sent as an SFU adaptation hint. `None` + /// requests 1920×1080 for video and screen-share. pub dimension: Option<(u32, u32)>, } impl SubscriptionTarget { - /// Subscribe to `track_type` from `session_id` using the SFU's default size. + /// Subscribe to `track_type` from `session_id`, at 1920×1080 for video. pub fn new(session_id: impl Into, track_type: TrackType) -> Self { Self { session_id: session_id.into(), @@ -43,78 +51,136 @@ impl SubscriptionTarget { } } -/// Which remote track kinds to subscribe to. -/// -/// Reactive: the call subscribes to every matching track published by every -/// other participant, and updates as participants publish/unpublish. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct SubscriptionConfig { - /// Subscribe to remote audio. - pub audio: bool, - /// Subscribe to remote video. - pub video: bool, - /// Subscribe to remote screen-share (video + audio). - pub screen_share: bool, - /// Preferred video dimension hint sent to the SFU (width, height). - pub video_dimension: Option<(u32, u32)>, +/// The subscription rule for a group of participants. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TrackSubscriptionConfig { + /// The remote track kinds to receive. + pub track_types: Vec, + /// Preferred camera video dimension (width, height), sent to the SFU as an + /// adaptation hint. + pub video_dimension: (u32, u32), + /// Preferred screen-share dimension (width, height), sent to the SFU as an + /// adaptation hint. + pub screenshare_dimension: (u32, u32), } -impl Default for SubscriptionConfig { - /// Audio-only — the backend-bot default (matches stream-py's usual path). +impl Default for TrackSubscriptionConfig { + /// No track kinds, 1920×1080 for video and screen-share. fn default() -> Self { Self { - audio: true, - video: false, - screen_share: false, - video_dimension: None, + track_types: Vec::new(), + video_dimension: DEFAULT_VIDEO_DIMENSION, + screenshare_dimension: DEFAULT_VIDEO_DIMENSION, } } } +/// Which remote tracks to subscribe to. +/// +/// Reactive: the call subscribes to the matching tracks of every other +/// participant, and updates as participants join, leave, change, and publish. +/// The default subscribes to nothing. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct SubscriptionConfig { + /// The rule for a participant whose roles have no rule in `role_filters`. + pub default: TrackSubscriptionConfig, + /// Rules by participant role. The first role of the participant that has a + /// rule selects it. + pub role_filters: HashMap, + /// The maximum number of subscribed tracks. + pub max_subscriptions: Option, +} + impl SubscriptionConfig { - /// Subscribe to audio from all participants (the default). + /// Subscribe to audio from all participants. pub fn audio_all() -> Self { - Self::default() + Self { + default: TrackSubscriptionConfig { + track_types: vec![TrackType::Audio], + ..Default::default() + }, + ..Default::default() + } } /// Subscribe to audio and video from all participants. pub fn audio_video() -> Self { Self { - audio: true, - video: true, - video_dimension: Some((1280, 720)), - ..Self::default() + default: TrackSubscriptionConfig { + track_types: vec![TrackType::Audio, TrackType::Video], + ..Default::default() + }, + ..Default::default() } } - /// Subscribe to audio, video, and screen-share. + /// Subscribe to audio, video, screen-share, and screen-share audio. pub fn all() -> Self { Self { - audio: true, - video: true, - screen_share: true, - video_dimension: Some((1280, 720)), + default: TrackSubscriptionConfig { + track_types: vec![ + TrackType::Audio, + TrackType::Video, + TrackType::ScreenShare, + TrackType::ScreenShareAudio, + ], + ..Default::default() + }, + ..Default::default() } } /// Subscribe to nothing (unsubscribe from all). pub fn none() -> Self { - Self { - audio: false, - video: false, - screen_share: false, - video_dimension: None, - } + Self::default() } - /// Whether this policy subscribes to `track_type`. - pub fn matches(&self, track_type: TrackType) -> bool { - match track_type { - TrackType::Audio => self.audio, - TrackType::Video => self.video, - TrackType::ScreenShare | TrackType::ScreenShareAudio => self.screen_share, - TrackType::Unspecified => false, + /// The subscriptions to the tracks of `participants`, in their order, except + /// the tracks in `unsubscribed`. + pub(crate) fn track_subscriptions<'a>( + &self, + participants: impl IntoIterator, + unsubscribed: &HashSet, + ) -> Vec { + let mut tracks = Vec::new(); + for participant in participants { + let rule = self.rule_for(participant); + for &published in &participant.published_tracks { + let Ok(track_type) = TrackType::try_from(published) else { + continue; + }; + if !rule.track_types.contains(&track_type) + || unsubscribed + .contains(&TrackKey::new(participant.session_id.clone(), track_type)) + { + continue; + } + let dimension = match track_type { + TrackType::Video => Some(rule.video_dimension), + TrackType::ScreenShare => Some(rule.screenshare_dimension), + _ => None, + }; + tracks.push(signal::TrackSubscriptionDetails { + user_id: participant.user_id.clone(), + session_id: participant.session_id.clone(), + track_type: published, + dimension: dimension + .map(|(width, height)| models::VideoDimension { width, height }), + }); + } + } + if let Some(max) = self.max_subscriptions { + tracks.truncate(max); } + tracks + } + + fn rule_for(&self, participant: &models::Participant) -> &TrackSubscriptionConfig { + participant + .roles + .iter() + .find_map(|role| self.role_filters.get(role)) + .unwrap_or(&self.default) } } @@ -138,27 +204,150 @@ impl TrackKey { mod tests { use super::*; + fn participant( + session_id: &str, + roles: &[&str], + published: &[TrackType], + ) -> models::Participant { + models::Participant { + user_id: format!("user-{session_id}"), + session_id: session_id.to_owned(), + roles: roles.iter().map(|role| (*role).to_owned()).collect(), + published_tracks: published + .iter() + .map(|track_type| *track_type as i32) + .collect(), + ..Default::default() + } + } + + fn subscribed( + config: &SubscriptionConfig, + participants: &[models::Participant], + ) -> Vec<(String, TrackType)> { + config + .track_subscriptions(participants, &HashSet::new()) + .into_iter() + .map(|track| (track.session_id.clone(), track.track_type())) + .collect() + } + + fn rule(track_types: &[TrackType]) -> TrackSubscriptionConfig { + TrackSubscriptionConfig { + track_types: track_types.to_vec(), + ..Default::default() + } + } + #[test] - fn default_is_audio_only() { - let c = SubscriptionConfig::default(); - assert!(c.matches(TrackType::Audio)); - assert!(!c.matches(TrackType::Video)); - assert!(!c.matches(TrackType::ScreenShare)); + fn a_role_rule_replaces_the_default_rule() { + let config = SubscriptionConfig { + default: rule(&[TrackType::Audio]), + role_filters: HashMap::from([("host".to_owned(), rule(&[TrackType::Video]))]), + ..Default::default() + }; + let both = [TrackType::Audio, TrackType::Video]; + let participants = [ + participant("host", &["host"], &both), + participant("guest", &["user"], &both), + ]; + + assert_eq!( + subscribed(&config, &participants), + [ + ("host".to_owned(), TrackType::Video), + ("guest".to_owned(), TrackType::Audio), + ] + ); + } + + #[test] + fn the_first_role_of_the_participant_with_a_rule_wins() { + let config = SubscriptionConfig { + role_filters: HashMap::from([ + ("admin".to_owned(), rule(&[TrackType::Audio])), + ("host".to_owned(), rule(&[TrackType::Video])), + ]), + ..Default::default() + }; + let both = [TrackType::Audio, TrackType::Video]; + let participants = [ + participant("a", &["user", "host", "admin"], &both), + participant("b", &["admin", "host"], &both), + ]; + + assert_eq!( + subscribed(&config, &participants), + [ + ("a".to_owned(), TrackType::Video), + ("b".to_owned(), TrackType::Audio), + ] + ); } #[test] - fn audio_video_opts_in_video() { - let c = SubscriptionConfig::audio_video(); - assert!(c.matches(TrackType::Audio)); - assert!(c.matches(TrackType::Video)); - assert!(!c.matches(TrackType::ScreenShare)); + fn video_and_screen_share_get_their_own_dimensions() { + let config = SubscriptionConfig { + default: TrackSubscriptionConfig { + track_types: vec![TrackType::Audio, TrackType::Video, TrackType::ScreenShare], + video_dimension: (640, 360), + screenshare_dimension: (2560, 1440), + }, + ..Default::default() + }; + let presenter = participant( + "presenter", + &[], + &[ + TrackType::Audio, + TrackType::Video, + TrackType::ScreenShare, + TrackType::ScreenShareAudio, + ], + ); + + let dimensions: Vec<_> = config + .track_subscriptions(&[presenter], &HashSet::new()) + .into_iter() + .map(|track| { + let dimension = track.dimension.map(|d| (d.width, d.height)); + (track.track_type(), dimension) + }) + .collect(); + + assert_eq!( + dimensions, + [ + (TrackType::Audio, None), + (TrackType::Video, Some((640, 360))), + (TrackType::ScreenShare, Some((2560, 1440))), + ] + ); } #[test] - fn none_matches_nothing() { - let c = SubscriptionConfig::none(); - assert!(!c.matches(TrackType::Audio)); - assert!(!c.matches(TrackType::Video)); + fn the_limit_keeps_the_first_tracks_in_participant_order() { + let config = SubscriptionConfig { + max_subscriptions: Some(2), + ..SubscriptionConfig::audio_all() + }; + let participants = ["c", "a", "b"].map(|id| participant(id, &[], &[TrackType::Audio])); + let first = |tracks: Vec| { + tracks + .into_iter() + .map(|track| track.session_id) + .collect::>() + }; + + assert_eq!( + first(config.track_subscriptions(&participants, &HashSet::new())), + ["c", "a"] + ); + let unsubscribed = HashSet::from([TrackKey::new("c", TrackType::Audio)]); + assert_eq!( + first(config.track_subscriptions(&participants, &unsubscribed)), + ["a", "b"] + ); } #[test] diff --git a/src/rtc/tracks/local.rs b/src/rtc/tracks/local.rs index 5ca186a..30f1c80 100644 --- a/src/rtc/tracks/local.rs +++ b/src/rtc/tracks/local.rs @@ -5,9 +5,11 @@ //! `StartWrite`). Three write paths feed the same outbound track: //! //! - [`LocalAudioTrack::write_pcm`] — raw [`PcmFrame`]s. Resampled to 48 kHz -//! mono and paced into 20 ms Opus frames by a background task that emits -//! silence on starve (stream-py `AudioStreamTrack` pacing). This is the PCM -//! republish / TTS-bot path. +//! mono, queued, and paced into 20 ms Opus frames by a background task that +//! emits silence on starve (stream-py `AudioStreamTrack` pacing). Pacing runs +//! after [`LocalAudioTrack::start_pacing`]; a published track paces only +//! while the SFU publisher is connected. This is the PCM republish / TTS-bot +//! path. //! - [`LocalAudioTrack::write_sample`] / [`LocalVideoTrack::write_sample`] — //! already-encoded media (Opus/VP8/…) plus a frame duration; the SDK //! packetizes and writes. The caller controls pacing. @@ -27,7 +29,7 @@ use std::collections::VecDeque; use std::num::NonZeroU8; use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU8, AtomicU16, AtomicU32, Ordering}; -use std::sync::{Arc, Mutex as StdMutex}; +use std::sync::{Arc, Mutex as StdMutex, Weak}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use bytes::Bytes; @@ -70,7 +72,7 @@ const VIDEO_BITRATE_KBPS: u32 = 1_000; const MAX_LOCAL_VIDEO_EDGE: u32 = 3_840; const MAX_LOCAL_VIDEO_PIXELS: u64 = 3_840 * 2_160; const MAX_LOCAL_VIDEO_I420_BYTES: usize = 3_840 * 2_160 * 3 / 2; -const PCM_QUEUE_CAPACITY_SAMPLES: usize = FRAME_SAMPLES_20MS * 10; +const PCM_QUEUE_CAPACITY: Duration = Duration::from_secs(60); const MAX_OPUS_PACKET_BYTES: usize = 1_500; const AUDIO_BITRATE_BPS: u32 = 32_000; @@ -324,10 +326,13 @@ struct AudioInner { core: TrackCore, /// Resampled 48 kHz mono PCM awaiting the 20 ms pacer. pcm: StdMutex>, + pcm_capacity_samples: usize, resampler: StdMutex, encoder: StdMutex, pacer: StdMutex>>, pacer_started: AtomicBool, + /// Set by `start_pacing`, cleared by `pause_pacing`. + pacing_enabled: AtomicBool, pcm_pacing: AtomicBool, write_guard: tokio::sync::Mutex<()>, } @@ -346,6 +351,9 @@ pub struct LocalAudioTrackConfig { pub expected_packet_loss_pct: u8, /// Discontinuous transmission: stop emitting packets during silence. pub dtx: bool, + /// Maximum PCM that [`LocalAudioTrack::write_pcm`] queues for the pacer. A + /// write above it drops the oldest queued samples. + pub pcm_queue_capacity: Duration, } impl Default for LocalAudioTrackConfig { @@ -355,6 +363,7 @@ impl Default for LocalAudioTrackConfig { inband_fec: true, expected_packet_loss_pct: EXPECTED_PACKET_LOSS_PCT, dtx: true, + pcm_queue_capacity: PCM_QUEUE_CAPACITY, } } } @@ -389,6 +398,13 @@ impl LocalAudioTrackConfig { self.dtx = dtx; self } + + /// Set the maximum PCM that [`LocalAudioTrack::write_pcm`] queues. + #[must_use] + pub fn with_pcm_queue_capacity(mut self, pcm_queue_capacity: Duration) -> Self { + self.pcm_queue_capacity = pcm_queue_capacity; + self + } } /// An outbound Opus audio track. @@ -433,14 +449,18 @@ impl LocalAudioTrack { encoder.set_inband_fec(config.inband_fec)?; encoder.set_packet_loss_perc(i32::from(config.expected_packet_loss_pct))?; encoder.set_dtx(config.dtx)?; + let pcm_capacity_samples = + (config.pcm_queue_capacity.as_secs_f64() * f64::from(OPUS_SAMPLE_RATE)) as usize; Ok(Self { inner: Arc::new(AudioInner { core, - pcm: StdMutex::new(VecDeque::with_capacity(PCM_QUEUE_CAPACITY_SAMPLES)), + pcm: StdMutex::new(VecDeque::new()), + pcm_capacity_samples, resampler: StdMutex::new(StreamResampler::to_opus_mono()), encoder: StdMutex::new(encoder), pacer: StdMutex::new(None), pacer_started: AtomicBool::new(false), + pacing_enabled: AtomicBool::new(false), pcm_pacing: AtomicBool::new(true), write_guard: tokio::sync::Mutex::new(()), }), @@ -449,19 +469,20 @@ impl LocalAudioTrack { /// Queue a PCM frame for the paced 20 ms Opus encoder. /// - /// The frame is resampled to 48 kHz mono and buffered for at most 200 ms; a - /// background task emits one Opus packet every 20 ms, writing silence when - /// the buffer runs dry. To keep interactive audio current, this method does - /// not backpressure a producer: overflow drops the oldest queued samples and - /// retains the newest audio. [`flush`](Self::flush) still drops all unsent - /// samples immediately for barge-in. + /// The frame is resampled to 48 kHz mono and buffered up to + /// [`LocalAudioTrackConfig::pcm_queue_capacity`]. While pacing runs (see + /// [`start_pacing`](Self::start_pacing)), a background task emits one Opus + /// packet every 20 ms, writing silence when the buffer runs dry. This method + /// does not backpressure a producer: overflow drops the oldest queued + /// samples and retains the newest audio. [`flush`](Self::flush) still drops + /// all unsent samples immediately for barge-in. /// /// # Errors /// /// Returns [`RtcError::PcmQueueOverflow`] after retaining the newest audio - /// when this write exceeds the 200 ms queue. The caller may continue writing; - /// the typed error makes overload observable without allowing stale audio to - /// accumulate. + /// when this write exceeds the queue capacity. The caller may continue + /// writing; the typed error makes overload observable without allowing + /// stale audio to accumulate. pub async fn write_pcm(&self, frame: PcmFrame) -> Result<()> { if self.inner.core.stopped.load(Ordering::SeqCst) { return Err(RtcError::IllegalState( @@ -479,21 +500,20 @@ impl LocalAudioTrack { }; let dropped = { let mut buf = self.inner.pcm.lock().unwrap_or_else(|e| e.into_inner()); - let dropped = push_bounded_pcm(&mut buf, resampled); + let dropped = push_bounded_pcm(&mut buf, resampled, self.inner.pcm_capacity_samples); if dropped > 0 { tracing::debug!( dropped_samples = dropped, - capacity_samples = PCM_QUEUE_CAPACITY_SAMPLES, + capacity_samples = self.inner.pcm_capacity_samples, "stream.rtc.audio.pcm_queue_overflow" ); } dropped }; - self.ensure_pacer(); if dropped > 0 { Err(RtcError::PcmQueueOverflow { dropped_samples: dropped, - capacity_samples: PCM_QUEUE_CAPACITY_SAMPLES, + capacity_samples: self.inner.pcm_capacity_samples, }) } else { Ok(()) @@ -569,6 +589,49 @@ impl LocalAudioTrack { buf.clear(); } + /// Start taking queued PCM: one 20 ms frame every 20 ms, or silence when + /// the queue is empty. Before this, [`write_pcm`](Self::write_pcm) only + /// fills the queue. + /// + /// [`Call::publish_audio`](crate::Call::publish_audio) starts and pauses + /// pacing with the SFU publisher connection. Call this yourself only for a + /// track on your own PeerConnection (see [`webrtc_track`](Self::webrtc_track)), + /// after that PeerConnection connects. + pub async fn start_pacing(&self) { + self.inner.pacing_enabled.store(true, Ordering::SeqCst); + self.ensure_pacer(); + } + + /// Stop taking queued PCM. The queue keeps its audio for the next + /// [`start_pacing`](Self::start_pacing). + pub fn pause_pacing(&self) { + self.inner.pacing_enabled.store(false, Ordering::SeqCst); + } + + #[cfg(test)] + pub(crate) fn is_pacing(&self) -> bool { + self.inner.pacing_enabled.load(Ordering::SeqCst) + } + + /// Continue the RTP sequence numbers and timestamps of `previous`, whose + /// sender this track takes over. The SFU drops a stream whose timestamps + /// go back. + fn continue_rtp_from(&self, previous: &LocalAudioTrack) { + let packetizer = previous + .inner + .core + .packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + *self + .inner + .core + .packetizer + .lock() + .unwrap_or_else(|e| e.into_inner()) = packetizer; + } + /// Stop the pacer and reject further writes. Called by `stop_publish`/`leave`. pub(crate) fn stop(&self) { self.inner.core.stop(); @@ -589,7 +652,8 @@ impl LocalAudioTrack { /// [`Call::publish_audio`](crate::Call::publish_audio) does this for the /// SFU; you only need it to send the same audio to a second peer, such as an /// AI provider's Realtime endpoint. Every write path (`write_pcm` and - /// friends) feeds all bound senders. + /// friends) feeds all bound senders. A track that is not published paces + /// `write_pcm` audio only after [`start_pacing`](Self::start_pacing). pub fn webrtc_track(&self) -> Arc { self.inner.core.track.clone() } @@ -603,7 +667,7 @@ impl LocalAudioTrack { } /// Spawn the 20 ms PCM/silence pacing task (idempotent). - pub(crate) fn ensure_pacer(&self) { + fn ensure_pacer(&self) { if self .inner .pacer_started @@ -612,25 +676,30 @@ impl LocalAudioTrack { { return; } - let inner = self.inner.clone(); - let handle = tokio::spawn(async move { pace_audio(inner).await }); + let track = Arc::downgrade(&self.inner); + let handle = tokio::spawn(async move { pace_audio(track).await }); *self.inner.pacer.lock().unwrap_or_else(|e| e.into_inner()) = Some(handle); } } /// The 20 ms pacing loop: pull one Opus frame worth of PCM (or silence) every -/// tick, encode it, and packetize it onto the outbound track. -async fn pace_audio(inner: Arc) { +/// tick, encode it, and packetize it onto the outbound track. It ends when the +/// track is stopped or dropped. +async fn pace_audio(track: Weak) { let mut interval = tokio::time::interval(Duration::from_millis(20)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let mut scratch = vec![0i16; FRAME_SAMPLES_20MS]; let mut encoded = vec![0u8; MAX_OPUS_PACKET_BYTES]; loop { interval.tick().await; + let Some(inner) = track.upgrade() else { + return; + }; if inner.core.stopped.load(Ordering::SeqCst) { return; } - if !inner.pcm_pacing.load(Ordering::SeqCst) { + if !inner.pcm_pacing.load(Ordering::SeqCst) || !inner.pacing_enabled.load(Ordering::SeqCst) + { continue; } let _write = inner.write_guard.lock().await; @@ -671,11 +740,11 @@ async fn pace_audio(inner: Arc) { } } -fn push_bounded_pcm(queue: &mut VecDeque, samples: Vec) -> usize { +fn push_bounded_pcm(queue: &mut VecDeque, samples: Vec, capacity: usize) -> usize { let overflow = queue .len() .saturating_add(samples.len()) - .saturating_sub(PCM_QUEUE_CAPACITY_SAMPLES); + .saturating_sub(capacity); let from_queue = overflow.min(queue.len()); queue.drain(..from_queue); let from_samples = overflow - from_queue; @@ -1926,10 +1995,19 @@ impl LocalTrack { } } - pub(crate) fn start_media(&self) { + pub(crate) async fn start_audio_pacing(&self) { + match self { + LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => { + track.start_pacing().await; + } + LocalTrack::Video { .. } => {} + } + } + + pub(crate) fn pause_audio_pacing(&self) { match self { LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => { - track.ensure_pacer(); + track.pause_pacing(); } LocalTrack::Video { .. } => {} } @@ -1942,6 +2020,18 @@ impl LocalTrack { } } + /// Continue the RTP timeline of the audio track `previous`, whose sender + /// this audio track takes over. + pub(crate) fn continue_rtp_from(&self, previous: &LocalTrack) { + if let ( + LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track), + LocalTrack::Audio(previous) | LocalTrack::ScreenShareAudio(previous), + ) = (self, previous) + { + track.continue_rtp_from(previous); + } + } + pub(crate) fn set_muted(&self, muted: bool) { match self { LocalTrack::Audio(track) | LocalTrack::ScreenShareAudio(track) => { @@ -1992,6 +2082,7 @@ impl LocalTrack { #[cfg(test)] mod tests { use super::*; + use crate::rtc::peer; use crate::rtc::proto::models::{Codec, VideoDimension}; /// One 20 ms frame of 440 Hz tone: FEC and DTX both key off whether the @@ -2122,19 +2213,9 @@ mod tests { } #[tokio::test] - async fn write_pcm_paces_without_binding() { - let track = LocalAudioTrack::opus().expect("opus track"); - let frame = PcmFrame::mono(vec![1000; FRAME_SAMPLES_20MS], OPUS_SAMPLE_RATE); - track.write_pcm(frame).await.expect("write_pcm"); - // Give the pacer a couple of ticks; it must not panic writing silence. - tokio::time::sleep(Duration::from_millis(50)).await; - track.stop(); - } - - #[tokio::test] - async fn publication_starts_paced_silence_with_truthful_level() { + async fn started_pacing_sends_silence_with_truthful_level() { let track = LocalAudioTrack::opus().expect("opus track"); - LocalTrack::Audio(track.clone()).start_media(); + track.start_pacing().await; tokio::time::sleep(Duration::from_millis(30)).await; assert!(track.inner.pacer_started.load(Ordering::SeqCst)); assert_eq!(track.inner.core.audio_level.load(Ordering::Relaxed), 127); @@ -2150,9 +2231,9 @@ mod tests { .pcm .lock() .unwrap_or_else(|error| error.into_inner()); - queue.extend(std::iter::repeat_n(1, PCM_QUEUE_CAPACITY_SAMPLES - 2)); - assert_eq!(push_bounded_pcm(&mut queue, vec![2, 3, 4, 5]), 2); - assert_eq!(queue.len(), PCM_QUEUE_CAPACITY_SAMPLES); + queue.extend(std::iter::repeat_n(1, 4)); + assert_eq!(push_bounded_pcm(&mut queue, vec![2, 3, 4, 5], 6), 2); + assert_eq!(queue.len(), 6); assert_eq!(queue.back(), Some(&5)); } track.flush(); @@ -2167,17 +2248,26 @@ mod tests { } #[tokio::test] - async fn write_pcm_reports_typed_overflow_after_retaining_newest_audio() { + async fn write_pcm_above_the_default_minute_keeps_the_newest_samples() { let track = LocalAudioTrack::opus().expect("opus track"); - let samples = vec![7; PCM_QUEUE_CAPACITY_SAMPLES + FRAME_SAMPLES_20MS]; + let minute = 60 * OPUS_SAMPLE_RATE as usize; + track + .write_pcm(PcmFrame::mono(vec![1; minute], OPUS_SAMPLE_RATE)) + .await + .expect("a minute fits the default queue"); + let result = track - .write_pcm(PcmFrame::mono(samples, OPUS_SAMPLE_RATE)) + .write_pcm(PcmFrame::mono( + vec![2; FRAME_SAMPLES_20MS], + OPUS_SAMPLE_RATE, + )) .await; + assert!(matches!( result, Err(RtcError::PcmQueueOverflow { dropped_samples: FRAME_SAMPLES_20MS, - capacity_samples: PCM_QUEUE_CAPACITY_SAMPLES, + capacity_samples: 2_880_000, }) )); let queue = track @@ -2185,12 +2275,144 @@ mod tests { .pcm .lock() .unwrap_or_else(|error| error.into_inner()); - assert_eq!(queue.len(), PCM_QUEUE_CAPACITY_SAMPLES); - assert!(queue.iter().all(|sample| *sample == 7)); + assert_eq!(queue.len(), minute); + assert_eq!(queue.front(), Some(&1)); + assert!( + queue + .iter() + .rev() + .take(FRAME_SAMPLES_20MS) + .all(|sample| *sample == 2) + ); drop(queue); track.stop(); } + #[tokio::test] + async fn pcm_queue_capacity_is_set_per_track() { + let track = LocalAudioTrack::opus_with_config( + LocalAudioTrackConfig::default().with_pcm_queue_capacity(Duration::from_millis(100)), + ) + .expect("opus track"); + + let result = track + .write_pcm(PcmFrame::mono( + vec![7; FRAME_SAMPLES_20MS * 6], + OPUS_SAMPLE_RATE, + )) + .await; + + assert!(matches!( + result, + Err(RtcError::PcmQueueOverflow { + dropped_samples: FRAME_SAMPLES_20MS, + capacity_samples: 4_800, + }) + )); + track.stop(); + } + + #[tokio::test] + async fn pcm_written_before_pacing_starts_stays_queued() { + let track = LocalAudioTrack::opus_with_config( + LocalAudioTrackConfig::default().with_pcm_queue_capacity(Duration::from_millis(100)), + ) + .expect("opus track"); + track + .write_pcm(PcmFrame::mono(vec![7; 4_800], OPUS_SAMPLE_RATE)) + .await + .expect("100 ms fits the queue"); + + // A running pacer takes 960 samples every 20 ms. + tokio::time::sleep(Duration::from_millis(100)).await; + let result = track + .write_pcm(PcmFrame::mono( + vec![7; FRAME_SAMPLES_20MS], + OPUS_SAMPLE_RATE, + )) + .await; + + assert!(matches!( + result, + Err(RtcError::PcmQueueOverflow { + dropped_samples: FRAME_SAMPLES_20MS, + .. + }) + )); + track.stop(); + } + + #[tokio::test] + async fn dropping_a_started_track_ends_its_pacer() { + let alive_tasks = || { + tokio::runtime::Handle::current() + .metrics() + .num_alive_tasks() + }; + let before = alive_tasks(); + let track = LocalAudioTrack::opus().expect("opus track"); + track.start_pacing().await; + assert_eq!(alive_tasks(), before + 1); + + drop(track); + + tokio::time::timeout(Duration::from_secs(1), async { + while alive_tasks() != before { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("the pacer ends with its track"); + } + + #[tokio::test] + async fn pcm_queued_before_the_connection_is_sent_from_its_first_sample() { + let track = LocalAudioTrack::opus().expect("opus track"); + let minute = tone_20ms().repeat(3_000); + let queued = minute.len(); + track + .write_pcm(PcmFrame::mono(minute, OPUS_SAMPLE_RATE)) + .await + .expect("a minute fits the default queue"); + + let sender = peer::new_peer_connection(&[]).await.expect("sender"); + sender + .add_track(track.webrtc_track()) + .await + .expect("add track"); + let (receiver, mut remote_rx) = peer::connect_audio_receiver(&sender).await; + + track.start_pacing().await; + let remote = tokio::time::timeout(Duration::from_secs(5), remote_rx.recv()) + .await + .expect("remote track") + .expect("remote track channel"); + let mut received = 0; + let paced_until = tokio::time::Instant::now() + Duration::from_secs(1); + while tokio::time::Instant::now() < paced_until { + remote.read_rtp().await.expect("paced packet"); + received += 1; + } + track.pause_pacing(); + while let Ok(Ok(_)) = + tokio::time::timeout(Duration::from_millis(200), remote.read_rtp()).await + { + received += 1; + } + + let taken = queued + - track + .inner + .pcm + .lock() + .unwrap_or_else(|error| error.into_inner()) + .len(); + assert_eq!(taken, received * FRAME_SAMPLES_20MS); + track.stop(); + let _ = sender.close().await; + let _ = receiver.close().await; + } + #[test] fn audio_level_dbov_maps_rms_to_rfc6464_levels() { assert_eq!(audio_level_dbov(0.0), 127, "digital silence"); @@ -2222,6 +2444,7 @@ mod tests { ); let loud = PcmFrame::mono(vec![i16::MAX / 2; FRAME_SAMPLES_20MS * 4], OPUS_SAMPLE_RATE); track.write_pcm(loud).await.expect("write_pcm"); + track.start_pacing().await; tokio::time::sleep(Duration::from_millis(60)).await; let level = track.inner.core.audio_level.load(Ordering::Relaxed); track.stop(); diff --git a/src/rtc/tracks/remote.rs b/src/rtc/tracks/remote.rs index e17a28b..8e36ac0 100644 --- a/src/rtc/tracks/remote.rs +++ b/src/rtc/tracks/remote.rs @@ -214,11 +214,16 @@ struct VideoDecode { struct AudioDecode { decoder: opus::Decoder, last_seq: Option, - ready: VecDeque>, + ready: VecDeque, + /// The RTP timestamp right after the last queued frame. A frame rebuilt + /// for a lost packet starts here. + next_pts: u32, /// Length of the last frame decoded from a real packet. libopus makes a /// rebuilt frame as long as the output buffer. A lost packet states no /// length, so the stream's own frame size is the best value to use. frame_samples: usize, + /// Decode output, reused for every packet. Queued frames are exact copies. + scratch: Vec, } /// How this track's payload is turned into something the caller can use. @@ -360,7 +365,8 @@ impl RemoteTrack { } } - /// Decode and return the next audio frame as 48 kHz mono s16 PCM. + /// Decode and return the next audio frame as 48 kHz mono s16 PCM, with + /// [`PcmFrame::pts`] set to the RTP timestamp of its first sample. /// /// Skips empty/comfort-noise packets and returns `None` only when the track /// ends. Returns `None` immediately for non-audio tracks. Concurrent reads @@ -371,17 +377,21 @@ impl RemoteTrack { }; let _read_guard = self.read_gate.lock().await; loop { - if let Some(samples) = state.lock().unwrap_or_else(|e| e.into_inner()).take_frame() { - return Some(PcmFrame::mono(samples, OPUS_SAMPLE_RATE)); + if let Some(frame) = state.lock().unwrap_or_else(|e| e.into_inner()).take_frame() { + return Some(frame); } let pkt = self.read_rtp_inner().await?; if pkt.payload.is_empty() { continue; } let mut state = state.lock().unwrap_or_else(|e| e.into_inner()); - state.push_packet(pkt.header.sequence_number, &pkt.payload); - if let Some(samples) = state.take_frame() { - return Some(PcmFrame::mono(samples, OPUS_SAMPLE_RATE)); + state.push_packet( + pkt.header.sequence_number, + pkt.header.timestamp, + &pkt.payload, + ); + if let Some(frame) = state.take_frame() { + return Some(frame); } } } @@ -514,7 +524,9 @@ impl AudioDecode { decoder, last_seq: None, ready: VecDeque::new(), + next_pts: 0, frame_samples: FRAME_SAMPLES_20MS, + scratch: vec![0; MAX_OPUS_FRAME_SAMPLES], } } @@ -528,7 +540,7 @@ impl AudioDecode { /// Only the frame directly before `sequence_number` can be rebuilt from real /// audio: in-band FEC puts a low-quality copy of a frame into the *next* /// packet, so anything lost earlier had its copy in a lost packet too. - fn push_packet(&mut self, sequence_number: u16, payload: &[u8]) { + fn push_packet(&mut self, sequence_number: u16, rtp_timestamp: u32, payload: &[u8]) { let missing = match self.last_seq { None => 0, Some(last) => { @@ -550,18 +562,19 @@ impl AudioDecode { self.last_seq = Some(sequence_number); for _ in 1..missing { - self.decode_frame(&[], false); + self.decode_frame(&[], false, self.next_pts); } if missing > 0 { - self.decode_frame(payload, true); + self.decode_frame(payload, true, self.next_pts); } - self.decode_frame(payload, false); + self.decode_frame(payload, false, rtp_timestamp); } - /// Decode one frame and queue it. An empty `payload` makes libopus build a - /// replacement for a lost frame. `fec` takes the copy of the previous frame - /// out of `payload` instead of decoding `payload` itself. - fn decode_frame(&mut self, payload: &[u8], fec: bool) { + /// Decode one frame that starts at RTP timestamp `pts` and queue it. An + /// empty `payload` makes libopus build a replacement for a lost frame. + /// `fec` takes the copy of the previous frame out of `payload` instead of + /// decoding `payload` itself. + fn decode_frame(&mut self, payload: &[u8], fec: bool, pts: u32) { let rebuilt = fec || payload.is_empty(); // A real packet states its own length and the buffer is only an upper // bound. For a rebuilt frame the buffer length is the length libopus @@ -571,17 +584,21 @@ impl AudioDecode { } else { MAX_OPUS_FRAME_SAMPLES }; - let mut out = vec![0i16; capacity]; - match self.decoder.decode(payload, &mut out, fec) { + match self + .decoder + .decode(payload, &mut self.scratch[..capacity], fec) + { Ok(samples) => { - out.truncate(samples); - if out.is_empty() { + if samples == 0 { return; } if !rebuilt { self.frame_samples = samples; } - self.ready.push_back(out); + let mut frame = PcmFrame::mono(self.scratch[..samples].to_vec(), OPUS_SAMPLE_RATE); + frame.pts = Some(pts); + self.ready.push_back(frame); + self.next_pts = pts.wrapping_add(samples as u32); } Err(error) => { tracing::debug!(error = %error, "stream.rtc.remote.opus_decode_failed"); @@ -589,7 +606,7 @@ impl AudioDecode { } } - fn take_frame(&mut self) -> Option> { + fn take_frame(&mut self) -> Option { self.ready.pop_front() } } @@ -807,18 +824,25 @@ mod tests { .unwrap_or(0) } + /// The RTP timestamp of packet `sequence_number` in a 20 ms stream. + fn rtp(sequence_number: u16) -> u32 { + u32::from(sequence_number) * FRAME_SAMPLES_20MS as u32 + } + #[test] fn an_unbroken_sequence_yields_one_frame_per_packet() { let packets = tone_packets(4, true); let mut state = audio_decode(); for (index, packet) in packets.iter().enumerate() { - state.push_packet(index as u16, packet); + state.push_packet(index as u16, rtp(index as u16), packet); + let frame = state.take_frame(); assert_eq!( - state.take_frame().map(|frame| frame.len()), + frame.as_ref().map(|frame| frame.samples.len()), Some(FRAME_SAMPLES_20MS), "packet {index} should yield exactly one frame" ); + assert_eq!(frame.and_then(|frame| frame.pts), Some(rtp(index as u16))); assert!( state.take_frame().is_none(), "packet {index} queued extra frames" @@ -826,25 +850,57 @@ mod tests { } } + #[test] + fn a_decoded_frame_holds_only_its_samples() { + let packets = tone_packets(1, true); + let mut state = audio_decode(); + + state.push_packet(0, rtp(0), &packets[0]); + + let frame = state.take_frame().expect("decoded frame"); + assert_eq!(frame.samples.capacity(), frame.samples.len()); + } + #[test] fn a_lost_packet_is_rebuilt_from_the_next_one() { let packets = tone_packets(3, true); let mut state = audio_decode(); - state.push_packet(0, &packets[0]); + state.push_packet(0, rtp(0), &packets[0]); assert!(state.take_frame().is_some()); // Packet 1 never arrives; packet 2 carries a copy of frame 1. - state.push_packet(2, &packets[2]); + state.push_packet(2, rtp(2), &packets[2]); let rebuilt = state.take_frame().expect("rebuilt frame"); let current = state.take_frame().expect("current frame"); assert!(state.take_frame().is_none(), "only two frames are owed"); - assert_eq!(rebuilt.len(), FRAME_SAMPLES_20MS); - assert_eq!(current.len(), FRAME_SAMPLES_20MS); + assert_eq!(rebuilt.samples.len(), FRAME_SAMPLES_20MS); + assert_eq!(current.samples.len(), FRAME_SAMPLES_20MS); + assert_eq!((rebuilt.pts, current.pts), (Some(rtp(1)), Some(rtp(2)))); assert!( - peak(&rebuilt) > 1_000, + peak(&rebuilt.samples) > 1_000, "rebuilt frame is silent (peak {})", - peak(&rebuilt) + peak(&rebuilt.samples) + ); + } + + #[test] + fn timestamps_continue_over_a_lost_packet_and_the_rtp_wrap() { + let packets = tone_packets(3, true); + let mut state = audio_decode(); + let first = u32::MAX - (FRAME_SAMPLES_20MS as u32 - 1); + + state.push_packet(0, first, &packets[0]); + // Packet 1 is lost; its timestamp wraps to 0. + state.push_packet(2, first.wrapping_add(rtp(2)), &packets[2]); + + let mut timestamps = Vec::new(); + while let Some(frame) = state.take_frame() { + timestamps.push(frame.pts); + } + assert_eq!( + timestamps, + [Some(first), Some(0), Some(FRAME_SAMPLES_20MS as u32)] ); } @@ -853,14 +909,14 @@ mod tests { let packets = tone_packets(3, false); let mut state = audio_decode(); - state.push_packet(0, &packets[0]); + state.push_packet(0, rtp(0), &packets[0]); assert!(state.take_frame().is_some()); - state.push_packet(2, &packets[2]); + state.push_packet(2, rtp(2), &packets[2]); assert_eq!( ( - state.take_frame().map(|frame| frame.len()), - state.take_frame().map(|frame| frame.len()) + state.take_frame().map(|frame| frame.samples.len()), + state.take_frame().map(|frame| frame.samples.len()) ), (Some(FRAME_SAMPLES_20MS), Some(FRAME_SAMPLES_20MS)), "a lost packet still owes two frames without FEC" @@ -881,13 +937,13 @@ mod tests { let packets = tone_packets(4, true); let mut state = audio_decode(); - state.push_packet(0, &packets[0]); + state.push_packet(0, rtp(0), &packets[0]); // Packet 1 is overtaken by 2, so its frame is rebuilt here. - state.push_packet(2, &packets[2]); + state.push_packet(2, rtp(2), &packets[2]); let before_late = drain(&mut state); - state.push_packet(1, &packets[1]); + state.push_packet(1, rtp(1), &packets[1]); let late = drain(&mut state); - state.push_packet(3, &packets[3]); + state.push_packet(3, rtp(3), &packets[3]); let after_late = drain(&mut state); assert_eq!(late, 0, "a late packet must not repeat a frame"); @@ -903,9 +959,9 @@ mod tests { let packets = tone_packets(1, true); let mut state = audio_decode(); - state.push_packet(7, &packets[0]); + state.push_packet(7, rtp(7), &packets[0]); let first = drain(&mut state); - state.push_packet(7, &packets[0]); + state.push_packet(7, rtp(7), &packets[0]); let second = drain(&mut state); assert_eq!((first, second), (1, 0)); @@ -916,10 +972,10 @@ mod tests { let packets = tone_packets(5, true); let mut state = audio_decode(); - state.push_packet(0, &packets[0]); + state.push_packet(0, rtp(0), &packets[0]); assert!(state.take_frame().is_some()); // Packets 1, 2 and 3 are lost. - state.push_packet(4, &packets[4]); + state.push_packet(4, rtp(4), &packets[4]); let mut frames = 0; while state.take_frame().is_some() { @@ -933,9 +989,10 @@ mod tests { let packets = tone_packets(2, true); let mut state = audio_decode(); - state.push_packet(0, &packets[0]); + state.push_packet(0, rtp(0), &packets[0]); assert!(state.take_frame().is_some()); - state.push_packet(AUDIO_MAX_FILLED_PACKETS + 2, &packets[1]); + let far = AUDIO_MAX_FILLED_PACKETS + 2; + state.push_packet(far, rtp(far), &packets[1]); assert!(state.take_frame().is_some(), "the arriving packet decodes"); assert!( @@ -949,9 +1006,9 @@ mod tests { let packets = tone_packets(3, true); let mut state = audio_decode(); - state.push_packet(9, &packets[0]); + state.push_packet(9, rtp(9), &packets[0]); assert_eq!(drain(&mut state), 1); - state.push_packet(4, &packets[1]); + state.push_packet(4, rtp(4), &packets[1]); assert_eq!(drain(&mut state), 0); } @@ -960,7 +1017,7 @@ mod tests { fn a_corrupt_payload_queues_nothing() { let mut state = audio_decode(); - state.push_packet(0, &[0xff; 4]); + state.push_packet(0, rtp(0), &[0xff; 4]); assert!(state.take_frame().is_none()); } diff --git a/src/video/call.rs b/src/video/call.rs index 4560543..9db5265 100644 --- a/src/video/call.rs +++ b/src/video/call.rs @@ -841,15 +841,22 @@ impl Call { /// Mints a finite, call-CID-scoped user token internally from the server secret, runs the /// coordinator join, establishes the publisher/subscriber PeerConnections, /// and completes the SFU handshake. Illegal (typed error) if already - /// `JOINING`/`JOINED`. Observe participants via [`Call::subscribe`]. + /// `JOINING`/`JOINED`. Observe participants via [`Call::sfu_events`]. pub async fn join(&self, data: crate::rtc::JoinCallData) -> crate::rtc::RtcResult<()> { - let source = crate::rtc::client::UserTokenSource::ServerMinted { - client: self.client.clone(), - user_id: data.user_id.clone(), - call_cid: self.cid(), - expiration: INTERNAL_RTC_TOKEN_LIFETIME, - }; - self.rtc.join_with_token_source(source, data).await + self.rtc().join(data).await + } + + /// The participant session of this handle as an [`RtcCall`](crate::rtc::RtcCall). + /// Both share one session; its join mints the user token as [`Call::join`] does. + pub fn rtc(&self) -> crate::rtc::RtcCall { + crate::rtc::RtcCall::new( + self.rtc.clone(), + crate::rtc::client::UserTokenSource::ServerMinted { + client: self.client.clone(), + call_cid: self.cid(), + expiration: INTERNAL_RTC_TOKEN_LIFETIME, + }, + ) } /// Leave the call, closing the SFU connection and PeerConnections. Succeeds @@ -858,23 +865,25 @@ impl Call { self.rtc.leave("user requested leave").await } - /// Subscribe to the typed SFU event stream (participant joined/left, tracks, - /// errors). Subscribe before or after [`Call::join`]. - pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver { - self.rtc.subscribe() + /// Subscribe to the events from the SFU (participant joined/left, tracks, + /// errors). A receiver gets only events sent after it subscribes. Subscribe + /// before [`Call::join`] to get the join events, or read [`Call::participants`]. + pub fn sfu_events(&self) -> tokio::sync::broadcast::Receiver { + self.rtc.sfu_events() } - /// Register a callback for typed call events. - pub fn on(&self, callback: F) -> tokio::task::AbortHandle - where - F: Fn(crate::rtc::CallEvent) + Send + 'static, - { - self.rtc.on(callback) + /// Subscribe to the call-scoped coordinator events. A receiver gets only + /// events sent after it subscribes. + pub fn coordinator_events( + &self, + ) -> tokio::sync::broadcast::Receiver { + self.rtc.coordinator_events() } - /// Remove a callback registered with [`Call::on`]. - pub fn off(&self, handler: &tokio::task::AbortHandle) { - self.rtc.off(handler); + /// Subscribe to the events that the SDK itself produces. See + /// [`RtcCore::client_events`](crate::rtc::RtcCore::client_events). + pub fn client_events(&self) -> tokio::sync::broadcast::Receiver { + self.rtc.client_events() } /// The current calling state (`Idle` / `Joining` / `Joined` / …). diff --git a/tests/rtc_join.rs b/tests/rtc_join.rs index ca839df..6a79f9b 100644 --- a/tests/rtc_join.rs +++ b/tests/rtc_join.rs @@ -13,13 +13,13 @@ use std::time::Duration; use getstream::TokenOptions; use getstream::models::UserRequest; use getstream::models::{CallRequest, DeleteCallRequest, GetOrCreateCallRequest, MemberRequest}; -use getstream::rtc::{CallEvent, JoinCallData, RtcClient}; +use getstream::rtc::{CallingState, ClientCallEvent, JoinCallData, RtcClient, SfuCallEvent}; use tokio::sync::broadcast::Receiver; /// Wait (up to `timeout`) for a `ParticipantJoined` whose `user_id` matches /// `other`. Returns `true` if observed. async fn observe_participant( - mut rx: Receiver, + mut rx: Receiver, other: String, timeout: Duration, ) -> bool { @@ -29,7 +29,7 @@ async fn observe_participant( tokio::select! { () = &mut deadline => return false, event = rx.recv() => match event { - Ok(CallEvent::ParticipantJoined(p)) if p.user_id == other => return true, + Ok(SfuCallEvent::ParticipantJoined(p)) if p.user_id == other => return true, Ok(_) => continue, // Lagged: keep waiting; the join event may still arrive. Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, @@ -96,11 +96,12 @@ async fn two_sessions_join_and_observe_each_other() { // The whole join/observe/leave dance must finish well within a minute. let outcome = tokio::time::timeout(Duration::from_secs(90), async { let call_a = client.video().call("default", &call_id); - let call_b = client.video().call("default", &call_id); + // Session B joins through the `RtcCall` view of a server-client call. + let call_b = client.video().call("default", &call_id).rtc(); // Subscribe BEFORE joining so no participant event is missed. - let rx_a = call_a.subscribe(); - let rx_b = call_b.subscribe(); + let rx_a = call_a.sfu_events(); + let rx_b = call_b.sfu_events(); // Session A joins first. call_a @@ -191,9 +192,11 @@ async fn provider_backed_client_loads_token_and_joins() { } /// Local decoding is not sufficient proof of authenticity: Stream must reject -/// a validly shaped participant token whose HS256 signature was altered. +/// a validly shaped participant token whose HS256 signature was altered. A +/// valid pre-minted token joins a call handle made before the join, and a +/// receiver subscribed before the join gets the join states. #[tokio::test] -async fn participant_token_signature_is_enforced() { +async fn preminted_token_client_gets_join_events_and_signature_is_enforced() { let Some(client) = common::client_or_skip() else { return; }; @@ -232,9 +235,22 @@ async fn participant_token_signature_is_enforced() { let outcome: Result<(), String> = tokio::time::timeout(Duration::from_secs(120), async { let allowed = RtcClient::new(client.api_key(), token) .map_err(|error| format!("build RTC client: {error}"))? - .join("default", &call_id, JoinCallData::new(&user_id)) + .call("default", &call_id); + let mut events = allowed.client_events(); + allowed + .join(JoinCallData::new(&user_id)) .await .map_err(|error| format!("valid token failed to join: {error}"))?; + let mut states = Vec::new(); + while let Ok(event) = events.try_recv() { + if let ClientCallEvent::CallingStateChanged(state) = event { + states.push(state); + } + } + if states != [CallingState::Joining, CallingState::Joined] { + let _ = allowed.leave().await; + return Err(format!("join states seen before the join: {states:?}")); + } allowed .leave() .await diff --git a/tests/rtc_media.rs b/tests/rtc_media.rs index 022cb1c..875950d 100644 --- a/tests/rtc_media.rs +++ b/tests/rtc_media.rs @@ -21,8 +21,9 @@ use getstream::models::UserRequest; use getstream::models::{CallRequest, DeleteCallRequest, GetOrCreateCallRequest, MemberRequest}; use getstream::rtc::proto::models::TrackType; use getstream::rtc::{ - CallEvent, ClientPublishOptions, JoinCallData, LocalAudioTrack, LocalTrack, LocalVideoTrack, - PcmFrame, PreferredVideoCodec, RemoteTrack, RtcError, SubscriptionConfig, VideoFrame, + CallingState, ClientCallEvent, ClientPublishOptions, JoinCallData, LocalAudioTrack, LocalTrack, + LocalVideoTrack, PcmFrame, PreferredVideoCodec, RemoteTrack, RtcError, SfuCallEvent, + SubscriptionConfig, SubscriptionTarget, TrackSubscriptionConfig, VideoFrame, }; use getstream::video::Call; use tokio::sync::mpsc::{Receiver, channel}; @@ -373,24 +374,23 @@ async fn drain_rms(remote: &RemoteTrack, target: usize, overall: Duration) -> f6 /// `TrackUnpublished` (`published == false`) event for `user`/`track_type`, /// draining unrelated events. Returns whether the event was observed. async fn await_track_event( - events: &mut tokio::sync::broadcast::Receiver, + events: &mut tokio::sync::broadcast::Receiver, user: &str, track_type: TrackType, published: bool, timeout: Duration, ) -> bool { use tokio::sync::broadcast::error::RecvError; - let want = track_type as i32; let deadline = tokio::time::sleep(timeout); tokio::pin!(deadline); loop { tokio::select! { () = &mut deadline => return false, recv = events.recv() => match recv { - Ok(CallEvent::TrackPublished { user_id, track_type: tt, .. }) - if published && user_id == user && tt == want => return true, - Ok(CallEvent::TrackUnpublished { user_id, track_type: tt, .. }) - if !published && user_id == user && tt == want => return true, + Ok(SfuCallEvent::TrackPublished { user_id, track_type: tt, .. }) + if published && user_id == user && tt == track_type => return true, + Ok(SfuCallEvent::TrackUnpublished { user_id, track_type: tt, .. }) + if !published && user_id == user && tt == track_type => return true, Ok(_) | Err(RecvError::Lagged(_)) => {} Err(RecvError::Closed) => return false, } @@ -914,6 +914,62 @@ async fn publish_blue_video_reaches_raw_rtp_and_i420_decoder() { outcome.expect("VP9 RTP/decode test timed out"); } +/// A video target without a dimension hint is accepted and delivers video. +#[tokio::test] +async fn video_target_without_a_dimension_receives_video() { + let Some(client) = common::client_or_skip() else { + return; + }; + init_tracing(); + + let user_a = common::unique_id("a"); + let user_b = common::unique_id("b"); + let (admin, call_id) = setup_call(&client, &[&user_a, &user_b]).await; + + let outcome = tokio::time::timeout(Duration::from_secs(120), async { + let call_a = client.video().call("default", &call_id); + let call_b = client.video().call("default", &call_id); + + call_a + .join(JoinCallData::new(&user_a)) + .await + .expect("A join"); + let video_a = LocalVideoTrack::vp9().expect("vp9 track"); + call_a + .publish_video(video_a.clone()) + .await + .expect("A publish_video"); + let feeder = spawn_blue_video(video_a); + + let mut rx_b = track_sink(&call_b); + call_b + .join(JoinCallData::new(&user_b)) + .await + .expect("B join"); + let session_a = call_a.session_id().await.expect("A session id"); + call_b + .update_subscription_targets(vec![SubscriptionTarget::new(session_a, TrackType::Video)]) + .await + .expect("B update_subscription_targets without a dimension"); + recv_track( + &mut rx_b, + &user_a, + TrackType::Video, + Duration::from_secs(60), + ) + .await + .expect("B did not receive A's video track"); + + feeder.abort(); + call_a.leave().await.expect("A leave"); + call_b.leave().await.expect("B leave"); + }) + .await; + + let _ = admin.delete(DeleteCallRequest { hard: Some(true) }).await; + outcome.expect("dimensionless video subscription test timed out"); +} + #[tokio::test] async fn vp9_svc_preserves_one_ssrc_and_adapts_all_spatial_layers() { let Some(client) = common::client_or_skip() else { @@ -995,10 +1051,12 @@ async fn vp9_svc_preserves_one_ssrc_and_adapts_all_spatial_layers() { // transition accidentally. call_b .update_subscriptions(SubscriptionConfig { - audio: false, - video: true, - screen_share: false, - video_dimension: Some((320, 180)), + default: TrackSubscriptionConfig { + track_types: vec![TrackType::Video], + video_dimension: (320, 180), + ..Default::default() + }, + ..Default::default() }) .await .map_err(|error| format!("VP9 SVC low-quality subscription failed: {error}"))?; @@ -1016,10 +1074,12 @@ async fn vp9_svc_preserves_one_ssrc_and_adapts_all_spatial_layers() { // picture with truthful SS dimensions after the encoder reconfiguration. call_b .update_subscriptions(SubscriptionConfig { - audio: false, - video: true, - screen_share: false, - video_dimension: Some((1280, 720)), + default: TrackSubscriptionConfig { + track_types: vec![TrackType::Video], + video_dimension: (1280, 720), + ..Default::default() + }, + ..Default::default() }) .await .map_err(|error| format!("VP9 SVC high-quality subscription failed: {error}"))?; @@ -1160,7 +1220,7 @@ async fn publish_h264_video_b_decodes_i420_frame() { /// Await both an `AudioLevelChanged` naming `session` as speaking and a /// `DominantSpeakerChanged` naming it, within `timeout`. async fn await_speaking( - events: &mut tokio::sync::broadcast::Receiver, + events: &mut tokio::sync::broadcast::Receiver, session: &str, timeout: Duration, ) -> (bool, bool) { @@ -1174,7 +1234,7 @@ async fn await_speaking( tokio::select! { () = &mut deadline => return (level_seen, dominant_seen), received = events.recv() => match received { - Ok(CallEvent::AudioLevelChanged(levels)) => { + Ok(SfuCallEvent::AudioLevelChanged(levels)) => { if levels .iter() .any(|l| l.session_id == session && l.is_speaking) @@ -1182,7 +1242,7 @@ async fn await_speaking( level_seen = true; } } - Ok(CallEvent::DominantSpeakerChanged { session_id, .. }) => { + Ok(SfuCallEvent::DominantSpeakerChanged { session_id, .. }) => { if session_id == session { dominant_seen = true; } @@ -1223,7 +1283,7 @@ async fn loud_publisher_is_reported_speaking_and_dominant() { // Join and subscribe before any participant publishes audio. A speaker // selected before B joins is present in JoinResponse, and the SFU does // not replay the earlier DominantSpeakerChanged event to B. - let mut events_b = call_b.subscribe(); + let mut events_b = call_b.sfu_events(); call_b .join(JoinCallData::new(&user_b)) .await @@ -1294,8 +1354,11 @@ async fn loud_publisher_is_reported_speaking_and_dominant() { /// typed event stream: it must see A's audio `TrackPublished`, then — once A /// stops the track — A's audio `TrackUnpublished`, confirming the SFU accepted /// the request and broadcast the corrected publication state to peers. +/// +/// A then publishes a new audio track on the same session. The publish must +/// succeed without a reconnect, and B must hear A again. #[tokio::test] -async fn stop_publish_of_sole_audio_is_accepted_and_unannounced() { +async fn sole_audio_can_be_stopped_and_published_again() { let Some(client) = common::client_or_skip() else { return; }; @@ -1309,6 +1372,7 @@ async fn stop_publish_of_sole_audio_is_accepted_and_unannounced() { let call_a = client.video().call("default", &call_id); let call_b = client.video().call("default", &call_id); + let mut tracks_b = track_sink(&call_b); call_b .join(JoinCallData::new(&user_b)) .await @@ -1317,12 +1381,13 @@ async fn stop_publish_of_sole_audio_is_accepted_and_unannounced() { .update_subscriptions(SubscriptionConfig::audio_all()) .await .expect("B update_subscriptions"); - let mut events_b = call_b.subscribe(); + let mut events_b = call_b.sfu_events(); call_a .join(JoinCallData::new(&user_a)) .await .expect("A join"); + let mut events_a = call_a.client_events(); let audio_a = LocalAudioTrack::opus().expect("opus track"); call_a .publish_audio(audio_a.clone()) @@ -1342,6 +1407,14 @@ async fn stop_publish_of_sole_audio_is_accepted_and_unannounced() { audio_published, "B never received A's audio TrackPublished before the stop" ); + let remote_a = recv_track( + &mut tracks_b, + &user_a, + TrackType::Audio, + Duration::from_secs(45), + ) + .await + .expect("B never received A's audio track"); // Stop the sole publication. Before the fix this renegotiated the // publisher with an empty track set and the SFU returned "Invalid @@ -1366,6 +1439,42 @@ async fn stop_publish_of_sole_audio_is_accepted_and_unannounced() { (mute state not propagated to peers)" ); + // Publish a new audio track on the same session, as after an unmute. + let audio_again = LocalAudioTrack::opus().expect("opus track"); + call_a + .publish_audio(audio_again.clone()) + .await + .expect("A publish_audio again on the same session"); + let feeder_again = spawn_tone(audio_again); + let audio_republished = await_track_event( + &mut events_b, + &user_a, + TrackType::Audio, + true, + Duration::from_secs(45), + ) + .await; + assert!( + audio_republished, + "B never received A's audio TrackPublished after the second publish" + ); + let rms = drain_rms(&remote_a, FRAME_20MS * 100, Duration::from_secs(30)).await; + feeder_again.abort(); + assert!( + rms > NON_SILENT_RMS, + "B got no audio after the second publish (rms={rms:.4})" + ); + let mut reconnected = false; + while let Ok(event) = events_a.try_recv() { + reconnected |= matches!( + event, + ClientCallEvent::CallingStateChanged( + CallingState::Reconnecting | CallingState::Migrating + ) + ); + } + assert!(!reconnected, "A reconnected after the second publish"); + call_a.leave().await.expect("A leave"); call_b.leave().await.expect("B leave"); }) diff --git a/tests/video_rest.rs b/tests/video_rest.rs index 27479b2..f6aa8fa 100644 --- a/tests/video_rest.rs +++ b/tests/video_rest.rs @@ -13,7 +13,7 @@ use getstream::models::{ StartClosedCaptionsRequest, StartFrameRecordingRequest, StopClosedCaptionsRequest, UpdateCallMembersRequest, UserRequest, }; -use getstream::rtc::{CallEvent, JoinCallData, LocalAudioTrack, LocalTrack, RtcError}; +use getstream::rtc::{CoordinatorEvent, JoinCallData, LocalAudioTrack, LocalTrack, RtcError}; use std::time::Duration; /// End-to-end call lifecycle: create → get → update members → query → end → delete. @@ -192,7 +192,7 @@ async fn audio_room_send_audio_permission_controls_publishing() { .expect("take permission test audio room live"); let participant = client.video().call("audio_room", &call_id); let outcome: Result<(), String> = tokio::time::timeout(Duration::from_secs(120), async { - let mut events = participant.subscribe(); + let mut events = participant.coordinator_events(); participant .join(JoinCallData::new(&publisher_id)) .await @@ -247,7 +247,7 @@ async fn audio_room_send_audio_permission_controls_publishing() { } async fn wait_for_audio_permission( - events: &mut tokio::sync::broadcast::Receiver, + events: &mut tokio::sync::broadcast::Receiver, expected: bool, ) -> Result<(), String> { let mut last_update = None; @@ -257,9 +257,6 @@ async fn wait_for_audio_permission( .recv() .await .map_err(|error| format!("permission event stream closed: {error}"))?; - let CallEvent::Coordinator(event) = event else { - continue; - }; if event.event_type != "call.permissions_updated" { continue; } @@ -332,10 +329,7 @@ async fn scoped_participant_service_lifecycle() { .await .map_err(|error| format!("enable incoming video failed: {error}"))?; - let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel(); - let handler = call.on(move |event| { - let _ = event_tx.send(event); - }); + let mut events = call.coordinator_events(); let mut custom = getstream::models::CustomData::new(); custom.insert("source".to_owned(), serde_json::json!("rust-live-test")); server_call @@ -347,9 +341,8 @@ async fn scoped_participant_service_lifecycle() { .await .map_err(|error| format!("send_custom_event failed: {error}"))?; let custom_event = tokio::time::timeout(Duration::from_secs(10), async { - while let Some(event) = event_rx.recv().await { - if let getstream::rtc::CallEvent::Coordinator(event) = event - && event.event_type == "custom" + while let Ok(event) = events.recv().await { + if event.event_type == "custom" && event .raw .pointer("/custom/source") @@ -359,7 +352,7 @@ async fn scoped_participant_service_lifecycle() { return Ok(()); } } - Err("call event handler closed before receiving custom event".to_owned()) + Err("coordinator event stream ended before the custom event".to_owned()) }) .await .map_err(|_| "timed out waiting for custom coordinator event".to_owned()) @@ -382,9 +375,8 @@ async fn scoped_participant_service_lifecycle() { .await .map_err(|error| format!("send_closed_caption failed: {error}"))?; tokio::time::timeout(Duration::from_secs(10), async { - while let Some(event) = event_rx.recv().await { - if let getstream::rtc::CallEvent::Coordinator(event) = event - && event.event_type == "call.closed_caption" + while let Ok(event) = events.recv().await { + if event.event_type == "call.closed_caption" && event .raw .pointer("/closed_caption/text") @@ -394,14 +386,13 @@ async fn scoped_participant_service_lifecycle() { return Ok(()); } } - Err("call event handler closed before receiving closed caption".to_owned()) + Err("coordinator event stream ended before the closed caption".to_owned()) }) .await .map_err(|_| "timed out waiting for closed caption event".to_owned()) .and_then(|result| result) } .await; - call.off(&handler); caption_result?; call.send_reaction(SendVideoReactionRequest {