Windows + Initial WebRTC sys & Safe WebRTC + Initial livekit-core

crates/livekit-core/src/lib.rs

@@ -0,0 +1,8 @@
+pub mod proto {
+    include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
+}
+
+mod rtc_engine;
+mod signal_client;
+
+pub mod room;

crates/livekit-core/src/room.rs

@@ -0,0 +1 @@
+pub struct Room {}

crates/livekit-core/src/rtc_engine.rs

@@ -0,0 +1,60 @@
+use crate::proto::signal_response;
+use crate::signal_client::{SignalClient, SignalClientError};
+use log::error;
+use tokio::sync::broadcast;
+
+pub struct RTCEngine {
+    signal_client: SignalClient,
+}
+
+impl RTCEngine {
+    pub fn new() -> RTCEngine {
+        Self {
+            signal_client: SignalClient::new(),
+        }
+    }
+
+    pub async fn connect(&mut self, url: &str, token: &str) -> Result<(), SignalClientError> {
+        self.signal_client.connect(url, token).await?;
+
+        tokio::spawn(Self::handle_rtc(
+            self.signal_client.response_rx.resubscribe(),
+        ));
+
+        Ok(())
+    }
+
+
+    pub fn update(&self) {}
+
+    async fn handle_rtc(mut signal_receiver: broadcast::Receiver<signal_response::Message>) {
+        loop {
+            let msg = match signal_receiver.recv().await {
+                Ok(msg) => msg,
+                Err(error) => {
+                    error!("Failed to receive SignalResponse: {:?}", error);
+                    continue;
+                }
+            };
+
+            match msg {
+                signal_response::Message::Join(join) => {}
+                signal_response::Message::Trickle(trickle) => {}
+                signal_response::Message::Answer(answer) => {}
+                signal_response::Message::Offer(offer) => {}
+                _ => {}
+            }
+        }
+    }
+}
+
+#[tokio::test]
+async fn test_test() {
+    env_logger::init();
+
+    let mut engine = RTCEngine::new();
+    engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap();
+
+    // Wait before exiting the program
+    tokio::time::sleep(core::time::Duration::from_millis(1000 * 25)).await;
+}

crates/livekit-core/src/signal_client.rs

@@ -0,0 +1,100 @@
+use futures_util::SinkExt;
+use futures_util::StreamExt;
+use log::{error, info};
+use prost::Message as ProtoMessage;
+use std::borrow::Borrow;
+use thiserror::Error;
+use tokio::net::TcpStream;
+use tokio::sync::broadcast;
+use tokio::task::JoinHandle;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
+
+use crate::{proto, proto::signal_response};
+
+#[derive(Error, Debug)]
+pub enum SignalClientError {
+    #[error("websocket failure")]
+    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error("failed to parse the url")]
+    UrlParse(#[from] url::ParseError),
+    #[error("failed to parse messages from server")]
+    ProtoParse(#[from] prost::DecodeError),
+}
+
+pub struct SignalClient {
+    ws_handle: Option<JoinHandle<Result<(), SignalClientError>>>,
+    response_tx: broadcast::Sender<signal_response::Message>,
+    pub response_rx: broadcast::Receiver<signal_response::Message>,
+}
+
+impl SignalClient {
+    pub fn new() -> Self {
+        let (tx, rx) = broadcast::channel(16);
+
+        Self {
+            response_tx: tx,
+            response_rx: rx,
+            ws_handle: None,
+        }
+    }
+
+    pub async fn connect(&mut self, url: &str, token: &str) -> Result<(), SignalClientError> {
+        let mut lk_url = url::Url::parse(url)?;
+        lk_url.set_path("/rtc");
+        lk_url
+            .query_pairs_mut()
+            .append_pair("access_token", token)
+            .append_pair("protocol", "8");
+
+        info!("Connecting to {}", lk_url);
+        let (ws, _) = connect_async(&lk_url).await?;
+
+        self.ws_handle = Some(tokio::spawn(Self::handle_ws(ws, self.response_tx.clone())));
+        Ok(())
+    }
+
+    pub async fn disconnect() {
+        unimplemented!()
+    }
+
+    async fn handle_ws(
+        mut ws: WebSocketStream<MaybeTlsStream<TcpStream>>,
+        response_tx: broadcast::Sender<signal_response::Message>,
+    ) -> Result<(), SignalClientError> {
+        loop {
+            tokio::select! {
+                next_msg = ws.next() => {
+                    let ws_msg = match next_msg {
+                        Some(msg) => msg?,
+                        None => break,
+                    };
+
+                    let data = match ws_msg {
+                        Message::Binary(data) => data,
+                        Message::Ping(data) => {
+                            ws.send(Message::Pong(data)).await?;
+                            continue
+                        },
+                        Message::Close(_frame) => break,
+                        _ => continue,
+                    };
+
+                    let proto_msg = proto::SignalResponse::decode(data.borrow())?;
+                    let signal_response = proto_msg.message.unwrap();
+
+                    match signal_response {
+                        signal_response::Message::Pong(ts) => {
+
+                        },
+                        _ => {
+                            response_tx.send(signal_response).unwrap();
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+}