diff --git a/docs/api.md b/docs/api.md index eefb6a9..4e256f9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -683,6 +683,33 @@ The bridge is a media-only operation. It creates separate internal bridge tracks } ``` +### ICE Commands + +#### Add Ice Candidate Command +**Purpose:** Trickle ICE - feeds a remote ICE candidate into an already-established WebRTC session, instead of waiting for the caller to gather all candidates before sending the offer. + +Client -> server only. The server always answers with a fully-gathered candidate set (server-side gathering is fast), so there's no matching server -> client event; this command exists purely so a WebRTC caller can send its offer immediately and stream candidates in as they're found. + +**Fields:** +- `command` (string): Always "addIceCandidate" +- `candidate` (string): ICE candidate string (as produced by `RTCIceCandidate.candidate` in the browser) +- `sdpMid` (string, optional): Media stream identification +- `sdpMLineIndex` (number, optional): Index of the media description this candidate is associated with + +```json +{ + "command": "addIceCandidate", + "candidate": "candidate:842163049 1 udp 1677729535 10.0.0.1 54321 typ host", + "sdpMid": "0", + "sdpMLineIndex": 0 +} +``` + +**Notes:** +- Only meaningful once the initial offer/answer has already been exchanged (i.e. after `invite`/`accept`). +- Applies to whichever track on the session is WebRTC-backed; a no-op on other track types. +- A candidate sent before the PeerConnection exists is rejected with an error. + ### CallOption Object Structure The `CallOption` object is used in `invite` and `accept` commands and contains the following fields: diff --git a/src/call/active_call.rs b/src/call/active_call.rs index 20e6132..f53feea 100644 --- a/src/call/active_call.rs +++ b/src/call/active_call.rs @@ -1055,6 +1055,15 @@ impl ActiveCall { } => self.do_interrupt(passage.unwrap_or_default()).await, Command::History { speaker, text } => self.do_history(speaker, text).await, Command::Custom { sender, data } => self.do_custom(sender, data), + Command::AddIceCandidate { + candidate, + sdp_mid, + sdp_mline_index, + } => { + self.media_stream + .add_ice_candidate(&candidate, sdp_mid.as_deref(), sdp_mline_index) + .await + } } } diff --git a/src/call/mod.rs b/src/call/mod.rs index 1b0c06b..a4dac45 100644 --- a/src/call/mod.rs +++ b/src/call/mod.rs @@ -121,6 +121,12 @@ pub enum Command { sender: Option, data: serde_json::Value, }, + /// Trickle ICE: feed a remote candidate into an already-established session. + AddIceCandidate { + candidate: String, + sdp_mid: Option, + sdp_mline_index: Option, + }, } /// Routing state for managing stateful load balancing diff --git a/src/media/stream.rs b/src/media/stream.rs index 8f4c6b6..745f22b 100644 --- a/src/media/stream.rs +++ b/src/media/stream.rs @@ -25,6 +25,9 @@ pub struct MediaStream { pub cancel_token: CancellationToken, recorder_option: Mutex>, tracks: Mutex, DtmfDetector)>>, + /// Trickle ICE candidates that arrived before any track existed to feed + /// them to. Drained into the next track started. + pending_ice_candidates: Mutex, Option)>>, suppressed_sources: Mutex>, event_sender: EventSender, pub packet_sender: TrackPacketSender, @@ -81,6 +84,7 @@ impl MediaStreamBuilder { cancel_token, recorder_option: Mutex::new(self.recorder_config), tracks, + pending_ice_candidates: Mutex::new(Vec::new()), suppressed_sources: Mutex::new(HashSet::new()), event_sender: self.event_sender, packet_sender: track_packet_sender, @@ -234,6 +238,21 @@ impl MediaStream { Ok(_) => { info!(session_id = self.id, track_id = track.id(), "track started"); let track_id = track.id().clone(); + if track_id.as_str() == self.id.as_str() { + let pending = std::mem::take(&mut *self.pending_ice_candidates.lock().await); + for (candidate, sdp_mid, sdp_mline_index) in pending { + if let Err(e) = + track.add_ice_candidate(&candidate, sdp_mid.as_deref(), sdp_mline_index) + { + warn!( + session_id = self.id, + track_id = track.id(), + "failed to apply buffered ICE candidate: {}", + e + ); + } + } + } self.tracks .lock() .await @@ -282,6 +301,29 @@ impl MediaStream { } } + /// Trickle ICE: feed a remote candidate into the ICE-backed (WebRTC) + /// track, if it's up yet, or buffer it for `update_track` to replay + /// otherwise. + pub async fn add_ice_candidate( + &self, + candidate: &str, + sdp_mid: Option<&str>, + sdp_mline_index: Option, + ) -> Result<()> { + let tracks = self.tracks.lock().await; + if let Some((track, _)) = tracks.get(self.id.as_str()) { + track.add_ice_candidate(candidate, sdp_mid, sdp_mline_index)?; + return Ok(()); + } + drop(tracks); + self.pending_ice_candidates.lock().await.push(( + candidate.to_string(), + sdp_mid.map(|s| s.to_string()), + sdp_mline_index, + )); + Ok(()) + } + pub async fn pause_playback(&self, id: TrackId) -> Result<()> { self.set_playback_paused(id, true).await } @@ -370,7 +412,6 @@ impl MediaStream { Err(anyhow::anyhow!("Track {} not found", track_id)) } } - } #[derive(Clone)] @@ -484,7 +525,12 @@ impl MediaStream { for (track, dtmf_detector) in tracks.values_mut() { if track.id() == &packet.track_id { - if let Samples::RTP { payload_type, payload, .. } = &packet.samples { + if let Samples::RTP { + payload_type, + payload, + .. + } = &packet.samples + { if let Some(digit) = dtmf_detector.detect_rtp(*payload_type, payload) { debug!(track_id = track.id(), digit, "DTMF detected"); event_sender diff --git a/src/media/track/mod.rs b/src/media/track/mod.rs index 3299f4c..7505291 100644 --- a/src/media/track/mod.rs +++ b/src/media/track/mod.rs @@ -91,4 +91,14 @@ pub trait Track: Send + Sync { self.stop().await } async fn send_packet(&mut self, packet: &AudioFrame) -> Result<()>; + /// Feed a remotely-gathered ICE candidate into this track's PeerConnection. + /// Default no-op for track types that aren't backed by ICE (file/tts/websocket tracks). + fn add_ice_candidate( + &self, + _candidate: &str, + _sdp_mid: Option<&str>, + _sdp_mline_index: Option, + ) -> Result<()> { + Ok(()) + } } diff --git a/src/media/track/rtc.rs b/src/media/track/rtc.rs index c18ad04..a8e9e46 100644 --- a/src/media/track/rtc.rs +++ b/src/media/track/rtc.rs @@ -13,7 +13,7 @@ use audio_codec::CodecType; use bytes::Bytes; use futures::{FutureExt, StreamExt, stream::FuturesUnordered}; use rustrtc::{ - AudioCapability, IceServer, MediaKind, PeerConnection, PeerConnectionEvent, + AudioCapability, IceCandidate, IceServer, MediaKind, PeerConnection, PeerConnectionEvent, PeerConnectionState, RtcConfiguration, RtpCodecParameters, SdpType, TransportMode, config::MediaCapabilities, media::{ @@ -869,6 +869,20 @@ impl Track for RtcTrack { } Ok(()) } + + fn add_ice_candidate( + &self, + candidate: &str, + // single audio m-line per track, so unused here + _sdp_mid: Option<&str>, + _sdp_mline_index: Option, + ) -> Result<()> { + let pc = self.peer_connection.as_ref().ok_or_else(|| { + anyhow::anyhow!("No PeerConnection available for track {}", self.track_id) + })?; + pc.add_ice_candidate(IceCandidate::from_sdp(candidate)?)?; + Ok(()) + } } impl RtcTrack {