independant SignalClient, SignalStream & events

SignalStream holds the connection to the websocket. It can be replaced at each connection

Cargo.lock

@@ -72,9 +72,6 @@ name = "cc"
 version = "1.0.73"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2fff2a6927b3bb87f9595d67196a70493f627687a71d87a0d692242c33f58c11"
-dependencies = [
- "jobserver",
-]
 
 [[package]]
 name = "cfg-if"
@@ -449,15 +446,6 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6c8af84674fe1f223a982c933a0ee1086ac4d4052aa0fb8060c12c6ad838e754"
 
-[[package]]
-name = "jobserver"
-version = "0.1.25"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "068b1ee6743e4d11fb9c6a1e6064b3693a1b600e7f5f5988047d98b3dc9fb90b"
-dependencies = [
- "libc",
-]
-
 [[package]]
 name = "lazy_static"
 version = "1.4.0"
@@ -474,7 +462,6 @@ checksum = "329c933548736bc49fd575ee68c89e8be4d260064184389a5b77517cddd99ffb"
 name = "libwebrtc-sys"
 version = "0.1.0"
 dependencies = [
- "cc",
  "cxx",
  "cxx-build",
  "env_logger",
@@ -503,7 +490,6 @@ dependencies = [
 name = "livekit-core"
 version = "0.1.0"
 dependencies = [
- "anyhow",
  "futures",
  "futures-util",
  "lazy_static",

crates/livekit-core/Cargo.toml

@@ -2,6 +2,7 @@
 name = "livekit-core"
 version = "0.1.0"
 edition = "2021"
+license = "Apache-2.0"
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
@@ -14,7 +15,6 @@ futures-util = "0.3.23"
 thiserror = "1.0"
 prost = "0.11.0"
 prost-types = "0.11.1"
-anyhow = "1.0.63"
 livekit-webrtc = { path = "../livekit-webrtc" }
 lazy_static = "1.4.0"
 tracing = "0.1"

crates/livekit-core/src/event.rs

@@ -0,0 +1,43 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use futures::Stream;
+use tokio::sync::mpsc;
+
+/// Using unbounded channels to prevent users from blocking internal logic ( e.g: ws heartbeat )
+/// Users must listen to all events to avoid the process from running out of memory
+
+#[derive(Clone, Debug)]
+pub struct Emitter<T> {
+    tx: mpsc::UnboundedSender<T>,
+}
+
+impl<T> Emitter<T> {
+    pub fn new() -> (Self, mpsc::UnboundedReceiver<T>) {
+        let (tx, rx) = mpsc::unbounded_channel();
+
+        (Self { tx }, rx)
+    }
+
+    pub fn event(&self, event: T) {
+        let _ = self.tx.send(event);
+    }
+}
+
+#[derive(Debug)]
+pub struct Events<T> {
+    rx: mpsc::UnboundedReceiver<T>,
+}
+
+impl<T> Events<T> {
+    pub fn new(rx: mpsc::UnboundedReceiver<T>) -> Self {
+        Self { rx }
+    }
+}
+
+impl<T> Stream for Events<T> {
+    type Item = T;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        self.rx.poll_recv(cx)
+    }
+}

crates/livekit-core/src/signal_client.rs

@@ -1,208 +0,0 @@
-use std::fmt::{Debug, Formatter};
-
-use futures::future::poll_fn;
-use futures_util::{SinkExt, StreamExt};
-use futures_util::stream::{SplitSink, SplitStream};
-use prost::Message as ProstMessage;
-use thiserror::Error;
-use tokio::net::TcpStream;
-use tokio::sync::{mpsc, oneshot};
-use tokio::task::JoinHandle;
-use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
-use tokio_tungstenite::tungstenite::{Error as WsError, Message};
-use tokio_tungstenite::tungstenite::protocol::CloseFrame;
-use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
-use tracing::{event, Level};
-
-use crate::proto::{signal_request, signal_response, SignalRequest, SignalResponse};
-use crate::signal_client::SendMessage::Pong;
-
-pub const PROTOCOL_VERSION: u32 = 8;
-
-#[derive(Error, Debug)]
-pub enum SignalError {
-    #[error("websocket failure")]
-    WsError(#[from] WsError),
-    #[error("failed to parse the url")]
-    UrlParse(#[from] url::ParseError),
-    #[error("failed to decode messages from server")]
-    ProtoParse(#[from] prost::DecodeError),
-}
-
-type SignalResult<T> = Result<T, SignalError>;
-type WebSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
-
-#[derive(Debug)]
-enum RecvMessage {
-    Signal {
-        response_chn: oneshot::Sender<Option<signal_response::Message>>,
-    },
-}
-
-#[derive(Debug)]
-enum SendMessage {
-    Signal {
-        signal: signal_request::Message,
-        response_chn: oneshot::Sender<SignalResult<()>>,
-    },
-    Pong {
-        ping_data: Vec<u8>,
-    },
-}
-
-pub struct SignalClient {
-    read_tx: mpsc::Sender<RecvMessage>,
-    write_tx: mpsc::Sender<SendMessage>,
-    read_handle: JoinHandle<()>,
-    write_handle: JoinHandle<()>,
-}
-
-impl Debug for SignalClient {
-    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "SignalClient")
-    }
-}
-
-#[tracing::instrument(skip(url, token))]
-pub async fn connect(url: &str, token: &str) -> SignalResult<SignalClient> {
-    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", PROTOCOL_VERSION.to_string().as_str());
-
-    event!(Level::DEBUG, "connecting to websocket: {}", lk_url);
-    let (ws_stream, _) = connect_async(lk_url.clone()).await?;
-    event!(Level::DEBUG, "connected to SignalClient");
-
-    let (ws_writer, ws_reader) = ws_stream.split();
-
-    let (read_tx, read_rx) = mpsc::channel::<RecvMessage>(8);
-    let (write_tx, write_rx) = mpsc::channel::<SendMessage>(8);
-
-    let read_handle = tokio::spawn(SignalClient::ws_read(read_rx, ws_reader, write_tx.clone()));
-    let write_handle = tokio::spawn(SignalClient::ws_write(write_rx, ws_writer));
-
-    Ok(SignalClient {
-        read_tx,
-        write_tx,
-        read_handle,
-        write_handle,
-    })
-}
-
-impl SignalClient {
-    pub async fn close(self) {
-        drop(self.read_tx);
-        drop(self.write_tx);
-
-        let _ = self.read_handle.await;
-        let _ = self.write_handle.await;
-    }
-
-    pub async fn recv(&self) -> Option<signal_response::Message> {
-        let (send, recv) = oneshot::channel();
-        let msg = RecvMessage::Signal { response_chn: send };
-        let _ = self.read_tx.send(msg).await;
-        recv.await.expect("channel closed")
-    }
-
-    pub async fn send(&self, signal: signal_request::Message) -> SignalResult<()> {
-        let (send, recv) = oneshot::channel();
-        let msg = SendMessage::Signal {
-            signal,
-            response_chn: send,
-        };
-        let _ = self.write_tx.send(msg).await;
-        recv.await.expect("channel closed")
-    }
-
-    async fn ws_write(
-        mut write_receiver: mpsc::Receiver<SendMessage>,
-        mut ws_writer: SplitSink<WebSocket, Message>,
-    ) {
-        while let Some(msg) = write_receiver.recv().await {
-            match msg {
-                SendMessage::Signal {
-                    signal,
-                    response_chn,
-                } => {
-                    event!(Level::TRACE, "sending: {:?}", signal);
-
-                    let req = SignalRequest {
-                        message: Some(signal),
-                    };
-
-                    let write_res = ws_writer.send(Message::Binary(req.encode_to_vec())).await;
-                    if let Err(err) = write_res {
-                        event!(Level::ERROR, "failed to send signal: {:?}", err);
-                        let _ = response_chn.send(Err(err.into()));
-                        break;
-                    }
-
-                    let _ = response_chn.send(Ok(()));
-                }
-                Pong { ping_data } => {
-                    if let Err(err) = ws_writer.send(Message::Pong(ping_data)).await {
-                        event!(Level::ERROR, "failed to send pong message: {:?}", err);
-                    }
-                }
-            }
-        }
-
-        let _ = ws_writer
-            .send(Message::Close(Some(CloseFrame {
-                code: CloseCode::Normal,
-                reason: "disconnected by client".into(),
-            })))
-            .await;
-        let _ = ws_writer.flush().await;
-    }
-
-    async fn ws_read(
-        mut read_receiver: mpsc::Receiver<RecvMessage>,
-        mut ws_reader: SplitStream<WebSocket>,
-        write_tx: mpsc::Sender<SendMessage>,
-    ) {
-        while let Some(RecvMessage::Signal { mut response_chn }) = read_receiver.recv().await {
-            tokio::select! {
-                read = Self::handle_msg(&mut ws_reader, &write_tx) => {
-                    let _ = response_chn.send(read);
-                }
-                _ = poll_fn(|cx| response_chn.poll_closed(cx)) => {
-                    continue; // Cancelled
-                },
-            }
-        }
-    }
-
-    async fn handle_msg(
-        ws_reader: &mut SplitStream<WebSocket>,
-        write_tx: &mpsc::Sender<SendMessage>,
-    ) -> Option<signal_response::Message> {
-        loop {
-            let read = ws_reader.next().await?;
-            match read {
-                Ok(Message::Binary(data)) => {
-                    let res = SignalResponse::decode(data.as_slice())
-                        .expect("failed to decode SignalResponse");
-                    event!(Level::TRACE, "received: {:?}", res);
-                    return Some(res.message.unwrap());
-                }
-                Ok(Message::Ping(data)) => {
-                    let _ = write_tx.send(Pong { ping_data: data });
-                    continue;
-                }
-                Ok(Message::Close(close)) => {
-                    event!(Level::DEBUG, "server closed the connection: {:?}", close);
-                    return None;
-                }
-                _ => {
-                    event!(Level::ERROR, "unhandled websocket message {:?}", read);
-                    return None;
-                }
-            }
-        }
-    }
-}

crates/livekit-core/src/signal_client/mod.rs

@@ -0,0 +1,71 @@
+use core::num::flt2dec::Sign;
+use std::fmt::Debug;
+
+use thiserror::Error;
+use tokio_tungstenite::tungstenite::Error as WsError;
+
+use crate::event::{Emitter, Events};
+use crate::proto::{signal_request, signal_response};
+use crate::signal_client::signal_stream::SignalStream;
+
+mod signal_stream;
+
+type SignalEmitter = Emitter<SignalEvent>;
+type SignalEvents = Events<SignalEvent>;
+type SignalResult<T> = Result<T, SignalError>;
+
+#[derive(Error, Debug)]
+pub enum SignalError {
+    #[error("websocket failure")]
+    WsError(#[from] WsError),
+    #[error("failed to parse the url")]
+    UrlParse(#[from] url::ParseError),
+    #[error("failed to decode messages from server")]
+    ProtoParse(#[from] prost::DecodeError),
+}
+
+/// Events used by the RTCEngine who will handle the reconnection logic
+#[derive(Clone, Debug)]
+pub(crate) enum SignalEvent {
+    Open,
+    Signal(signal_response::Message),
+    Close,
+}
+
+#[derive(Debug)]
+pub(crate) struct SignalOptions {
+    reconnect: bool,
+    auto_subscribe: bool,
+    sid: String,
+    adaptive_stream: bool,
+}
+
+#[derive(Debug)]
+pub struct SignalClient {
+    stream: SignalStream,
+    emitter: SignalEmitter,
+}
+
+impl SignalClient {
+    pub async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> SignalResult<(Self, SignalEvents)> {
+        // TODO(theomonnom) Retry initial connection
+        let (emitter, receiver) = SignalEmitter::new();
+        let events = SignalEvents::new(receiver);
+        let stream = SignalStream::connect(url, token, options, emitter.clone()).await?;
+        Ok((Self { stream, emitter }, events))
+    }
+
+    pub async fn send(&self, signal: signal_request::Message) {
+        if let Err(_) = self.stream.send(signal).await {
+            // TODO(theomonnom) Queue message ( Ignore on full reconnect )
+        }
+    }
+
+    pub async fn reconnect(&self) {
+        // TODO(theomonnom) Close & recreate SignalStream, also send the queue if needed
+    }
+}

crates/livekit-core/src/signal_client/signal_stream.rs

@@ -0,0 +1,208 @@
+use futures_util::{SinkExt, StreamExt};
+use futures_util::stream::{SplitSink, SplitStream};
+use prost::Message as ProstMessage;
+use tokio::net::TcpStream;
+use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
+use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::protocol::CloseFrame;
+use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
+use tracing::{event, Level};
+
+use crate::proto::{signal_request, SignalRequest, SignalResponse};
+use crate::signal_client::{SignalEmitter, SignalEvent, SignalOptions, SignalResult};
+
+pub const PROTOCOL_VERSION: u32 = 8;
+
+type WebSocket = WebSocketStream<MaybeTlsStream<TcpStream>>;
+
+#[derive(Debug)]
+enum InternalMessage {
+    Signal {
+        signal: signal_request::Message,
+        response_chn: oneshot::Sender<SignalResult<()>>,
+    },
+    Pong {
+        ping_data: Vec<u8>,
+    },
+    Close {
+        close_frame: Option<CloseFrame<'static>>,
+    },
+}
+
+/// SignalStream hold the WebSocket connection
+///
+/// It is replaced by [SignalClient] at each reconnection.
+#[derive(Debug)]
+pub(crate) struct SignalStream {
+    internal_tx: mpsc::Sender<InternalMessage>,
+    read_handle: JoinHandle<()>,
+    write_handle: JoinHandle<()>,
+}
+
+impl SignalStream {
+    /// Connect to livekit websocket.
+    /// Return SignalError if the connections failed
+    ///
+    /// SignalStream will never try to reconnect if the connection has been
+    /// closed.
+    pub(super) async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+        emitter: SignalEmitter,
+    ) -> SignalResult<Self> {
+        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", PROTOCOL_VERSION.to_string().as_str())
+            .append_pair("reconnect", if options.reconnect { "1" } else { "0" })
+            .append_pair(
+                "auto_subscribe",
+                if options.auto_subscribe { "1" } else { "0" },
+            )
+            .append_pair(
+                "adaptive_stream",
+                if options.adaptive_stream { "1" } else { "0" },
+            );
+
+        event!(Level::DEBUG, "connecting to websocket: {}", lk_url);
+        let (ws_stream, _) = connect_async(lk_url).await?;
+        event!(Level::DEBUG, "connected to websocket");
+        emitter.event(SignalEvent::Open);
+
+        let (ws_writer, ws_reader) = ws_stream.split();
+        let (internal_tx, internal_rx) = mpsc::channel::<InternalMessage>(8);
+
+        let write_handle =
+            tokio::spawn(Self::handle_write(internal_rx, ws_writer, emitter.clone()));
+        let read_handle = tokio::spawn(Self::handle_read(internal_tx.clone(), ws_reader, emitter));
+
+        Ok(Self {
+            internal_tx,
+            read_handle,
+            write_handle,
+        })
+    }
+
+    /// Close the websocket
+    /// It sends a CloseFrame to the server before closing
+    pub async fn close(self) {
+        let _ = self
+            .internal_tx
+            .send(InternalMessage::Close {
+                close_frame: Some(CloseFrame {
+                    code: CloseCode::Normal,
+                    reason: "disconnected by client".into(),
+                }),
+            })
+            .await;
+
+        let _ = self.write_handle.await;
+        let _ = self.read_handle.await;
+    }
+
+    /// Send a SignalRequest to the websocket
+    /// It also waits for the message to be sent
+    pub async fn send(&self, signal: signal_request::Message) -> SignalResult<()> {
+        let (send, recv) = oneshot::channel();
+        let msg = InternalMessage::Signal {
+            signal,
+            response_chn: send,
+        };
+        let _ = self.internal_tx.send(msg).await;
+        recv.await.expect("channel closed")
+    }
+
+    /// This task is used to send messages to the websocket
+    /// It is also responsible for closing the connection
+    pub async fn handle_write(
+        mut internal_rx: mpsc::Receiver<InternalMessage>,
+        mut ws_writer: SplitSink<WebSocket, Message>,
+        emitter: SignalEmitter,
+    ) {
+        while let Some(msg) = internal_rx.recv().await {
+            match msg {
+                InternalMessage::Signal {
+                    signal,
+                    response_chn,
+                } => {
+                    event!(Level::TRACE, "sending SignalRequest: {:?}", signal);
+
+                    let data = Message::Binary(
+                        SignalRequest {
+                            message: Some(signal),
+                        }
+                            .encode_to_vec(),
+                    );
+
+                    if let Err(err) = ws_writer.send(data).await {
+                        event!(Level::ERROR, "failed to send signal: {:?}", err);
+                        let _ = response_chn.send(Err(err.into()));
+                        break;
+                    }
+
+                    let _ = response_chn.send(Ok(()));
+                }
+                InternalMessage::Pong { ping_data } => {
+                    if let Err(err) = ws_writer.send(Message::Pong(ping_data)).await {
+                        event!(Level::ERROR, "failed to send pong message: {:?}", err);
+                    }
+                }
+                InternalMessage::Close { close_frame } => {
+                    if let Some(close_frame) = close_frame {
+                        let _ = ws_writer.send(Message::Close(Some(close_frame))).await;
+                        let _ = ws_writer.flush().await;
+                    }
+                    break;
+                }
+            }
+        }
+
+        let _ = ws_writer.close().await;
+        emitter.event(SignalEvent::Close);
+    }
+
+    /// This task is used to read incoming messages from the websocket
+    /// and dispatch them through the EventEmitter.
+    ///
+    /// It can also send messages to [handle_write] task ( Used e.g. answer to pings )
+    pub async fn handle_read(
+        internal_tx: mpsc::Sender<InternalMessage>,
+        mut ws_reader: SplitStream<WebSocket>,
+        emitter: SignalEmitter,
+    ) {
+        while let Some(msg) = ws_reader.next().await {
+            match msg {
+                Ok(Message::Binary(data)) => {
+                    let res = SignalResponse::decode(data.as_slice())
+                        .expect("failed to decode SignalResponse");
+
+                    event!(Level::TRACE, "received SignalResponse: {:?}", res);
+                    emitter.event(SignalEvent::Signal(res.message.unwrap()));
+                }
+                Ok(Message::Ping(data)) => {
+                    let _ = internal_tx
+                        .send(InternalMessage::Pong { ping_data: data })
+                        .await;
+                    continue;
+                }
+                Ok(Message::Close(close)) => {
+                    event!(Level::DEBUG, "server closed the connection: {:?}", close);
+                    break;
+                }
+                _ => {
+                    event!(Level::ERROR, "unhandled websocket message {:?}", msg);
+                    break;
+                }
+            }
+        }
+
+        let _ = internal_tx
+            .send(InternalMessage::Close { close_frame: None })
+            .await;
+    }
+}

examples/Cargo.lock

@@ -70,9 +70,6 @@ name = "cc"
 version = "1.0.73"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2fff2a6927b3bb87f9595d67196a70493f627687a71d87a0d692242c33f58c11"
-dependencies = [
- "jobserver",
-]
 
 [[package]]
 name = "cfg-if"
@@ -428,15 +425,6 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6c8af84674fe1f223a982c933a0ee1086ac4d4052aa0fb8060c12c6ad838e754"
 
-[[package]]
-name = "jobserver"
-version = "0.1.25"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "068b1ee6743e4d11fb9c6a1e6064b3693a1b600e7f5f5988047d98b3dc9fb90b"
-dependencies = [
- "libc",
-]
-
 [[package]]
 name = "lazy_static"
 version = "1.4.0"
@@ -453,7 +441,6 @@ checksum = "329c933548736bc49fd575ee68c89e8be4d260064184389a5b77517cddd99ffb"
 name = "libwebrtc-sys"
 version = "0.1.0"
 dependencies = [
- "cc",
  "cxx",
  "cxx-build",
  "glob",
@@ -481,7 +468,6 @@ dependencies = [
 name = "livekit-core"
 version = "0.1.0"
 dependencies = [
- "anyhow",
  "futures",
  "futures-util",
  "lazy_static",