From c1326dab930f2224ed842750b9d3d0d48ded7e46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Th=C3=A9o=20Monnom?= Date: Sat, 8 Oct 2022 23:00:56 +0200 Subject: [PATCH] independant SignalClient, SignalStream & events SignalStream holds the connection to the websocket. It can be replaced at each connection --- Cargo.lock | 14 -- crates/livekit-core/Cargo.toml | 2 +- crates/livekit-core/src/event.rs | 43 ++++ crates/livekit-core/src/signal_client.rs | 208 ------------------ crates/livekit-core/src/signal_client/mod.rs | 71 ++++++ .../src/signal_client/signal_stream.rs | 208 ++++++++++++++++++ examples/Cargo.lock | 14 -- 7 files changed, 323 insertions(+), 237 deletions(-) create mode 100644 crates/livekit-core/src/event.rs delete mode 100644 crates/livekit-core/src/signal_client.rs create mode 100644 crates/livekit-core/src/signal_client/mod.rs create mode 100644 crates/livekit-core/src/signal_client/signal_stream.rs diff --git a/Cargo.lock b/Cargo.lock index ad4685b..b50b8ab 100644 --- a/Cargo.lock +++ b/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", diff --git a/crates/livekit-core/Cargo.toml b/crates/livekit-core/Cargo.toml index faf009c..ab79746 100644 --- a/crates/livekit-core/Cargo.toml +++ b/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" diff --git a/crates/livekit-core/src/event.rs b/crates/livekit-core/src/event.rs new file mode 100644 index 0000000..680a014 --- /dev/null +++ b/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 { + tx: mpsc::UnboundedSender, +} + +impl Emitter { + pub fn new() -> (Self, mpsc::UnboundedReceiver) { + 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 { + rx: mpsc::UnboundedReceiver, +} + +impl Events { + pub fn new(rx: mpsc::UnboundedReceiver) -> Self { + Self { rx } + } +} + +impl Stream for Events { + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + self.rx.poll_recv(cx) + } +} diff --git a/crates/livekit-core/src/signal_client.rs b/crates/livekit-core/src/signal_client.rs deleted file mode 100644 index ddd23e7..0000000 --- a/crates/livekit-core/src/signal_client.rs +++ /dev/null @@ -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 = Result; -type WebSocket = WebSocketStream>; - -#[derive(Debug)] -enum RecvMessage { - Signal { - response_chn: oneshot::Sender>, - }, -} - -#[derive(Debug)] -enum SendMessage { - Signal { - signal: signal_request::Message, - response_chn: oneshot::Sender>, - }, - Pong { - ping_data: Vec, - }, -} - -pub struct SignalClient { - read_tx: mpsc::Sender, - write_tx: mpsc::Sender, - 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 { - 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::(8); - let (write_tx, write_rx) = mpsc::channel::(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 { - 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, - mut ws_writer: SplitSink, - ) { - 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, - mut ws_reader: SplitStream, - write_tx: mpsc::Sender, - ) { - 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, - write_tx: &mpsc::Sender, - ) -> Option { - 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; - } - } - } - } -} diff --git a/crates/livekit-core/src/signal_client/mod.rs b/crates/livekit-core/src/signal_client/mod.rs new file mode 100644 index 0000000..b02e526 --- /dev/null +++ b/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; +type SignalEvents = Events; +type SignalResult = Result; + +#[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 + } +} diff --git a/crates/livekit-core/src/signal_client/signal_stream.rs b/crates/livekit-core/src/signal_client/signal_stream.rs new file mode 100644 index 0000000..5c0d7bd --- /dev/null +++ b/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>; + +#[derive(Debug)] +enum InternalMessage { + Signal { + signal: signal_request::Message, + response_chn: oneshot::Sender>, + }, + Pong { + ping_data: Vec, + }, + Close { + close_frame: Option>, + }, +} + +/// SignalStream hold the WebSocket connection +/// +/// It is replaced by [SignalClient] at each reconnection. +#[derive(Debug)] +pub(crate) struct SignalStream { + internal_tx: mpsc::Sender, + 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 { + 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::(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, + mut ws_writer: SplitSink, + 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, + mut ws_reader: SplitStream, + 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; + } +} diff --git a/examples/Cargo.lock b/examples/Cargo.lock index 1bc62fc..9a656ed 100644 --- a/examples/Cargo.lock +++ b/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",