Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 //! Owns one WebRTC peer and ordered event channel; the helper runtime bounds their lifetime. 2 //! Raw upstream errors and data messages are discarded; typed events use the existing sideband. 3 //! Candidate admission is bounded before an answer can mutate the peer or start ICE work. 4 //! Remotely opened channels are rejected; only the locally created event channel is used. 5 6 use std::collections::HashSet; 7 use std::sync::Arc; 8 use std::time::Duration; 9 10 use futures::future::BoxFuture; 11 use tokio::sync::Notify; 12 use tokio::sync::watch; 13 use tokio::task::JoinHandle; 14 use tokio::time::timeout; 15 use webrtc::data_channel::DataChannel; 16 use webrtc::data_channel::DataChannelEvent; 17 use webrtc::peer_connection::PeerConnection; 18 use webrtc::peer_connection::PeerConnectionBuilder; 19 use webrtc::peer_connection::PeerConnectionEventHandler; 20 use webrtc::peer_connection::RTCIceCandidateInit; 21 use webrtc::peer_connection::RTCIceGatheringState; 22 use webrtc::peer_connection::RTCSessionDescription; 23 24 const WAIT: Duration = Duration::from_secs(/*secs*/ 15); 25 const MAX_REMOTE_CANDIDATES: usize = 32; 26 type Result<T> = std::result::Result<T, &'static str>; 27 28 struct Events(Arc<Notify>); 29 30 // The upstream trait requires async-trait's boxed-future ABI. 31 impl PeerConnectionEventHandler for Events { 32 fn on_data_channel<'a, 'async_trait>( 33 &'a self, 34 channel: Arc<dyn DataChannel>, 35 ) -> BoxFuture<'async_trait, ()> 36 where 37 'a: 'async_trait, 38 Self: 'async_trait, 39 { 40 Box::pin(async move { 41 // Locally created channels do not invoke this callback. 42 let _ = channel.close().await; 43 }) 44 } 45 46 fn on_ice_gathering_state_change<'a, 'async_trait>( 47 &'a self, 48 state: RTCIceGatheringState, 49 ) -> BoxFuture<'async_trait, ()> 50 where 51 'a: 'async_trait, 52 Self: 'async_trait, 53 { 54 Box::pin(async move { 55 if state == RTCIceGatheringState::Complete { 56 self.0.notify_one(); 57 } 58 }) 59 } 60 } 61 62 #[derive(Debug, PartialEq)] 63 pub(crate) enum AnswerOutcome { 64 Ready, 65 TimedOut, 66 } 67 68 pub(crate) struct Transport { 69 pub(crate) incoming: crate::incoming::Incoming, 70 pub(crate) audio: crate::audio_track::AudioTrack, 71 connection: Arc<dyn PeerConnection>, 72 gathered: Arc<Notify>, 73 observer: JoinHandle<()>, 74 pub(crate) ready: watch::Receiver<bool>, 75 } 76 77 impl Transport { 78 pub(crate) async fn new() -> Result<Self> { 79 Self::with_runtime(Arc::new(crate::transport_runtime::VoiceRuntime::default())).await 80 } 81 82 async fn with_runtime(runtime: Arc<dyn webrtc::runtime::Runtime>) -> Result<Self> { 83 let (media, audio) = crate::audio_track::AudioTrack::new()?; 84 let (incoming, ingress) = crate::incoming::Incoming::new(); 85 let gathered = Arc::new(Notify::new()); 86 // Upstream defaults exhaust checks after 1.4s, including checks sent before 87 // a TCP connection exists. Keep probing throughout our negotiation deadline. 88 let check_interval = Duration::from_millis(/*millis*/ 200); 89 let attempts = u16::try_from(WAIT.as_millis() / check_interval.as_millis()) 90 .map_err(|_| "invalid voice ICE attempt budget")?; 91 let mut settings = webrtc::peer_connection::SettingEngine::default(); 92 settings.set_ice_connection_attempts(Some(check_interval), Some(attempts)); 93 let connection: Arc<dyn PeerConnection> = Arc::new( 94 PeerConnectionBuilder::new() 95 .with_interceptor_registry(rtc::interceptor::Registry::from(ingress)) 96 .with_media_engine(media) 97 .with_runtime(runtime) 98 .with_setting_engine(settings) 99 .with_handler(Arc::new(Events(gathered.clone()))) 100 .with_udp_addrs(vec!["0.0.0.0:0", "[::]:0"]) 101 .with_tcp_addrs(vec!["0.0.0.0:0", "[::]:0"]) 102 .with_data_channel_send_buffer_limit(/*bytes*/ 64 * 1024) 103 .with_sctp_receive_buffer_size(/*size*/ 64 * 1024) 104 .build() 105 .await 106 .map_err(|_| "failed to create voice peer")?, 107 ); 108 if connection.add_track(audio.track.clone()).await.is_err() { 109 let _ = timeout(Duration::from_secs(/*secs*/ 2), connection.close()).await; 110 return Err("failed to attach voice track"); 111 } 112 let channel = match connection 113 .create_data_channel("oai-events", /*options*/ None) 114 .await 115 { 116 Ok(channel) => channel, 117 Err(_) => { 118 let _ = timeout(Duration::from_secs(/*secs*/ 2), connection.close()).await; 119 return Err("failed to create voice event channel"); 120 } 121 }; 122 let (sender, ready) = watch::channel(false); 123 let observer = tokio::spawn(async move { 124 while let Some(event) = channel.poll().await { 125 match event { 126 DataChannelEvent::OnOpen => { 127 sender.send_replace(true); 128 } 129 DataChannelEvent::OnError 130 | DataChannelEvent::OnClosing 131 | DataChannelEvent::OnClose => break, 132 DataChannelEvent::OnMessage(_) 133 | DataChannelEvent::OnBufferedAmountLow 134 | DataChannelEvent::OnBufferedAmountHigh => {} 135 } 136 } 137 sender.send_replace(false); 138 }); 139 Ok(Self { 140 incoming, 141 audio, 142 connection, 143 gathered, 144 observer, 145 ready, 146 }) 147 } 148 149 pub(crate) async fn offer(&self) -> Result<String> { 150 timeout(WAIT, async { 151 let offer = self 152 .connection 153 .create_offer(/*options*/ None) 154 .await 155 .map_err(|_| "failed to create voice offer")?; 156 self.connection 157 .set_local_description(offer) 158 .await 159 .map_err(|_| "failed to set voice offer")?; 160 self.gathered.notified().await; 161 self.connection 162 .local_description() 163 .await 164 .map(|offer| offer.sdp) 165 .ok_or("missing voice offer") 166 }) 167 .await 168 .map_err(|_| "timed out gathering voice candidates")? 169 } 170 171 pub(crate) async fn apply_answer(&self, sdp: String) -> Result<AnswerOutcome> { 172 let answer = RTCSessionDescription::answer(sdp).map_err(|_| "invalid voice answer")?; 173 let parsed = answer.unmarshal().map_err(|_| "invalid voice answer")?; 174 let mut seen = HashSet::new(); 175 let mut candidates = Vec::new(); 176 // Count all occurrences, including duplicates and other components/media sections. 177 // set_remote_description itself installs ICE candidates, so validate before that call. 178 for (index, attribute) in parsed 179 .media_descriptions 180 .iter() 181 .flat_map(|media| &media.attributes) 182 .filter(|attribute| attribute.key == "candidate") 183 .enumerate() 184 { 185 if index >= MAX_REMOTE_CANDIDATES { 186 return Err("too many voice candidates"); 187 } 188 let candidate = attribute 189 .value 190 .as_deref() 191 .ok_or("invalid voice candidate")?; 192 let parsed = rtc::ice::candidate::unmarshal_candidate(candidate) 193 .map_err(|_| "invalid voice candidate")?; 194 if parsed.component() == 1 && seen.insert(candidate) { 195 candidates.push(RTCIceCandidateInit { 196 candidate: format!("candidate:{candidate}"), 197 ..Default::default() 198 }); 199 } 200 } 201 timeout(WAIT, async { 202 self.connection 203 .set_remote_description(answer) 204 .await 205 .map_err(|_| "failed to apply voice answer")?; 206 // The async driver requires explicit candidates to start TCP dialing, 207 // even when they were bundled in the non-trickle answer. 208 for candidate in candidates { 209 self.connection 210 .add_ice_candidate(candidate) 211 .await 212 .map_err(|_| "failed to apply voice candidate")?; 213 } 214 self.ready 215 .clone() 216 .wait_for(|ready| *ready) 217 .await 218 .map(|_| ()) 219 .map_err(|_| "voice event channel closed") 220 }) 221 .await 222 .map_or(Ok(AnswerOutcome::TimedOut), |result| { 223 result.map(|()| AnswerOutcome::Ready) 224 }) 225 } 226 227 pub(crate) async fn close(&mut self) -> Result<()> { 228 self.observer.abort(); 229 let _ = (&mut self.observer).await; 230 timeout(Duration::from_secs(/*secs*/ 2), self.connection.close()) 231 .await 232 .map_err(|_| "timed out closing voice peer")? 233 .map_err(|_| "failed to close voice peer") 234 } 235 } 236 237 impl Drop for Transport { 238 fn drop(&mut self) { 239 self.observer.abort(); 240 // The helper owns the runtime: destroying it cancels all remaining peer tasks. 241 } 242 } 243 244 #[cfg(test)] 245 #[path = "transport_tests.rs"] 246 mod tests;