1 #![warn(rust_2018_idioms)]
2 #![allow(dead_code)]
3
4 use anyhow::Result;
5 use hyper::service::{make_service_fn, service_fn};
6 use hyper::{Body, Method, Request, Response, Server, StatusCode};
7 use std::net::SocketAddr;
8 use std::str::FromStr;
9 use std::sync::Arc;
10 use tokio::sync::{mpsc, Mutex};
11
12 #[macro_use]
13 extern crate lazy_static;
14
15 lazy_static! {
16 static ref SDP_CHAN_TX_MUTEX: Arc<Mutex<Option<mpsc::Sender<String>>>> =
17 Arc::new(Mutex::new(None));
18 }
19
20 // HTTP Listener to get sdp
remote_handler(req: Request<Body>) -> Result<Response<Body>, hyper::Error>21 async fn remote_handler(req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
22 match (req.method(), req.uri().path()) {
23 // A HTTP handler that processes a SessionDescription given to us from the other WebRTC-rs or Pion process
24 (&Method::POST, "/sdp") => {
25 //println!("remote_handler receive from /sdp");
26 let sdp_str = match std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
27 {
28 Ok(s) => s.to_owned(),
29 Err(err) => panic!("{}", err),
30 };
31
32 {
33 let sdp_chan_tx = SDP_CHAN_TX_MUTEX.lock().await;
34 if let Some(tx) = &*sdp_chan_tx {
35 let _ = tx.send(sdp_str).await;
36 }
37 }
38
39 let mut response = Response::new(Body::empty());
40 *response.status_mut() = StatusCode::OK;
41 Ok(response)
42 }
43 // Return the 404 Not Found for other routes.
44 _ => {
45 let mut not_found = Response::default();
46 *not_found.status_mut() = StatusCode::NOT_FOUND;
47 Ok(not_found)
48 }
49 }
50 }
51
52 /// http_sdp_server starts a HTTP Server that consumes SDPs
http_sdp_server(port: u16) -> mpsc::Receiver<String>53 pub async fn http_sdp_server(port: u16) -> mpsc::Receiver<String> {
54 let (sdp_chan_tx, sdp_chan_rx) = mpsc::channel::<String>(1);
55 {
56 let mut tx = SDP_CHAN_TX_MUTEX.lock().await;
57 *tx = Some(sdp_chan_tx);
58 }
59
60 tokio::spawn(async move {
61 let addr = SocketAddr::from_str(&format!("0.0.0.0:{port}")).unwrap();
62 let service =
63 make_service_fn(|_| async { Ok::<_, hyper::Error>(service_fn(remote_handler)) });
64 let server = Server::bind(&addr).serve(service);
65 // Run this server for... forever!
66 if let Err(e) = server.await {
67 eprintln!("server error: {e}");
68 }
69 });
70
71 sdp_chan_rx
72 }
73
74 /// must_read_stdin blocks until input is received from stdin
must_read_stdin() -> Result<String>75 pub fn must_read_stdin() -> Result<String> {
76 let mut line = String::new();
77
78 std::io::stdin().read_line(&mut line)?;
79 line = line.trim().to_owned();
80 println!();
81
82 Ok(line)
83 }
84
85 // Allows compressing offer/answer to bypass terminal input limits.
86 // const COMPRESS: bool = false;
87
88 /// encode encodes the input in base64
89 /// It can optionally zip the input before encoding
encode(b: &str) -> String90 pub fn encode(b: &str) -> String {
91 //if COMPRESS {
92 // b = zip(b)
93 //}
94
95 base64::encode(b)
96 }
97
98 /// decode decodes the input from base64
99 /// It can optionally unzip the input after decoding
decode(s: &str) -> Result<String>100 pub fn decode(s: &str) -> Result<String> {
101 let b = base64::decode(s)?;
102
103 //if COMPRESS {
104 // b = unzip(b)
105 //}
106
107 let s = String::from_utf8(b)?;
108 Ok(s)
109 }
110 /*
111 func zip(in []byte) []byte {
112 var b bytes.Buffer
113 gz := gzip.NewWriter(&b)
114 _, err := gz.Write(in)
115 if err != nil {
116 panic(err)
117 }
118 err = gz.Flush()
119 if err != nil {
120 panic(err)
121 }
122 err = gz.Close()
123 if err != nil {
124 panic(err)
125 }
126 return b.Bytes()
127 }
128
129 func unzip(in []byte) []byte {
130 var b bytes.Buffer
131 _, err := b.Write(in)
132 if err != nil {
133 panic(err)
134 }
135 r, err := gzip.NewReader(&b)
136 if err != nil {
137 panic(err)
138 }
139 res, err := ioutil.ReadAll(r)
140 if err != nil {
141 panic(err)
142 }
143 return res
144 }
145 */
146