webrtc_signal/
lib.rs

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

webrtc_signal/
lib.rs
