webrtc.rs (11234B)
1 use std::{fs::File, sync::Arc}; 2 3 use matrix_sdk::{ruma::events::room::message::RoomMessageEventContent, Room}; 4 use tokio::sync::{Mutex, Notify}; 5 use tracing::{error, info}; 6 use webrtc::{ 7 api::{ 8 interceptor_registry::register_default_interceptors, 9 media_engine::{MediaEngine, MIME_TYPE_OPUS}, 10 APIBuilder, 11 }, 12 ice_transport::{ 13 ice_candidate::RTCIceCandidate, ice_connection_state::RTCIceConnectionState, 14 ice_server::RTCIceServer, 15 }, 16 interceptor::registry::Registry, 17 media::io::ogg_writer::OggWriter, 18 peer_connection::{configuration::RTCConfiguration, RTCPeerConnection}, 19 rtp_transceiver::rtp_codec::{RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType}, 20 track::{ 21 track_local::{track_local_static_rtp::TrackLocalStaticRTP, TrackLocal}, 22 track_remote::TrackRemote, 23 }, 24 }; 25 26 lazy_static::lazy_static! { 27 static ref PENDING_CANDIDATES: Arc<Mutex<Vec<RTCIceCandidate>>> = Arc::new(Mutex::new(vec![])); 28 } 29 30 async fn save_to_disk( 31 writer: Arc<Mutex<dyn webrtc::media::io::Writer + Send + Sync>>, 32 track: Arc<TrackRemote>, 33 notify: Arc<Notify>, 34 ) -> color_eyre::Result<()> { 35 loop { 36 tokio::select! { 37 result = track.read_rtp() => { 38 if let Ok((rtp_packet, _)) = result { 39 let mut w = writer.lock().await; 40 w.write_rtp(&rtp_packet)?; 41 }else{ 42 info!("file closing begin after read_rtp error"); 43 let mut w = writer.lock().await; 44 if let Err(err) = w.close() { 45 error!("file close err: {err}"); 46 } 47 info!("file closing end after read_rtp error"); 48 return Ok(()); 49 } 50 } 51 _ = notify.notified() => { 52 info!("file closing begin after notified"); 53 let mut w = writer.lock().await; 54 if let Err(err) = w.close() { 55 error!("file close err: {err}"); 56 } 57 info!("file closing end after notified"); 58 return Ok(()); 59 } 60 } 61 } 62 } 63 64 // FIXME: This is for now just a reference 65 pub async fn start_webrtc_call_to_sip(room: Room) -> color_eyre::Result<Arc<RTCPeerConnection>> { 66 let ogg_writer: Arc<Mutex<dyn webrtc::media::io::Writer + Send + Sync>> = Arc::new(Mutex::new( 67 OggWriter::new(File::create("./test.opus")?, 48000, 2)?, 68 )); 69 // Create a MediaEngine object to configure the supported codec 70 let mut m = MediaEngine::default(); 71 72 // We only add SIP save codecs here 73 m.register_codec( 74 RTCRtpCodecParameters { 75 capability: RTCRtpCodecCapability { 76 mime_type: MIME_TYPE_OPUS.to_owned(), 77 clock_rate: 48000, 78 channels: 2, 79 sdp_fmtp_line: "minptime=10;useinbandfec=1".to_owned(), 80 rtcp_feedback: vec![], 81 }, 82 payload_type: 111, 83 ..Default::default() 84 }, 85 RTPCodecType::Audio, 86 )?; 87 88 // m.register_codec( 89 // RTCRtpCodecParameters { 90 // capability: RTCRtpCodecCapability { 91 // mime_type: MIME_TYPE_PCMU.to_owned(), 92 // clock_rate: 8000, 93 // channels: 0, 94 // sdp_fmtp_line: "".to_owned(), 95 // rtcp_feedback: vec![], 96 // }, 97 // payload_type: 0, 98 // ..Default::default() 99 // }, 100 // RTPCodecType::Audio, 101 // )?; 102 103 // Create a InterceptorRegistry. This is the user configurable RTP/RTCP Pipeline. 104 // This provides NACKs, RTCP Reports and other features. If you use `webrtc.NewPeerConnection` 105 // this is enabled by default. If you are manually managing You MUST create a InterceptorRegistry 106 // for each PeerConnection. 107 let mut registry = Registry::new(); 108 109 // Use the default set of Interceptors 110 registry = register_default_interceptors(registry, &mut m)?; 111 112 // Create the API object with the MediaEngine 113 let api = APIBuilder::new() 114 .with_media_engine(m) 115 .with_interceptor_registry(registry) 116 .build(); 117 118 // Prepare the configuration 119 let config = RTCConfiguration { 120 ice_servers: vec![RTCIceServer { 121 urls: vec!["stun:stun.l.google.com:19302".to_owned()], 122 ..Default::default() 123 }], 124 ..Default::default() 125 }; 126 127 // Create a new RTCPeerConnection 128 let peer_connection = Arc::new(api.new_peer_connection(config).await?); 129 130 // Create audio 2 way track 131 let audio_track = Arc::new(TrackLocalStaticRTP::new( 132 RTCRtpCodecCapability { 133 mime_type: MIME_TYPE_OPUS.to_owned(), 134 clock_rate: 48000, 135 channels: 2, 136 sdp_fmtp_line: "".to_owned(), 137 rtcp_feedback: vec![], 138 }, 139 "audio".to_owned(), 140 "webrtc-rs".to_owned(), 141 )); 142 143 // TODO: We probably want this later when we start to negotiate so we can easily get the sender reference around 144 let _sender = peer_connection 145 .add_track(Arc::clone(&audio_track) as Arc<dyn TrackLocal + Send + Sync>) 146 .await?; 147 148 // // This must exist. 149 // peer_connection 150 // .add_transceiver_from_kind( 151 // RTPCodecType::Audio, 152 // Some(RTCRtpTransceiverInit { 153 // direction: RTCRtpTransceiverDirection::Recvonly, 154 // send_encodings: vec![], 155 // }), 156 // ) 157 // .await?; 158 159 // Prepare udp conns 160 // Also update incoming packets with expected PayloadType, the browser may use 161 // a different value. We have to modify so our stream matches what rtp-forwarder.sdp expects 162 // FIXME: This should handle the incoming stream. Which I think should be matrix 163 // let mut udp_conns = HashMap::new(); 164 // udp_conns.insert( 165 // "audio".to_owned(), 166 // UdpConn { 167 // conn: { 168 // let sock = UdpSocket::bind("127.0.0.1:0").await?; 169 // sock.connect(format!("127.0.0.1:{}", 4000)).await?; 170 // Arc::new(sock) 171 // }, 172 // payload_type: 111, 173 // }, 174 // ); 175 176 // Set a handler for when a new remote track starts, this handler will forward data to 177 // our UDP listeners. 178 // In your application this is where you would handle/process audio/video 179 //let pc = Arc::downgrade(&peer_connection); 180 let notify_tx = Arc::new(Notify::new()); 181 let notify_rx = notify_tx.clone(); 182 peer_connection.on_track(Box::new(move |track, _, _| { 183 // Retrieve udp connection 184 // TODO: Use this to get the matrix connection? 185 // let c = if let Some(c) = udp_conns.get(&track.kind().to_string()) { 186 // c.clone() 187 // } else { 188 // return Box::pin(async {}); 189 // }; 190 191 // Send a PLI on an interval so that the publisher is pushing a keyframe every rtcpPLIInterval 192 //let media_ssrc = track.ssrc(); 193 //let pc2 = pc.clone(); 194 // tokio::spawn(async move { 195 // let mut result = color_eyre::Result::<usize>::Ok(0); 196 // while result.is_ok() { 197 // let timeout = tokio::time::sleep(Duration::from_secs(3)); 198 // tokio::pin!(timeout); 199 200 // tokio::select! { 201 // _ = timeout.as_mut() =>{ 202 // if let Some(pc) = pc2.upgrade(){ 203 // result = pc.write_rtcp(&[Box::new(PictureLossIndication{ 204 // sender_ssrc: 0, 205 // media_ssrc, 206 // })]).await.map_err(Into::into); 207 // }else{ 208 // break; 209 // } 210 // } 211 // }; 212 // } 213 // }); 214 215 // tokio::spawn(async move { 216 // let mut b = vec![0u8; 1500]; 217 // while let Ok((mut rtp_packet, _)) = track.read(&mut b).await { 218 // // Update the PayloadType 219 // rtp_packet.header.payload_type = c.payload_type; 220 221 // // Marshal into original buffer with updated PayloadType 222 223 // let n = rtp_packet.marshal_to(&mut b)?; 224 225 // // Write 226 // if let Err(err) = c.conn.send(&b[..n]).await { 227 // // For this particular example, third party applications usually timeout after a short 228 // // amount of time during which the user doesn't have enough time to provide the answer 229 // // to the browser. 230 // // That's why, for this particular example, the user first needs to provide the answer 231 // // to the browser then open the third party application. Therefore we must not kill 232 // // the forward on "connection refused" errors 233 // //if opError, ok := err.(*net.OpError); ok && opError.Err.Error() == "write: connection refused" { 234 // // continue 235 // //} 236 // //panic(err) 237 // if err.to_string().contains("Connection refused") { 238 // continue; 239 // } else { 240 // error!("conn send err: {err}"); 241 // break; 242 // } 243 // } 244 // } 245 246 // color_eyre::Result::<()>::Ok(()) 247 // }); 248 249 //TODO: Remove when going live 250 let notify_rx2 = Arc::clone(¬ify_rx); 251 let ogg_writer2 = Arc::clone(&ogg_writer); 252 Box::pin(async move { 253 let codec = track.codec(); 254 let mime_type = codec.capability.mime_type.to_lowercase(); 255 if mime_type == MIME_TYPE_OPUS.to_lowercase() { 256 info!("Got Opus track, saving to disk as output.opus (48 kHz, 2 channels)"); 257 tokio::spawn(async move { 258 let _ = save_to_disk(ogg_writer2, track, notify_rx2).await; 259 }); 260 } 261 }) 262 263 //Box::pin(async {}) 264 })); 265 266 // Set the handler for ICE connection state 267 // This will notify you when the peer has connected/disconnected 268 peer_connection.on_ice_connection_state_change(Box::new( 269 move |connection_state: RTCIceConnectionState| { 270 info!("Connection State has changed {connection_state}"); 271 if connection_state == RTCIceConnectionState::Failed { 272 notify_tx.notify_waiters(); 273 info!("PeerConnection failed, closing"); 274 let error_message = 275 RoomMessageEventContent::notice_plain(format!("🚨Call ended🚨",)); 276 let room_clone = room.clone(); 277 return Box::pin(async move { 278 room_clone 279 .clone() 280 .send(error_message, None) 281 .await 282 .expect("Failed to send error message"); 283 }); 284 } 285 Box::pin(async {}) 286 }, 287 )); 288 289 Ok(peer_connection) 290 }