1 use anyhow::Result;
2 use clap::{AppSettings, Arg, Command};
3 use std::collections::HashMap;
4 use std::io::Write;
5 use std::sync::Arc;
6 use tokio::time::Duration;
7 use webrtc::api::interceptor_registry::register_default_interceptors;
8 use webrtc::api::media_engine::{MediaEngine, MIME_TYPE_VP8};
9 use webrtc::api::APIBuilder;
10 use webrtc::ice_transport::ice_server::RTCIceServer;
11 use webrtc::interceptor::registry::Registry;
12 use webrtc::peer_connection::configuration::RTCConfiguration;
13 use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
14 use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
15 use webrtc::rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication;
16 use webrtc::rtp_transceiver::rtp_codec::{
17 RTCRtpCodecCapability, RTCRtpHeaderExtensionCapability, RTPCodecType,
18 };
19 use webrtc::track::track_local::track_local_static_rtp::TrackLocalStaticRTP;
20 use webrtc::track::track_local::{TrackLocal, TrackLocalWriter};
21 use webrtc::Error;
22
23 #[tokio::main]
main() -> Result<()>24 async fn main() -> Result<()> {
25 let mut app = Command::new("simulcast")
26 .version("0.1.0")
27 .author("Rain Liu <[email protected]>")
28 .about("An example of simulcast.")
29 .setting(AppSettings::DeriveDisplayOrder)
30 .subcommand_negates_reqs(true)
31 .arg(
32 Arg::new("FULLHELP")
33 .help("Prints more detailed help information")
34 .long("fullhelp"),
35 )
36 .arg(
37 Arg::new("debug")
38 .long("debug")
39 .short('d')
40 .help("Prints debug log information"),
41 );
42
43 let matches = app.clone().get_matches();
44
45 if matches.is_present("FULLHELP") {
46 app.print_long_help().unwrap();
47 std::process::exit(0);
48 }
49
50 let debug = matches.is_present("debug");
51 if debug {
52 env_logger::Builder::new()
53 .format(|buf, record| {
54 writeln!(
55 buf,
56 "{}:{} [{}] {} - {}",
57 record.file().unwrap_or("unknown"),
58 record.line().unwrap_or(0),
59 record.level(),
60 chrono::Local::now().format("%H:%M:%S.%6f"),
61 record.args()
62 )
63 })
64 .filter(None, log::LevelFilter::Trace)
65 .init();
66 }
67
68 // Everything below is the WebRTC-rs API! Thanks for using it ❤️.
69
70 // Create a MediaEngine object to configure the supported codec
71 let mut m = MediaEngine::default();
72
73 m.register_default_codecs()?;
74
75 // Enable Extension Headers needed for Simulcast
76 for extension in [
77 "urn:ietf:params:rtp-hdrext:sdes:mid",
78 "urn:ietf:params:rtp-hdrext:sdes:rtp-stream-id",
79 "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id",
80 ] {
81 m.register_header_extension(
82 RTCRtpHeaderExtensionCapability {
83 uri: extension.to_owned(),
84 },
85 RTPCodecType::Video,
86 None,
87 )?;
88 }
89 // Create a InterceptorRegistry. This is the user configurable RTP/RTCP Pipeline.
90 // This provides NACKs, RTCP Reports and other features. If you use `webrtc.NewPeerConnection`
91 // this is enabled by default. If you are manually managing You MUST create a InterceptorRegistry
92 // for each PeerConnection.
93 let mut registry = Registry::new();
94
95 // Use the default set of Interceptors
96 registry = register_default_interceptors(registry, &mut m)?;
97
98 // Create the API object with the MediaEngine
99 let api = APIBuilder::new()
100 .with_media_engine(m)
101 .with_interceptor_registry(registry)
102 .build();
103
104 // Prepare the configuration
105 let config = RTCConfiguration {
106 ice_servers: vec![RTCIceServer {
107 urls: vec!["stun:stun.l.google.com:19302".to_owned()],
108 ..Default::default()
109 }],
110 ..Default::default()
111 };
112
113 // Create a new RTCPeerConnection
114 let peer_connection = Arc::new(api.new_peer_connection(config).await?);
115
116 // Create Track that we send video back to browser on
117 let mut output_tracks = HashMap::new();
118 for s in ["q", "h", "f"] {
119 let output_track = Arc::new(TrackLocalStaticRTP::new(
120 RTCRtpCodecCapability {
121 mime_type: MIME_TYPE_VP8.to_owned(),
122 ..Default::default()
123 },
124 format!("video_{s}"),
125 format!("webrtc-rs_{s}"),
126 ));
127
128 // Add this newly created track to the PeerConnection
129 let rtp_sender = peer_connection
130 .add_track(Arc::clone(&output_track) as Arc<dyn TrackLocal + Send + Sync>)
131 .await?;
132
133 // Read incoming RTCP packets
134 // Before these packets are returned they are processed by interceptors. For things
135 // like NACK this needs to be called.
136 tokio::spawn(async move {
137 let mut rtcp_buf = vec![0u8; 1500];
138 while let Ok((_, _)) = rtp_sender.read(&mut rtcp_buf).await {}
139 Result::<()>::Ok(())
140 });
141
142 output_tracks.insert(s.to_owned(), output_track);
143 }
144
145 // Wait for the offer to be pasted
146 let line = signal::must_read_stdin()?;
147 let desc_data = signal::decode(line.as_str())?;
148 let offer = serde_json::from_str::<RTCSessionDescription>(&desc_data)?;
149
150 // Set the remote SessionDescription
151 peer_connection.set_remote_description(offer).await?;
152
153 // Set a handler for when a new remote track starts
154 let pc = Arc::downgrade(&peer_connection);
155 peer_connection.on_track(Box::new(move |track, _, _| {
156 println!("Track has started");
157
158 let rid = track.rid().to_owned();
159 let output_track = if let Some(output_track) = output_tracks.get(&rid) {
160 Arc::clone(output_track)
161 } else {
162 println!("output_track not found for rid = {rid}");
163 return Box::pin(async {});
164 };
165
166 // Start reading from all the streams and sending them to the related output track
167 let media_ssrc = track.ssrc();
168 let pc2 = pc.clone();
169 tokio::spawn(async move {
170 let mut result = Result::<usize>::Ok(0);
171 while result.is_ok() {
172 println!("Sending pli for stream with rid: {rid}, ssrc: {media_ssrc}");
173
174 let timeout = tokio::time::sleep(Duration::from_secs(3));
175 tokio::pin!(timeout);
176
177 tokio::select! {
178 _ = timeout.as_mut() =>{
179 if let Some(pc) = pc2.upgrade(){
180 result = pc.write_rtcp(&[Box::new(PictureLossIndication{
181 sender_ssrc: 0,
182 media_ssrc,
183 })]).await.map_err(Into::into);
184 }else{
185 break;
186 }
187 }
188 };
189 }
190 });
191
192 tokio::spawn(async move {
193 // Read RTP packets being sent to webrtc-rs
194 println!("enter track loop {}", track.rid());
195 while let Ok((rtp, _)) = track.read_rtp().await {
196 if let Err(err) = output_track.write_rtp(&rtp).await {
197 if Error::ErrClosedPipe != err {
198 println!("output track write_rtp got error: {err} and break");
199 break;
200 } else {
201 println!("output track write_rtp got error: {err}");
202 }
203 }
204 }
205 println!("exit track loop {}", track.rid());
206 });
207
208 Box::pin(async {})
209 }));
210
211 let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<()>(1);
212
213 // Set the handler for Peer connection state
214 // This will notify you when the peer has connected/disconnected
215 peer_connection.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
216 println!("Peer Connection State has changed: {s}");
217
218 if s == RTCPeerConnectionState::Failed {
219 // Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
220 // Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
221 // Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
222 println!("Peer Connection has gone to failed exiting");
223 let _ = done_tx.try_send(());
224 }
225
226 Box::pin(async {})
227 }));
228
229 // Create an answer
230 let answer = peer_connection.create_answer(None).await?;
231
232 // Create channel that is blocked until ICE Gathering is complete
233 let mut gather_complete = peer_connection.gathering_complete_promise().await;
234
235 // Sets the LocalDescription, and starts our UDP listeners
236 peer_connection.set_local_description(answer).await?;
237
238 // Block until ICE Gathering is complete, disabling trickle ICE
239 // we do this because we only can exchange one signaling message
240 // in a production application you should exchange ICE Candidates via OnICECandidate
241 let _ = gather_complete.recv().await;
242
243 // Output the answer in base64 so we can paste it in browser
244 if let Some(local_desc) = peer_connection.local_description().await {
245 let json_str = serde_json::to_string(&local_desc)?;
246 let b64 = signal::encode(&json_str);
247 println!("{b64}");
248 } else {
249 println!("generate local_description failed!");
250 }
251
252 println!("Press ctrl-c to stop");
253 tokio::select! {
254 _ = done_rx.recv() => {
255 println!("received done signal!");
256 }
257 _ = tokio::signal::ctrl_c() => {
258 println!();
259 }
260 };
261
262 peer_connection.close().await?;
263
264 Ok(())
265 }
266