1 use std::future::Future; 2 use std::pin::Pin; 3 use std::sync::atomic::{AtomicU8, Ordering}; 4 use std::sync::Arc; 5 6 use arc_swap::ArcSwapOption; 7 use ice::candidate::Candidate; 8 use ice::state::ConnectionState; 9 use tokio::sync::{mpsc, Mutex}; 10 use util::Conn; 11 12 use ice_candidate::RTCIceCandidate; 13 use ice_candidate_pair::RTCIceCandidatePair; 14 use ice_gatherer::RTCIceGatherer; 15 use ice_role::RTCIceRole; 16 17 use crate::error::{flatten_errs, Error, Result}; 18 use crate::ice_transport::ice_parameters::RTCIceParameters; 19 use crate::ice_transport::ice_transport_state::RTCIceTransportState; 20 use crate::mux::endpoint::Endpoint; 21 use crate::mux::mux_func::MatchFunc; 22 use crate::mux::{Config, Mux}; 23 use crate::stats::stats_collector::StatsCollector; 24 use crate::stats::ICETransportStats; 25 use crate::stats::StatsReportType::Transport; 26 27 #[cfg(test)] 28 mod ice_transport_test; 29 30 pub mod ice_candidate; 31 pub mod ice_candidate_pair; 32 pub mod ice_candidate_type; 33 pub mod ice_connection_state; 34 pub mod ice_credential_type; 35 pub mod ice_gatherer; 36 pub mod ice_gatherer_state; 37 pub mod ice_gathering_state; 38 pub mod ice_parameters; 39 pub mod ice_protocol; 40 pub mod ice_role; 41 pub mod ice_server; 42 pub mod ice_transport_state; 43 44 pub type OnConnectionStateChangeHdlrFn = Box< 45 dyn (FnMut(RTCIceTransportState) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 46 + Send 47 + Sync, 48 >; 49 50 pub type OnSelectedCandidatePairChangeHdlrFn = Box< 51 dyn (FnMut(RTCIceCandidatePair) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 52 + Send 53 + Sync, 54 >; 55 56 #[derive(Default)] 57 struct ICETransportInternal { 58 role: RTCIceRole, 59 conn: Option<Arc<dyn Conn + Send + Sync>>, //AgentConn 60 mux: Option<Mux>, 61 cancel_tx: Option<mpsc::Sender<()>>, 62 } 63 64 /// ICETransport allows an application access to information about the ICE 65 /// transport over which packets are sent and received. 66 #[derive(Default)] 67 pub struct RTCIceTransport { 68 pub(crate) gatherer: Arc<RTCIceGatherer>, 69 on_connection_state_change_handler: Arc<ArcSwapOption<Mutex<OnConnectionStateChangeHdlrFn>>>, 70 on_selected_candidate_pair_change_handler: 71 Arc<ArcSwapOption<Mutex<OnSelectedCandidatePairChangeHdlrFn>>>, 72 state: Arc<AtomicU8>, // ICETransportState 73 internal: Mutex<ICETransportInternal>, 74 } 75 76 impl RTCIceTransport { 77 /// creates a new new_icetransport. new(gatherer: Arc<RTCIceGatherer>) -> Self78 pub(crate) fn new(gatherer: Arc<RTCIceGatherer>) -> Self { 79 RTCIceTransport { 80 state: Arc::new(AtomicU8::new(RTCIceTransportState::New as u8)), 81 gatherer, 82 ..Default::default() 83 } 84 } 85 86 /// get_selected_candidate_pair returns the selected candidate pair on which packets are sent 87 /// if there is no selected pair nil is returned get_selected_candidate_pair(&self) -> Option<RTCIceCandidatePair>88 pub async fn get_selected_candidate_pair(&self) -> Option<RTCIceCandidatePair> { 89 if let Some(agent) = self.gatherer.get_agent().await { 90 if let Some(ice_pair) = agent.get_selected_candidate_pair() { 91 let local = RTCIceCandidate::from(&ice_pair.local); 92 let remote = RTCIceCandidate::from(&ice_pair.remote); 93 return Some(RTCIceCandidatePair::new(local, remote)); 94 } 95 } 96 None 97 } 98 99 /// Start incoming connectivity checks based on its configured role. start(&self, params: &RTCIceParameters, role: Option<RTCIceRole>) -> Result<()>100 pub async fn start(&self, params: &RTCIceParameters, role: Option<RTCIceRole>) -> Result<()> { 101 if self.state() != RTCIceTransportState::New { 102 return Err(Error::ErrICETransportNotInNew); 103 } 104 105 self.ensure_gatherer().await?; 106 107 if let Some(agent) = self.gatherer.get_agent().await { 108 let state = Arc::clone(&self.state); 109 110 let on_connection_state_change_handler = 111 Arc::clone(&self.on_connection_state_change_handler); 112 agent.on_connection_state_change(Box::new(move |ice_state: ConnectionState| { 113 let s = RTCIceTransportState::from(ice_state); 114 let on_connection_state_change_handler_clone = 115 Arc::clone(&on_connection_state_change_handler); 116 state.store(s as u8, Ordering::SeqCst); 117 Box::pin(async move { 118 if let Some(handler) = &*on_connection_state_change_handler_clone.load() { 119 let mut f = handler.lock().await; 120 f(s).await; 121 } 122 }) 123 })); 124 125 let on_selected_candidate_pair_change_handler = 126 Arc::clone(&self.on_selected_candidate_pair_change_handler); 127 agent.on_selected_candidate_pair_change(Box::new( 128 move |local: &Arc<dyn Candidate + Send + Sync>, 129 remote: &Arc<dyn Candidate + Send + Sync>| { 130 let on_selected_candidate_pair_change_handler_clone = 131 Arc::clone(&on_selected_candidate_pair_change_handler); 132 let local = RTCIceCandidate::from(local); 133 let remote = RTCIceCandidate::from(remote); 134 Box::pin(async move { 135 if let Some(handler) = 136 &*on_selected_candidate_pair_change_handler_clone.load() 137 { 138 let mut f = handler.lock().await; 139 f(RTCIceCandidatePair::new(local, remote)).await; 140 } 141 }) 142 }, 143 )); 144 145 let role = if let Some(role) = role { 146 role 147 } else { 148 RTCIceRole::Controlled 149 }; 150 151 let (cancel_tx, cancel_rx) = mpsc::channel(1); 152 { 153 let mut internal = self.internal.lock().await; 154 internal.role = role; 155 internal.cancel_tx = Some(cancel_tx); 156 } 157 158 let conn: Arc<dyn Conn + Send + Sync> = match role { 159 RTCIceRole::Controlling => { 160 agent 161 .dial( 162 cancel_rx, 163 params.username_fragment.clone(), 164 params.password.clone(), 165 ) 166 .await? 167 } 168 169 RTCIceRole::Controlled => { 170 agent 171 .accept( 172 cancel_rx, 173 params.username_fragment.clone(), 174 params.password.clone(), 175 ) 176 .await? 177 } 178 179 _ => return Err(Error::ErrICERoleUnknown), 180 }; 181 182 let config = Config { 183 conn: Arc::clone(&conn), 184 buffer_size: self.gatherer.setting_engine.get_receive_mtu(), 185 }; 186 187 { 188 let mut internal = self.internal.lock().await; 189 internal.conn = Some(conn); 190 internal.mux = Some(Mux::new(config)); 191 } 192 193 Ok(()) 194 } else { 195 Err(Error::ErrICEAgentNotExist) 196 } 197 } 198 199 /// restart is not exposed currently because ORTC has users create a whole new ICETransport 200 /// so for now lets keep it private so we don't cause ORTC users to depend on non-standard APIs restart(&self) -> Result<()>201 pub(crate) async fn restart(&self) -> Result<()> { 202 if let Some(agent) = self.gatherer.get_agent().await { 203 agent 204 .restart( 205 self.gatherer 206 .setting_engine 207 .candidates 208 .username_fragment 209 .clone(), 210 self.gatherer.setting_engine.candidates.password.clone(), 211 ) 212 .await?; 213 } else { 214 return Err(Error::ErrICEAgentNotExist); 215 } 216 self.gatherer.gather().await 217 } 218 219 /// Stop irreversibly stops the ICETransport. stop(&self) -> Result<()>220 pub async fn stop(&self) -> Result<()> { 221 self.set_state(RTCIceTransportState::Closed); 222 223 let mut errs: Vec<Error> = vec![]; 224 { 225 let mut internal = self.internal.lock().await; 226 internal.cancel_tx.take(); 227 if let Some(mut mux) = internal.mux.take() { 228 mux.close().await; 229 } 230 if let Some(conn) = internal.conn.take() { 231 if let Err(err) = conn.close().await { 232 errs.push(err.into()); 233 } 234 } 235 } 236 237 if let Err(err) = self.gatherer.close().await { 238 errs.push(err); 239 } 240 241 flatten_errs(errs) 242 } 243 244 /// on_selected_candidate_pair_change sets a handler that is invoked when a new 245 /// ICE candidate pair is selected on_selected_candidate_pair_change(&self, f: OnSelectedCandidatePairChangeHdlrFn)246 pub fn on_selected_candidate_pair_change(&self, f: OnSelectedCandidatePairChangeHdlrFn) { 247 self.on_selected_candidate_pair_change_handler 248 .store(Some(Arc::new(Mutex::new(f)))); 249 } 250 251 /// on_connection_state_change sets a handler that is fired when the ICE 252 /// connection state changes. on_connection_state_change(&self, f: OnConnectionStateChangeHdlrFn)253 pub fn on_connection_state_change(&self, f: OnConnectionStateChangeHdlrFn) { 254 self.on_connection_state_change_handler 255 .store(Some(Arc::new(Mutex::new(f)))); 256 } 257 258 /// Role indicates the current role of the ICE transport. role(&self) -> RTCIceRole259 pub async fn role(&self) -> RTCIceRole { 260 let internal = self.internal.lock().await; 261 internal.role 262 } 263 264 /// set_remote_candidates sets the sequence of candidates associated with the remote ICETransport. set_remote_candidates(&self, remote_candidates: &[RTCIceCandidate]) -> Result<()>265 pub async fn set_remote_candidates(&self, remote_candidates: &[RTCIceCandidate]) -> Result<()> { 266 self.ensure_gatherer().await?; 267 268 if let Some(agent) = self.gatherer.get_agent().await { 269 for rc in remote_candidates { 270 let c: Arc<dyn Candidate + Send + Sync> = Arc::new(rc.to_ice()?); 271 agent.add_remote_candidate(&c)?; 272 } 273 Ok(()) 274 } else { 275 Err(Error::ErrICEAgentNotExist) 276 } 277 } 278 279 /// adds a candidate associated with the remote ICETransport. add_remote_candidate( &self, remote_candidate: Option<RTCIceCandidate>, ) -> Result<()>280 pub async fn add_remote_candidate( 281 &self, 282 remote_candidate: Option<RTCIceCandidate>, 283 ) -> Result<()> { 284 self.ensure_gatherer().await?; 285 286 if let Some(agent) = self.gatherer.get_agent().await { 287 if let Some(r) = remote_candidate { 288 let c: Arc<dyn Candidate + Send + Sync> = Arc::new(r.to_ice()?); 289 agent.add_remote_candidate(&c)?; 290 } 291 292 Ok(()) 293 } else { 294 Err(Error::ErrICEAgentNotExist) 295 } 296 } 297 298 /// State returns the current ice transport state. state(&self) -> RTCIceTransportState299 pub fn state(&self) -> RTCIceTransportState { 300 RTCIceTransportState::from(self.state.load(Ordering::SeqCst)) 301 } 302 set_state(&self, s: RTCIceTransportState)303 pub(crate) fn set_state(&self, s: RTCIceTransportState) { 304 self.state.store(s as u8, Ordering::SeqCst) 305 } 306 new_endpoint(&self, f: MatchFunc) -> Option<Arc<Endpoint>>307 pub(crate) async fn new_endpoint(&self, f: MatchFunc) -> Option<Arc<Endpoint>> { 308 let internal = self.internal.lock().await; 309 if let Some(mux) = &internal.mux { 310 Some(mux.new_endpoint(f).await) 311 } else { 312 None 313 } 314 } 315 ensure_gatherer(&self) -> Result<()>316 pub(crate) async fn ensure_gatherer(&self) -> Result<()> { 317 if self.gatherer.get_agent().await.is_none() { 318 self.gatherer.create_agent().await 319 } else { 320 Ok(()) 321 } 322 } 323 collect_stats(&self, collector: &StatsCollector)324 pub(crate) async fn collect_stats(&self, collector: &StatsCollector) { 325 if let Some(agent) = self.gatherer.get_agent().await { 326 let stats = ICETransportStats::new("ice_transport".to_string(), agent); 327 328 collector.insert("ice_transport".to_string(), Transport(stats)); 329 } 330 } 331 have_remote_credentials_change( &self, new_ufrag: &str, new_pwd: &str, ) -> bool332 pub(crate) async fn have_remote_credentials_change( 333 &self, 334 new_ufrag: &str, 335 new_pwd: &str, 336 ) -> bool { 337 if let Some(agent) = self.gatherer.get_agent().await { 338 let (ufrag, upwd) = agent.get_remote_user_credentials().await; 339 ufrag != new_ufrag || upwd != new_pwd 340 } else { 341 false 342 } 343 } 344 set_remote_credentials( &self, new_ufrag: String, new_pwd: String, ) -> Result<()>345 pub(crate) async fn set_remote_credentials( 346 &self, 347 new_ufrag: String, 348 new_pwd: String, 349 ) -> Result<()> { 350 if let Some(agent) = self.gatherer.get_agent().await { 351 Ok(agent.set_remote_credentials(new_ufrag, new_pwd).await?) 352 } else { 353 Err(Error::ErrICEAgentNotExist) 354 } 355 } 356 } 357