1 use anyhow::Result;
2 use clap::{AppSettings, Arg, Command};
3 use hyper::service::{make_service_fn, service_fn};
4 use hyper::{Body, Client, Method, Request, Response, Server, StatusCode};
5 use std::io::Write;
6 use std::net::SocketAddr;
7 use std::str::FromStr;
8 use std::sync::Arc;
9 use tokio::sync::Mutex;
10 use tokio::time::Duration;
11 use webrtc::api::interceptor_registry::register_default_interceptors;
12 use webrtc::api::media_engine::MediaEngine;
13 use webrtc::api::APIBuilder;
14 use webrtc::data_channel::data_channel_message::DataChannelMessage;
15 use webrtc::data_channel::RTCDataChannel;
16 use webrtc::ice_transport::ice_candidate::{RTCIceCandidate, RTCIceCandidateInit};
17 use webrtc::ice_transport::ice_server::RTCIceServer;
18 use webrtc::interceptor::registry::Registry;
19 use webrtc::peer_connection::configuration::RTCConfiguration;
20 use webrtc::peer_connection::math_rand_alpha;
21 use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
22 use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
23 use webrtc::peer_connection::RTCPeerConnection;
24
25 #[macro_use]
26 extern crate lazy_static;
27
28 lazy_static! {
29 static ref PEER_CONNECTION_MUTEX: Arc<Mutex<Option<Arc<RTCPeerConnection>>>> =
30 Arc::new(Mutex::new(None));
31 static ref PENDING_CANDIDATES: Arc<Mutex<Vec<RTCIceCandidate>>> = Arc::new(Mutex::new(vec![]));
32 static ref ADDRESS: Arc<Mutex<String>> = Arc::new(Mutex::new(String::new()));
33 }
34
signal_candidate(addr: &str, c: &RTCIceCandidate) -> Result<()>35 async fn signal_candidate(addr: &str, c: &RTCIceCandidate) -> Result<()> {
36 /*println!(
37 "signal_candidate Post candidate to {}",
38 format!("http://{}/candidate", addr)
39 );*/
40 let payload = c.to_json()?.candidate;
41 let req = match Request::builder()
42 .method(Method::POST)
43 .uri(format!("http://{addr}/candidate"))
44 .header("content-type", "application/json; charset=utf-8")
45 .body(Body::from(payload))
46 {
47 Ok(req) => req,
48 Err(err) => {
49 println!("{err}");
50 return Err(err.into());
51 }
52 };
53
54 let _resp = match Client::new().request(req).await {
55 Ok(resp) => resp,
56 Err(err) => {
57 println!("{err}");
58 return Err(err.into());
59 }
60 };
61 //println!("signal_candidate Response: {}", resp.status());
62
63 Ok(())
64 }
65
66 // HTTP Listener to get ICE Credentials/Candidate from remote Peer
remote_handler(req: Request<Body>) -> Result<Response<Body>, hyper::Error>67 async fn remote_handler(req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
68 let pc = {
69 let pcm = PEER_CONNECTION_MUTEX.lock().await;
70 pcm.clone().unwrap()
71 };
72 let addr = {
73 let addr = ADDRESS.lock().await;
74 addr.clone()
75 };
76
77 match (req.method(), req.uri().path()) {
78 // A HTTP handler that allows the other WebRTC-rs or Pion instance to send us ICE candidates
79 // This allows us to add ICE candidates faster, we don't have to wait for STUN or TURN
80 // candidates which may be slower
81 (&Method::POST, "/candidate") => {
82 //println!("remote_handler receive from /candidate");
83 let candidate =
84 match std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?) {
85 Ok(s) => s.to_owned(),
86 Err(err) => panic!("{}", err),
87 };
88
89 if let Err(err) = pc
90 .add_ice_candidate(RTCIceCandidateInit {
91 candidate,
92 ..Default::default()
93 })
94 .await
95 {
96 panic!("{}", err);
97 }
98
99 let mut response = Response::new(Body::empty());
100 *response.status_mut() = StatusCode::OK;
101 Ok(response)
102 }
103
104 // A HTTP handler that processes a SessionDescription given to us from the other WebRTC-rs or Pion process
105 (&Method::POST, "/sdp") => {
106 //println!("remote_handler receive from /sdp");
107 let sdp_str = match std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
108 {
109 Ok(s) => s.to_owned(),
110 Err(err) => panic!("{}", err),
111 };
112 let sdp = match serde_json::from_str::<RTCSessionDescription>(&sdp_str) {
113 Ok(s) => s,
114 Err(err) => panic!("{}", err),
115 };
116
117 if let Err(err) = pc.set_remote_description(sdp).await {
118 panic!("{}", err);
119 }
120
121 // Create an answer to send to the other process
122 let answer = match pc.create_answer(None).await {
123 Ok(a) => a,
124 Err(err) => panic!("{}", err),
125 };
126
127 /*println!(
128 "remote_handler Post answer to {}",
129 format!("http://{}/sdp", addr)
130 );*/
131
132 // Send our answer to the HTTP server listening in the other process
133 let payload = match serde_json::to_string(&answer) {
134 Ok(p) => p,
135 Err(err) => panic!("{}", err),
136 };
137
138 let req = match Request::builder()
139 .method(Method::POST)
140 .uri(format!("http://{addr}/sdp"))
141 .header("content-type", "application/json; charset=utf-8")
142 .body(Body::from(payload))
143 {
144 Ok(req) => req,
145 Err(err) => panic!("{}", err),
146 };
147
148 let _resp = match Client::new().request(req).await {
149 Ok(resp) => resp,
150 Err(err) => {
151 println!("{err}");
152 return Err(err);
153 }
154 };
155 //println!("remote_handler Response: {}", resp.status());
156
157 // Sets the LocalDescription, and starts our UDP listeners
158 if let Err(err) = pc.set_local_description(answer).await {
159 panic!("{}", err);
160 }
161
162 {
163 let cs = PENDING_CANDIDATES.lock().await;
164 for c in &*cs {
165 if let Err(err) = signal_candidate(&addr, c).await {
166 panic!("{}", err);
167 }
168 }
169 }
170
171 let mut response = Response::new(Body::empty());
172 *response.status_mut() = StatusCode::OK;
173 Ok(response)
174 }
175 // Return the 404 Not Found for other routes.
176 _ => {
177 let mut not_found = Response::default();
178 *not_found.status_mut() = StatusCode::NOT_FOUND;
179 Ok(not_found)
180 }
181 }
182 }
183
184 #[tokio::main]
main() -> Result<()>185 async fn main() -> Result<()> {
186 let mut app = Command::new("Answer")
187 .version("0.1.0")
188 .author("Rain Liu <[email protected]>")
189 .about("An example of WebRTC-rs Answer.")
190 .setting(AppSettings::DeriveDisplayOrder)
191 .subcommand_negates_reqs(true)
192 .arg(
193 Arg::new("FULLHELP")
194 .help("Prints more detailed help information")
195 .long("fullhelp"),
196 )
197 .arg(
198 Arg::new("debug")
199 .long("debug")
200 .short('d')
201 .help("Prints debug log information"),
202 )
203 .arg(
204 Arg::new("offer-address")
205 .takes_value(true)
206 .default_value("localhost:50000")
207 .long("offer-address")
208 .help("Address that the Offer HTTP server is hosted on."),
209 )
210 .arg(
211 Arg::new("answer-address")
212 .takes_value(true)
213 .default_value("0.0.0.0:60000")
214 .long("answer-address")
215 .help("Address that the Answer HTTP server is hosted on."),
216 );
217
218 let matches = app.clone().get_matches();
219
220 if matches.is_present("FULLHELP") {
221 app.print_long_help().unwrap();
222 std::process::exit(0);
223 }
224
225 let debug = matches.is_present("debug");
226 if debug {
227 env_logger::Builder::new()
228 .format(|buf, record| {
229 writeln!(
230 buf,
231 "{}:{} [{}] {} - {}",
232 record.file().unwrap_or("unknown"),
233 record.line().unwrap_or(0),
234 record.level(),
235 chrono::Local::now().format("%H:%M:%S.%6f"),
236 record.args()
237 )
238 })
239 .filter(None, log::LevelFilter::Trace)
240 .init();
241 }
242
243 let offer_addr = matches.value_of("offer-address").unwrap().to_owned();
244 let answer_addr = matches.value_of("answer-address").unwrap().to_owned();
245
246 {
247 let mut oa = ADDRESS.lock().await;
248 *oa = offer_addr.clone();
249 }
250
251 // Prepare the configuration
252 let config = RTCConfiguration {
253 ice_servers: vec![RTCIceServer {
254 urls: vec!["stun:stun.l.google.com:19302".to_owned()],
255 ..Default::default()
256 }],
257 ..Default::default()
258 };
259
260 // Create a MediaEngine object to configure the supported codec
261 let mut m = MediaEngine::default();
262 m.register_default_codecs()?;
263
264 let mut registry = Registry::new();
265
266 // Use the default set of Interceptors
267 registry = register_default_interceptors(registry, &mut m)?;
268
269 // Create the API object with the MediaEngine
270 let api = APIBuilder::new()
271 .with_media_engine(m)
272 .with_interceptor_registry(registry)
273 .build();
274
275 // Create a new RTCPeerConnection
276 let peer_connection = Arc::new(api.new_peer_connection(config).await?);
277
278 // When an ICE candidate is available send to the other Pion instance
279 // the other Pion instance will add this candidate by calling AddICECandidate
280 let pc = Arc::downgrade(&peer_connection);
281 let pending_candidates2 = Arc::clone(&PENDING_CANDIDATES);
282 let addr2 = offer_addr.clone();
283 peer_connection.on_ice_candidate(Box::new(move |c: Option<RTCIceCandidate>| {
284 //println!("on_ice_candidate {:?}", c);
285
286 let pc2 = pc.clone();
287 let pending_candidates3 = Arc::clone(&pending_candidates2);
288 let addr3 = addr2.clone();
289 Box::pin(async move {
290 if let Some(c) = c {
291 if let Some(pc) = pc2.upgrade() {
292 let desc = pc.remote_description().await;
293 if desc.is_none() {
294 let mut cs = pending_candidates3.lock().await;
295 cs.push(c);
296 } else if let Err(err) = signal_candidate(&addr3, &c).await {
297 panic!("{}", err);
298 }
299 }
300 }
301 })
302 }));
303
304 println!("Listening on http://{answer_addr}");
305 {
306 let mut pcm = PEER_CONNECTION_MUTEX.lock().await;
307 *pcm = Some(Arc::clone(&peer_connection));
308 }
309
310 tokio::spawn(async move {
311 let addr = SocketAddr::from_str(&answer_addr).unwrap();
312 let service =
313 make_service_fn(|_| async { Ok::<_, hyper::Error>(service_fn(remote_handler)) });
314 let server = Server::bind(&addr).serve(service);
315 // Run this server for... forever!
316 if let Err(e) = server.await {
317 eprintln!("server error: {e}");
318 }
319 });
320
321 let (done_tx, mut done_rx) = tokio::sync::mpsc::channel::<()>(1);
322
323 // Set the handler for Peer connection state
324 // This will notify you when the peer has connected/disconnected
325 peer_connection.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
326 println!("Peer Connection State has changed: {s}");
327
328 if s == RTCPeerConnectionState::Failed {
329 // Wait until PeerConnection has had no network activity for 30 seconds or another failure. It may be reconnected using an ICE Restart.
330 // Use webrtc.PeerConnectionStateDisconnected if you are interested in detecting faster timeout.
331 // Note that the PeerConnection may come back from PeerConnectionStateDisconnected.
332 println!("Peer Connection has gone to failed exiting");
333 let _ = done_tx.try_send(());
334 }
335
336 Box::pin(async {})
337 }));
338
339 // Register data channel creation handling
340 peer_connection.on_data_channel(Box::new(move |d: Arc<RTCDataChannel>| {
341 let d_label = d.label().to_owned();
342 let d_id = d.id();
343 println!("New DataChannel {d_label} {d_id}");
344
345 Box::pin(async move{
346 // Register channel opening handling
347 let d2 = Arc::clone(&d);
348 let d_label2 = d_label.clone();
349 let d_id2 = d_id;
350 d.on_open(Box::new(move || {
351 println!("Data channel '{d_label2}'-'{d_id2}' open. Random messages will now be sent to any connected DataChannels every 5 seconds");
352 Box::pin(async move {
353 let mut result = Result::<usize>::Ok(0);
354 while result.is_ok() {
355 let timeout = tokio::time::sleep(Duration::from_secs(5));
356 tokio::pin!(timeout);
357
358 tokio::select! {
359 _ = timeout.as_mut() =>{
360 let message = math_rand_alpha(15);
361 println!("Sending '{message}'");
362 result = d2.send_text(message).await.map_err(Into::into);
363 }
364 };
365 }
366 })
367 }));
368
369 // Register text message handling
370 d.on_message(Box::new(move |msg: DataChannelMessage| {
371 let msg_str = String::from_utf8(msg.data.to_vec()).unwrap();
372 println!("Message from DataChannel '{d_label}': '{msg_str}'");
373 Box::pin(async{})
374 }));
375 })
376 }));
377
378 println!("Press ctrl-c to stop");
379 tokio::select! {
380 _ = done_rx.recv() => {
381 println!("received done signal!");
382 }
383 _ = tokio::signal::ctrl_c() => {
384 println!();
385 }
386 };
387
388 peer_connection.close().await?;
389
390 Ok(())
391 }
392