1 use crate::api::setting_engine::SettingEngine; 2 use crate::error::{Error, Result}; 3 use crate::ice_transport::ice_candidate::*; 4 use crate::ice_transport::ice_candidate_type::RTCIceCandidateType; 5 use crate::ice_transport::ice_gatherer_state::RTCIceGathererState; 6 use crate::ice_transport::ice_parameters::RTCIceParameters; 7 use crate::ice_transport::ice_server::RTCIceServer; 8 use crate::peer_connection::policy::ice_transport_policy::RTCIceTransportPolicy; 9 use crate::stats::stats_collector::StatsCollector; 10 use crate::stats::SourceStatsType::*; 11 use crate::stats::{ICECandidatePairStats, StatsReportType}; 12 13 use ice::agent::Agent; 14 use ice::candidate::{Candidate, CandidateType}; 15 use ice::url::Url; 16 17 use arc_swap::ArcSwapOption; 18 use std::collections::HashMap; 19 use std::future::Future; 20 use std::pin::Pin; 21 use std::sync::atomic::{AtomicU8, Ordering}; 22 use std::sync::Arc; 23 use tokio::sync::Mutex; 24 25 /// ICEGatherOptions provides options relating to the gathering of ICE candidates. 26 #[derive(Default, Debug, Clone)] 27 pub struct RTCIceGatherOptions { 28 pub ice_servers: Vec<RTCIceServer>, 29 pub ice_gather_policy: RTCIceTransportPolicy, 30 } 31 32 pub type OnLocalCandidateHdlrFn = Box< 33 dyn (FnMut(Option<RTCIceCandidate>) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 34 + Send 35 + Sync, 36 >; 37 38 pub type OnICEGathererStateChangeHdlrFn = Box< 39 dyn (FnMut(RTCIceGathererState) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 40 + Send 41 + Sync, 42 >; 43 44 pub type OnGatheringCompleteHdlrFn = 45 Box<dyn (FnMut() -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) + Send + Sync>; 46 47 /// ICEGatherer gathers local host, server reflexive and relay 48 /// candidates, as well as enabling the retrieval of local Interactive 49 /// Connectivity Establishment (ICE) parameters which can be 50 /// exchanged in signaling. 51 #[derive(Default)] 52 pub struct RTCIceGatherer { 53 pub(crate) validated_servers: Vec<Url>, 54 pub(crate) gather_policy: RTCIceTransportPolicy, 55 pub(crate) setting_engine: Arc<SettingEngine>, 56 57 pub(crate) state: Arc<AtomicU8>, //ICEGathererState, 58 pub(crate) agent: Mutex<Option<Arc<ice::agent::Agent>>>, 59 60 pub(crate) on_local_candidate_handler: Arc<ArcSwapOption<Mutex<OnLocalCandidateHdlrFn>>>, 61 pub(crate) on_state_change_handler: Arc<ArcSwapOption<Mutex<OnICEGathererStateChangeHdlrFn>>>, 62 63 // Used for gathering_complete_promise 64 pub(crate) on_gathering_complete_handler: Arc<ArcSwapOption<Mutex<OnGatheringCompleteHdlrFn>>>, 65 } 66 67 impl RTCIceGatherer { new( validated_servers: Vec<Url>, gather_policy: RTCIceTransportPolicy, setting_engine: Arc<SettingEngine>, ) -> Self68 pub(crate) fn new( 69 validated_servers: Vec<Url>, 70 gather_policy: RTCIceTransportPolicy, 71 setting_engine: Arc<SettingEngine>, 72 ) -> Self { 73 RTCIceGatherer { 74 gather_policy, 75 validated_servers, 76 setting_engine, 77 state: Arc::new(AtomicU8::new(RTCIceGathererState::New as u8)), 78 ..Default::default() 79 } 80 } 81 create_agent(&self) -> Result<()>82 pub(crate) async fn create_agent(&self) -> Result<()> { 83 // NOTE: A lock is held for the duration of this function in order to 84 // avoid potential double-agent creations. Care should be taken to 85 // ensure we do not do anything expensive other than the actual agent 86 // creation in this function. 87 let mut agent = self.agent.lock().await; 88 89 if agent.is_some() || self.state() != RTCIceGathererState::New { 90 return Ok(()); 91 } 92 93 let mut candidate_types = vec![]; 94 if self.setting_engine.candidates.ice_lite { 95 candidate_types.push(ice::candidate::CandidateType::Host); 96 } else if self.gather_policy == RTCIceTransportPolicy::Relay { 97 candidate_types.push(ice::candidate::CandidateType::Relay); 98 } 99 100 let nat_1to1_cand_type = match self.setting_engine.candidates.nat_1to1_ip_candidate_type { 101 RTCIceCandidateType::Host => CandidateType::Host, 102 RTCIceCandidateType::Srflx => CandidateType::ServerReflexive, 103 _ => CandidateType::Unspecified, 104 }; 105 106 let mdns_mode = self.setting_engine.candidates.multicast_dns_mode; 107 108 let mut config = ice::agent::agent_config::AgentConfig { 109 udp_network: self.setting_engine.udp_network.clone(), 110 lite: self.setting_engine.candidates.ice_lite, 111 urls: self.validated_servers.clone(), 112 disconnected_timeout: self.setting_engine.timeout.ice_disconnected_timeout, 113 failed_timeout: self.setting_engine.timeout.ice_failed_timeout, 114 keepalive_interval: self.setting_engine.timeout.ice_keepalive_interval, 115 candidate_types, 116 host_acceptance_min_wait: self.setting_engine.timeout.ice_host_acceptance_min_wait, 117 srflx_acceptance_min_wait: self.setting_engine.timeout.ice_srflx_acceptance_min_wait, 118 prflx_acceptance_min_wait: self.setting_engine.timeout.ice_prflx_acceptance_min_wait, 119 relay_acceptance_min_wait: self.setting_engine.timeout.ice_relay_acceptance_min_wait, 120 interface_filter: self.setting_engine.candidates.interface_filter.clone(), 121 ip_filter: self.setting_engine.candidates.ip_filter.clone(), 122 nat_1to1_ips: self.setting_engine.candidates.nat_1to1_ips.clone(), 123 nat_1to1_ip_candidate_type: nat_1to1_cand_type, 124 net: self.setting_engine.vnet.clone(), 125 multicast_dns_mode: mdns_mode, 126 multicast_dns_host_name: self 127 .setting_engine 128 .candidates 129 .multicast_dns_host_name 130 .clone(), 131 local_ufrag: self.setting_engine.candidates.username_fragment.clone(), 132 local_pwd: self.setting_engine.candidates.password.clone(), 133 //TODO: TCPMux: self.setting_engine.iceTCPMux, 134 //TODO: ProxyDialer: self.setting_engine.iceProxyDialer, 135 ..Default::default() 136 }; 137 138 let requested_network_types = if self.setting_engine.candidates.ice_network_types.is_empty() 139 { 140 ice::network_type::supported_network_types() 141 } else { 142 self.setting_engine.candidates.ice_network_types.clone() 143 }; 144 145 config.network_types.extend(requested_network_types); 146 147 *agent = Some(Arc::new(ice::agent::Agent::new(config).await?)); 148 149 Ok(()) 150 } 151 152 /// Gather ICE candidates. gather(&self) -> Result<()>153 pub async fn gather(&self) -> Result<()> { 154 self.create_agent().await?; 155 self.set_state(RTCIceGathererState::Gathering).await; 156 157 if let Some(agent) = self.get_agent().await { 158 let state = Arc::clone(&self.state); 159 let on_local_candidate_handler = Arc::clone(&self.on_local_candidate_handler); 160 let on_state_change_handler = Arc::clone(&self.on_state_change_handler); 161 let on_gathering_complete_handler = Arc::clone(&self.on_gathering_complete_handler); 162 163 agent.on_candidate(Box::new( 164 move |candidate: Option<Arc<dyn Candidate + Send + Sync>>| { 165 let state_clone = Arc::clone(&state); 166 let on_local_candidate_handler_clone = Arc::clone(&on_local_candidate_handler); 167 let on_state_change_handler_clone = Arc::clone(&on_state_change_handler); 168 let on_gathering_complete_handler_clone = 169 Arc::clone(&on_gathering_complete_handler); 170 171 Box::pin(async move { 172 if let Some(cand) = candidate { 173 if let Some(handler) = &*on_local_candidate_handler_clone.load() { 174 let mut f = handler.lock().await; 175 f(Some(RTCIceCandidate::from(&cand))).await; 176 } 177 } else { 178 state_clone 179 .store(RTCIceGathererState::Complete as u8, Ordering::SeqCst); 180 181 if let Some(handler) = &*on_state_change_handler_clone.load() { 182 let mut f = handler.lock().await; 183 f(RTCIceGathererState::Complete).await; 184 } 185 186 if let Some(handler) = &*on_gathering_complete_handler_clone.load() { 187 let mut f = handler.lock().await; 188 f().await; 189 } 190 191 if let Some(handler) = &*on_local_candidate_handler_clone.load() { 192 let mut f = handler.lock().await; 193 f(None).await; 194 } 195 } 196 }) 197 }, 198 )); 199 200 agent.gather_candidates()?; 201 } 202 203 Ok(()) 204 } 205 206 /// Close prunes all local candidates, and closes the ports. close(&self) -> Result<()>207 pub async fn close(&self) -> Result<()> { 208 self.set_state(RTCIceGathererState::Closed).await; 209 210 let agent = { 211 let mut agent_opt = self.agent.lock().await; 212 agent_opt.take() 213 }; 214 215 if let Some(agent) = agent { 216 agent.close().await?; 217 } 218 219 Ok(()) 220 } 221 222 /// get_local_parameters returns the ICE parameters of the ICEGatherer. get_local_parameters(&self) -> Result<RTCIceParameters>223 pub async fn get_local_parameters(&self) -> Result<RTCIceParameters> { 224 self.create_agent().await?; 225 226 let (frag, pwd) = if let Some(agent) = self.get_agent().await { 227 agent.get_local_user_credentials().await 228 } else { 229 return Err(Error::ErrICEAgentNotExist); 230 }; 231 232 Ok(RTCIceParameters { 233 username_fragment: frag, 234 password: pwd, 235 ice_lite: false, 236 }) 237 } 238 239 /// get_local_candidates returns the sequence of valid local candidates associated with the ICEGatherer. get_local_candidates(&self) -> Result<Vec<RTCIceCandidate>>240 pub async fn get_local_candidates(&self) -> Result<Vec<RTCIceCandidate>> { 241 self.create_agent().await?; 242 243 let ice_candidates = if let Some(agent) = self.get_agent().await { 244 agent.get_local_candidates().await? 245 } else { 246 return Err(Error::ErrICEAgentNotExist); 247 }; 248 249 Ok(rtc_ice_candidates_from_ice_candidates(&ice_candidates)) 250 } 251 252 /// on_local_candidate sets an event handler which fires when a new local ICE candidate is available 253 /// Take note that the handler is gonna be called with a nil pointer when gathering is finished. on_local_candidate(&self, f: OnLocalCandidateHdlrFn)254 pub fn on_local_candidate(&self, f: OnLocalCandidateHdlrFn) { 255 self.on_local_candidate_handler 256 .store(Some(Arc::new(Mutex::new(f)))); 257 } 258 259 /// on_state_change sets an event handler which fires any time the ICEGatherer changes on_state_change(&self, f: OnICEGathererStateChangeHdlrFn)260 pub fn on_state_change(&self, f: OnICEGathererStateChangeHdlrFn) { 261 self.on_state_change_handler 262 .store(Some(Arc::new(Mutex::new(f)))); 263 } 264 265 /// on_gathering_complete sets an event handler which fires any time the ICEGatherer changes on_gathering_complete(&self, f: OnGatheringCompleteHdlrFn)266 pub fn on_gathering_complete(&self, f: OnGatheringCompleteHdlrFn) { 267 self.on_gathering_complete_handler 268 .store(Some(Arc::new(Mutex::new(f)))); 269 } 270 271 /// State indicates the current state of the ICE gatherer. state(&self) -> RTCIceGathererState272 pub fn state(&self) -> RTCIceGathererState { 273 self.state.load(Ordering::SeqCst).into() 274 } 275 set_state(&self, s: RTCIceGathererState)276 pub async fn set_state(&self, s: RTCIceGathererState) { 277 self.state.store(s as u8, Ordering::SeqCst); 278 279 if let Some(handler) = &*self.on_state_change_handler.load() { 280 let mut f = handler.lock().await; 281 f(s).await; 282 } 283 } 284 get_agent(&self) -> Option<Arc<Agent>>285 pub(crate) async fn get_agent(&self) -> Option<Arc<Agent>> { 286 let agent = self.agent.lock().await; 287 agent.clone() 288 } 289 collect_stats(&self, collector: &StatsCollector)290 pub(crate) async fn collect_stats(&self, collector: &StatsCollector) { 291 if let Some(agent) = self.get_agent().await { 292 let mut reports = HashMap::new(); 293 294 for stats in agent.get_candidate_pairs_stats().await { 295 let stats: ICECandidatePairStats = stats.into(); 296 reports.insert(stats.id.clone(), StatsReportType::CandidatePair(stats)); 297 } 298 299 for stats in agent.get_local_candidates_stats().await { 300 reports.insert( 301 stats.id.clone(), 302 StatsReportType::from(LocalCandidate(stats)), 303 ); 304 } 305 306 for stats in agent.get_remote_candidates_stats().await { 307 reports.insert( 308 stats.id.clone(), 309 StatsReportType::from(RemoteCandidate(stats)), 310 ); 311 } 312 313 collector.merge(reports); 314 } 315 } 316 } 317 318 #[cfg(test)] 319 mod test { 320 use super::*; 321 use crate::api::APIBuilder; 322 use crate::ice_transport::ice_gatherer::RTCIceGatherOptions; 323 use crate::ice_transport::ice_server::RTCIceServer; 324 use tokio::sync::mpsc; 325 326 #[tokio::test] test_new_ice_gatherer_success() -> Result<()>327 async fn test_new_ice_gatherer_success() -> Result<()> { 328 let opts = RTCIceGatherOptions { 329 ice_servers: vec![RTCIceServer { 330 urls: vec!["stun:stun.l.google.com:19302".to_owned()], 331 ..Default::default() 332 }], 333 ..Default::default() 334 }; 335 336 let gatherer = APIBuilder::new().build().new_ice_gatherer(opts)?; 337 338 assert_eq!( 339 gatherer.state(), 340 RTCIceGathererState::New, 341 "Expected gathering state new" 342 ); 343 344 let (gather_finished_tx, mut gather_finished_rx) = mpsc::channel::<()>(1); 345 let gather_finished_tx = Arc::new(Mutex::new(Some(gather_finished_tx))); 346 gatherer.on_local_candidate(Box::new(move |c: Option<RTCIceCandidate>| { 347 let gather_finished_tx_clone = Arc::clone(&gather_finished_tx); 348 Box::pin(async move { 349 if c.is_none() { 350 let mut tx = gather_finished_tx_clone.lock().await; 351 tx.take(); 352 } 353 }) 354 })); 355 356 gatherer.gather().await?; 357 358 let _ = gather_finished_rx.recv().await; 359 360 let params = gatherer.get_local_parameters().await?; 361 362 assert!( 363 !params.username_fragment.is_empty() && !params.password.is_empty(), 364 "Empty local username or password frag" 365 ); 366 367 let candidates = gatherer.get_local_candidates().await?; 368 369 assert!(!candidates.is_empty(), "No candidates gathered"); 370 371 gatherer.close().await?; 372 373 Ok(()) 374 } 375 376 #[tokio::test] test_ice_gather_mdns_candidate_gathering() -> Result<()>377 async fn test_ice_gather_mdns_candidate_gathering() -> Result<()> { 378 let mut s = SettingEngine::default(); 379 s.set_ice_multicast_dns_mode(ice::mdns::MulticastDnsMode::QueryAndGather); 380 381 let gatherer = APIBuilder::new() 382 .with_setting_engine(s) 383 .build() 384 .new_ice_gatherer(RTCIceGatherOptions::default())?; 385 386 let (done_tx, mut done_rx) = mpsc::channel::<()>(1); 387 let done_tx = Arc::new(Mutex::new(Some(done_tx))); 388 gatherer.on_local_candidate(Box::new(move |c: Option<RTCIceCandidate>| { 389 let done_tx_clone = Arc::clone(&done_tx); 390 Box::pin(async move { 391 if let Some(c) = c { 392 if c.address.ends_with(".local") { 393 let mut tx = done_tx_clone.lock().await; 394 tx.take(); 395 } 396 } 397 }) 398 })); 399 400 gatherer.gather().await?; 401 402 let _ = done_rx.recv().await; 403 404 gatherer.close().await?; 405 406 Ok(()) 407 } 408 } 409