refactored RTCEngine

Added comments & separated code into multiple parts.
Maybe the next step is to put PeerConnections into another file 🤔

crates/livekit-core/src/lib.rs

@@ -7,5 +7,6 @@ mod signal_client;
 mod pc_transport;
 mod rtc_engine;
 mod local_participant;
+mod event;
 
-pub mod room;
+pub mod room;

crates/livekit-core/src/lk_runtime.rs

@@ -6,7 +6,7 @@ use livekit_webrtc::peer_connection_factory::PeerConnectionFactory;
 use livekit_webrtc::webrtc::RTCRuntime;
 
 /// SAFETY: The order of initialization and deletion is important for LKRuntime.
-/// See the C++ constructors & destructor of these fields
+/// See the C++ constructors & destructors of these fields
 
 pub struct LKRuntime {
     pub pc_factory: PeerConnectionFactory,

crates/livekit-core/src/local_participant.rs

@@ -1,6 +1,4 @@
 use std::sync::Arc;
-
-use futures_util::TryFutureExt;
 use tokio::sync::Mutex;
 
 use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};

crates/livekit-core/src/pc_transport.rs

@@ -134,4 +134,3 @@ impl PCTransport {
         Ok(())
     }
 }
-

crates/livekit-core/src/room.rs

@@ -2,7 +2,6 @@ use std::sync::Arc;
 
 use thiserror::Error;
 use tokio::sync::Mutex;
-use tracing::{event, Level, trace};
 
 use crate::local_participant::LocalParticipant;
 use crate::rtc_engine;
@@ -14,6 +13,11 @@ pub enum RoomError {
     Engine(#[from] EngineError),
 }
 
+#[derive(Debug)]
+pub enum RoomEvent { 
+    
+}
+
 pub struct Room {
     sid: String,
     name: String,
@@ -24,18 +28,16 @@ pub struct Room {
 #[tracing::instrument(skip(url, token))]
 pub async fn connect(url: &str, token: &str) -> Result<Room, RoomError> {
     let engine = rtc_engine::connect(url, token).await?;
-
-    engine.on_data(Box::new(|packet| {
-        event!(Level::DEBUG, "received data");
-        Box::pin(async move {})
-    })).await;
-
     let join = engine.join_response().await;
     let engine = Arc::new(Mutex::new(engine));
     let local_participant = LocalParticipant::from(join.participant.unwrap(), engine.clone());
-
     let internal = Arc::new(RoomInternal::new(engine));
     let room_info = join.room.unwrap();
+
+    tokio::spawn(async move {
+
+    });
+
     Ok(Room {
         sid: room_info.sid,
         name: room_info.name,
@@ -45,7 +47,11 @@ pub async fn connect(url: &str, token: &str) -> Result<Room, RoomError> {
 }
 
 impl Room {
-    pub fn local_participant(&mut self) -> &mut LocalParticipant {
+    pub fn local_participant(&self) -> &LocalParticipant {
+        &self.local_participant
+    }
+
+    pub fn local_participant_mut(&mut self) -> &mut LocalParticipant {
         &mut self.local_participant
     }
 

crates/livekit-core/src/rtc_engine/engine_internal.rs

@@ -1,47 +1,38 @@
 use std::fmt::{Debug, Formatter};
-use std::future::Future;
-use std::pin::Pin;
-use std::sync::{Arc, Weak};
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
+use std::sync::Arc;
 use std::time::Duration;
 
-use lazy_static::lazy_static;
 use prost::Message;
 use serde::{Deserialize, Serialize};
-use thiserror::Error;
 use tokio::sync::{mpsc, Mutex};
 use tokio::time;
 use tracing::{event, Level};
 
-use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState};
+use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataState};
-use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription};
+use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
 use livekit_webrtc::peer_connection::{
     IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
 };
 use livekit_webrtc::peer_connection_factory::{
     ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
 };
-use livekit_webrtc::rtc_error::RTCError;
 
-use crate::{proto, signal_client};
 use crate::lk_runtime::LKRuntime;
 use crate::pc_transport::PCTransport;
-use crate::proto::{
+use crate::proto;
-    data_packet, DataPacket, JoinResponse, signal_request, signal_response, SignalTarget,
-    TrickleRequest, UserPacket,
-};
 use crate::proto::data_packet::Value;
-use crate::signal_client::{SignalClient, SignalError};
+use crate::proto::{
+    data_packet, signal_request, signal_response, DataPacket, JoinResponse, SignalTarget,
+    TrickleRequest,
+};
+use crate::rtc_engine::{EngineError, MAX_ICE_CONNECT_TIMEOUT};
+use crate::signal_client::SignalClient;
 
 const LOSSY_DC_LABEL: &str = "_lossy";
 const RELIABLE_DC_LABEL: &str = "_reliable";
-const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
-
-lazy_static! {
-    // Share one LKRuntime across all RTCEngine instances
-    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
-}
 
+// Used to communicate IceCandidate with the server
 #[derive(Serialize, Deserialize)]
 #[allow(non_snake_case)]
 struct IceCandidateJSON {
@@ -50,36 +41,8 @@ struct IceCandidateJSON {
     candidate: String,
 }
 
-pub struct Packet {
-    pub data: UserPacket,
-    pub kind: data_packet::Kind,
-}
-
-pub type OnDataHandler =
-Box<dyn (FnMut(Packet) -> Pin<Box<dyn Future<Output=()> + Send + 'static>>) + Send + Sync>;
-
-#[derive(Error, Debug)]
-pub enum EngineError {
-    #[error("signal failure")]
-    Signal(#[from] SignalError),
-    #[error("internal webrtc failure")]
-    Rtc(#[from] RTCError),
-    #[error("failed to parse sdp")]
-    Parse(#[from] SdpParseError),
-    #[error("serde error")]
-    Serde(#[from] serde_json::Error),
-    #[error("failed to send data to the datachannel")]
-    Data(#[from] DataSendError),
-    #[error("connection error: {0}")]
-    Connection(String),
-    #[error("decode error")]
-    Decode(#[from] prost::DecodeError),
-    #[error("internal error: {0}")]
-    Internal(String), // Unexpected error
-}
-
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
-enum PCState {
+pub(crate) enum PCState {
     New,
     Connected,
     Disconnected,
@@ -88,7 +51,7 @@ enum PCState {
 }
 
 #[derive(Debug)]
-pub enum EngineMessage {
+pub(crate) enum InternalMessage {
     IceCandidate {
         ice_candidate: IceCandidate,
         publisher: bool,
@@ -109,113 +72,37 @@ pub enum EngineMessage {
     },
 }
 
+pub(crate) struct EngineInternal {
+    pub(super) publisher_pc: Arc<Mutex<PCTransport>>,
+    pub(super) subscriber_pc: Arc<Mutex<PCTransport>>,
+    pub(super) lossy_dc: Arc<Mutex<DataChannel>>,
+    pub(super) reliable_dc: Arc<Mutex<DataChannel>>,
+    pub(super) lossy_dc_sub: Arc<Mutex<Option<DataChannel>>>,
+    pub(super) reliable_dc_sub: Arc<Mutex<Option<DataChannel>>>,
 
-#[derive(Debug)]
+    pub(super) msg_sender: mpsc::Sender<InternalMessage>,
-pub struct RTCEngine {
+    pub(super) join_response: Mutex<JoinResponse>,
-    signal_client: Arc<SignalClient>,
+    pub(super) pc_state: AtomicU8, // casted to PCState
-    internal: Arc<EngineInternal>,
+    pub(super) has_published: AtomicBool,
-
-    #[allow(unused)]
-    lk_runtime: Arc<LKRuntime>, // Keep a reference while we're using the RTCEngine
 }
 
-#[tracing::instrument(skip(url, token))]
+impl Debug for EngineInternal {
-pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-    // Acquire an existing/a new LKRuntime
+        write!(f, "EngineInternal")
-    let mut lk_runtime_ref = LK_RUNTIME.lock().await;
-    let mut lk_runtime = lk_runtime_ref.upgrade();
-
-    if lk_runtime.is_none() {
-        let new_runtime = Arc::new(LKRuntime::new());
-        *lk_runtime_ref = Arc::downgrade(&new_runtime);
-        lk_runtime = Some(new_runtime);
     }
-    let lk_runtime = lk_runtime.unwrap();
-    let signal_client = Arc::new(signal_client::connect(url, token).await?);
-
-    if let Some(signal_response::Message::Join(join_response)) = signal_client.recv().await {
-        event!(Level::DEBUG, "received JoinResponse: {:?}", join_response);
-        let (sender, receiver) = mpsc::channel(8);
-        let internal = Arc::new(EngineInternal::new(
-            lk_runtime.clone(),
-            sender,
-            join_response.clone(),
-        )?);
-
-        if !join_response.subscriber_primary {
-            internal.publisher_pc.lock().await.negotiate().await?;
-        }
-
-        tokio::spawn({
-            let signal_client = signal_client.clone();
-            let internal = internal.clone();
-
-            async move {
-                internal.run(receiver, signal_client).await;
-            }
-        });
-
-        Ok(RTCEngine {
-            lk_runtime,
-            signal_client,
-            internal,
-        })
-    } else {
-        panic!("the first received message isn't a JoinResponse");
-    }
-}
-
-impl RTCEngine {
-    /// Send data to other participants in the Room
-    #[tracing::instrument]
-    pub async fn publish_data(
-        &mut self,
-        data: &DataPacket,
-        kind: data_packet::Kind,
-    ) -> Result<(), EngineError> {
-        self.internal.ensure_publisher_connected(kind).await?;
-        self.internal.data_channel(kind)
-            .lock()
-            .await
-            .send(&data.encode_to_vec(), true)
-            .map_err(Into::into)
-    }
-
-    /// Return the last received JoinResponse
-    pub async fn join_response(&self) -> JoinResponse {
-        self.internal.join_response.lock().await.clone()
-    }
-
-    pub async fn on_data(&self, f: OnDataHandler) {
-        *self.internal.on_data_handler.lock().await = Some(f);
-    }
-}
-
-struct EngineInternal {
-    publisher_pc: Arc<Mutex<PCTransport>>,
-    subscriber_pc: Arc<Mutex<PCTransport>>,
-    lossy_dc: Arc<Mutex<DataChannel>>,
-    reliable_dc: Arc<Mutex<DataChannel>>,
-    lossy_dc_sub: Arc<Mutex<Option<DataChannel>>>,
-    reliable_dc_sub: Arc<Mutex<Option<DataChannel>>>,
-
-    msg_sender: mpsc::Sender<EngineMessage>,
-    join_response: Mutex<JoinResponse>,
-    pc_state: AtomicU8,
-    // PCState
-    has_published: AtomicBool,
-
-    // Listeners
-    on_data_handler: Arc<Mutex<Option<OnDataHandler>>>,
 }
 
 impl EngineInternal {
-    /// New internal is created on connect & on reconnect
+    /// Configure the PeerConnections
-    /// It creates the PeerConnections, the DataChannels and the libwebrtc listeners
+    ///
+    /// This is called on connect & on full reconnect.
+    /// Create the PeerConnections & the DataChannels.
+    /// Register listeners and send the internal messages
+    /// to the event_loop.
     #[tracing::instrument]
-    fn new(
+    pub(super) fn configure(
         lk_runtime: Arc<LKRuntime>,
-        sender: mpsc::Sender<EngineMessage>,
+        sender: mpsc::Sender<InternalMessage>,
         join: JoinResponse,
     ) -> Result<Self, EngineError> {
         let rtc_config = RTCConfiguration {
@@ -245,7 +132,7 @@ impl EngineInternal {
         publisher_pc.peer_connection().on_ice_candidate(Box::new({
             let sender = sender.clone();
             move |ice_candidate| {
-                let _ = sender.blocking_send(EngineMessage::IceCandidate {
+                let _ = sender.blocking_send(InternalMessage::IceCandidate {
                     ice_candidate,
                     publisher: true,
                 });
@@ -255,7 +142,7 @@ impl EngineInternal {
         subscriber_pc.peer_connection().on_ice_candidate(Box::new({
             let sender = sender.clone();
             move |ice_candidate| {
-                let _ = sender.blocking_send(EngineMessage::IceCandidate {
+                let _ = sender.blocking_send(InternalMessage::IceCandidate {
                     ice_candidate,
                     publisher: false,
                 });
@@ -268,7 +155,7 @@ impl EngineInternal {
                 let sender = sender.clone();
 
                 tokio::spawn(async move {
-                    let _ = sender.send(EngineMessage::PublisherOffer { offer }).await;
+                    let _ = sender.send(InternalMessage::PublisherOffer { offer }).await;
                 });
 
                 Box::pin(async move {})
@@ -285,7 +172,7 @@ impl EngineInternal {
                 let sender = sender.clone();
                 move |data_channel| {
                     let _ =
-                        sender.blocking_send(EngineMessage::PrimaryDataChannel { data_channel });
+                        sender.blocking_send(InternalMessage::PrimaryDataChannel { data_channel });
                 }
             }));
         }
@@ -293,7 +180,7 @@ impl EngineInternal {
         primary_pc.peer_connection().on_connection_change(Box::new({
             let sender = sender.clone();
             move |state| {
-                let _ = sender.blocking_send(EngineMessage::ConnectionChange {
+                let _ = sender.blocking_send(InternalMessage::ConnectionChange {
                     state,
                     primary: true,
                 });
@@ -305,15 +192,13 @@ impl EngineInternal {
             .on_connection_change(Box::new({
                 let sender = sender.clone();
                 move |state| {
-                    let _ = sender.blocking_send(EngineMessage::ConnectionChange {
+                    let _ = sender.blocking_send(InternalMessage::ConnectionChange {
                         state,
                         primary: false,
                     });
                 }
             }));
 
-        // Note that when subscriber_primary feature is enabled,
-        // the subscriber uses his own data channels created by the server.
         let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(
             LOSSY_DC_LABEL,
             DataChannelInit {
@@ -345,26 +230,26 @@ impl EngineInternal {
             join_response: Mutex::new(join),
             pc_state: AtomicU8::new(PCState::New as u8),
             has_published: AtomicBool::new(false),
-            on_data_handler: Default::default(),
         })
     }
 
-    /// Map the libwebrtc listeners to a mpsc channel
+    /// Send InternalMessage when a datachannel receives data
     #[tracing::instrument]
-    fn configure_dc(
+    fn configure_dc(data_channel: &mut DataChannel, sender: mpsc::Sender<InternalMessage>) {
-        data_channel: &mut DataChannel,
-        sender: mpsc::Sender<EngineMessage>,
-    ) {
         data_channel.on_message(Box::new(move |data, binary| {
-            let _ = sender.blocking_send(EngineMessage::Data {
+            let _ = sender.blocking_send(InternalMessage::Data {
                 data: data.to_vec(),
                 binary,
             });
         }));
     }
 
+    /// Ensure the publisher PeerConnection is connected
+    ///
+    /// When subscriber_primary is enabled, only the subscriber PeerConnection is negotiated.
+    /// This allows for faster connection when we don't need the publisher
     #[tracing::instrument]
-    async fn ensure_publisher_connected(
+    pub(super) async fn ensure_publisher_connected(
         self: &Arc<Self>,
         kind: data_packet::Kind,
     ) -> Result<(), EngineError> {
@@ -377,7 +262,7 @@ impl EngineInternal {
             let mut publisher = publisher.lock().await;
             if !publisher.is_connected()
                 && publisher.peer_connection().ice_connection_state()
-                != IceConnectionState::IceConnectionChecking
+                    != IceConnectionState::IceConnectionChecking
             {
                 tokio::spawn({
                     let internal = self.clone();
@@ -406,7 +291,7 @@ impl EngineInternal {
                 interval.tick().await;
             }
         })
-            .await;
+        .await;
 
         if res.is_err() {
             let err =
@@ -418,88 +303,11 @@ impl EngineInternal {
         }
     }
 
+    /// Run the event_loop of the RTCEngine
     #[tracing::instrument]
-    async fn handle_signal(
+    pub(super) async fn run(
         self: &Arc<Self>,
-        signal: signal_response::Message,
+        mut receiver: mpsc::Receiver<InternalMessage>,
-        signal_client: Arc<SignalClient>,
-    ) -> Result<(), EngineError> {
-        match signal {
-            signal_response::Message::Answer(answer) => {
-                event!(Level::TRACE, "received answer for publisher: {:?}", answer);
-                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
-                self.publisher_pc
-                    .lock()
-                    .await
-                    .set_remote_description(sdp)
-                    .await?;
-            }
-            signal_response::Message::Offer(offer) => {
-                event!(Level::TRACE, "received offer for subscriber: {:?}", offer);
-                let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
-
-                self.subscriber_pc
-                    .lock()
-                    .await
-                    .set_remote_description(sdp)
-                    .await?;
-                let answer = self.subscriber_pc
-                    .lock()
-                    .await
-                    .peer_connection()
-                    .create_answer(RTCOfferAnswerOptions::default())
-                    .await?;
-                self.subscriber_pc
-                    .lock()
-                    .await
-                    .peer_connection()
-                    .set_local_description(answer.clone())
-                    .await?;
-
-                tokio::spawn(async move {
-                    let _ = signal_client.send(signal_request::Message::Answer(
-                        proto::SessionDescription {
-                            r#type: "answer".to_string(),
-                            sdp: answer.to_string(),
-                        },
-                    )).await;
-                });
-            }
-            signal_response::Message::Trickle(trickle) => {
-                let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?;
-                let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
-
-                event!(
-                    Level::TRACE,
-                    "received ice_candidate ({:?}) - {:?}",
-                    SignalTarget::from_i32(trickle.target).unwrap(),
-                    ice
-                );
-
-                if trickle.target == SignalTarget::Publisher as i32 {
-                    self.publisher_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice)
-                        .await?;
-                } else {
-                    self.subscriber_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice)
-                        .await?;
-                }
-            }
-            _ => {}
-        }
-
-        Ok(())
-    }
-
-    #[tracing::instrument]
-    pub async fn run(
-        self: &Arc<Self>,
-        mut receiver: mpsc::Receiver<EngineMessage>,
         signal_client: Arc<SignalClient>,
     ) {
         loop {
@@ -533,17 +341,116 @@ impl EngineInternal {
         }
     }
 
+    /// Handle SignalResponse messages coming from the server
+    ///
+    /// Run the needed livekit-protocol
+    #[tracing::instrument]
+    async fn handle_signal(
+        self: &Arc<Self>,
+        signal: signal_response::Message,
+        signal_client: Arc<SignalClient>,
+    ) -> Result<(), EngineError> {
+        match signal {
+            signal_response::Message::Answer(answer) => {
+                event!(Level::TRACE, "received answer for publisher: {:?}", answer);
+                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
+                self.publisher_pc
+                    .lock()
+                    .await
+                    .set_remote_description(sdp)
+                    .await?;
+            }
+            signal_response::Message::Offer(offer) => {
+                // Handle the subscriber offer & send an answer to livekit-server
+                // We always get an offer from the server when connecting
+                event!(Level::TRACE, "received offer for subscriber: {:?}", offer);
+                let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
+
+                self.subscriber_pc
+                    .lock()
+                    .await
+                    .set_remote_description(sdp)
+                    .await?;
+                let answer = self
+                    .subscriber_pc
+                    .lock()
+                    .await
+                    .peer_connection()
+                    .create_answer(RTCOfferAnswerOptions::default())
+                    .await?;
+                self.subscriber_pc
+                    .lock()
+                    .await
+                    .peer_connection()
+                    .set_local_description(answer.clone())
+                    .await?;
+
+                tokio::spawn(async move {
+                    let _ = signal_client
+                        .send(signal_request::Message::Answer(proto::SessionDescription {
+                            r#type: "answer".to_string(),
+                            sdp: answer.to_string(),
+                        }))
+                        .await;
+                });
+            }
+            signal_response::Message::Trickle(trickle) => {
+                // Add the IceCandidate received from the livekit-server
+                let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?;
+                let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
+
+                event!(
+                    Level::TRACE,
+                    "received ice_candidate ({:?}) - {:?}",
+                    SignalTarget::from_i32(trickle.target).unwrap(),
+                    ice
+                );
+
+                if trickle.target == SignalTarget::Publisher as i32 {
+                    self.publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
+                } else {
+                    self.subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
+                }
+            }
+            _ => {}
+        }
+
+        Ok(())
+    }
+
+    /// Handle libwebrtc messages
+    ///
+    /// Every message used inside this function comes from libwebrtc.
+    /// The messages are received in [EngineInternal](#run)
+    /// We're not handling the messages inside the signaling_thread, to return
+    /// as quickly as possible.
     #[tracing::instrument]
     async fn handle_message(
         self: &Arc<Self>,
-        msg: EngineMessage,
+        msg: InternalMessage,
         signal_client: Arc<SignalClient>,
     ) -> Result<(), EngineError> {
         match msg {
-            EngineMessage::IceCandidate {
+            InternalMessage::IceCandidate {
                 ice_candidate,
                 publisher,
             } => {
+                // Send the IceCandidate to livekit-server
+                // Note that ContinualGatheringPolicy is set to GatherContinually
+                let json = serde_json::to_string(&IceCandidateJSON {
+                    sdpMid: ice_candidate.sdp_mid(),
+                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                    candidate: ice_candidate.candidate(),
+                })?;
+
                 let target = if publisher {
                     SignalTarget::Publisher
                 } else {
@@ -557,23 +464,19 @@ impl EngineInternal {
                     ice_candidate
                 );
 
-                let json = serde_json::to_string(&IceCandidateJSON {
-                    sdpMid: ice_candidate.sdp_mid(),
-                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
-                    candidate: ice_candidate.candidate(),
-                })?;
-
-                // Send the ice_candidate to the server
                 tokio::spawn(async move {
-                    let _ = signal_client.send(signal_request::Message::Trickle(
+                    let _ = signal_client
-                        TrickleRequest {
+                        .send(signal_request::Message::Trickle(TrickleRequest {
                             candidate_init: json,
                             target: target as i32,
-                        },
+                        }))
-                    )).await;
+                        .await;
                 });
             }
-            EngineMessage::ConnectionChange { state, primary } => {
+            InternalMessage::ConnectionChange { state, primary } => {
+                // PeerConnectionState changed
+                // Reconnect if we've been disconnected unexpectedly
+                // If connected for the first time, send OnConnect event
                 if primary && state == PeerConnectionState::Connected {
                     let old_state = self.pc_state.load(Ordering::SeqCst);
                     self.pc_state
@@ -583,18 +486,21 @@ impl EngineInternal {
                         // TODO(theomonnom) OnConnected
                     }
                 } else if state == PeerConnectionState::Failed {
-                    self.pc_state.store(PCState::Disconnected as u8, Ordering::SeqCst);
+                    self.pc_state
+                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
 
                     // TODO(theomonnom) handle Disconnect
                 }
             }
-            EngineMessage::PrimaryDataChannel { mut data_channel } => {
+            InternalMessage::PrimaryDataChannel { mut data_channel } => {
+                // Received datachannel from the primary PeerConnection.
+                // If subscriber_primary is enabled, the datachannel is used for downstream data
                 let reliable = data_channel.label() == RELIABLE_DC_LABEL;
                 Self::configure_dc(&mut data_channel, self.msg_sender.clone());
 
                 event!(
                     Level::TRACE,
-                    "received subscriber data_channel - {:?}",
+                    "received primary data_channel - {:?}",
                     data_channel
                 );
 
@@ -604,44 +510,40 @@ impl EngineInternal {
                     *self.lossy_dc_sub.lock().await = Some(data_channel);
                 }
             }
-            EngineMessage::PublisherOffer { offer } => {
+            InternalMessage::PublisherOffer { offer } => {
-                event!(
+                // Send the publisher offer to livekit-server
-                    Level::TRACE,
+                event!(Level::TRACE, "sending publisher offer - {:?}", offer);
-                    "sending publisher offer - {:?}",
-                    offer
-                );
 
-                // Send the offer to the server
                 tokio::spawn(async move {
-                    let _ = signal_client.send(signal_request::Message::Offer(
+                    let _ = signal_client
-                        proto::SessionDescription {
+                        .send(signal_request::Message::Offer(proto::SessionDescription {
                             r#type: "offer".to_string(),
                             sdp: offer.to_string(),
-                        },
+                        }))
-                    )).await;
+                        .await;
                 });
             }
-            EngineMessage::Data {
+            InternalMessage::Data { data, binary } => {
-                data,
+                // Received data from a datachannel
-                binary,
+                // If this is a Speaker DataPacket, update the active speakers
-            } => {
+                // Send SpeakersChanged/OnData event
                 if !binary {
                     return Err(EngineError::Internal(
-                        "text messages aren't supported by LiveKit".to_string(),
+                        "text messages aren't supported".to_string(),
                     ));
                 }
 
                 let data = DataPacket::decode(&*data)?;
                 match data.value.unwrap() {
                     Value::User(user) => {
-                        let mut handler = self.on_data_handler.lock().await;
+                        /*let mut handler = self.on_data_handler.lock().await;
                         if let Some(f) = &mut *handler {
                             f(Packet {
                                 data: user,
                                 kind: data_packet::Kind::from_i32(data.kind).unwrap(),
                             })
                                 .await;
-                        }
+                        }*/
                     }
                     Value::Speaker(_) => {
                         // TODO(theomonnonm)
@@ -657,18 +559,14 @@ impl EngineInternal {
     async fn negotiate_publisher(self: &Arc<Self>) -> Result<(), EngineError> {
         self.has_published.store(true, Ordering::SeqCst);
         if let Err(err) = self.publisher_pc.lock().await.negotiate().await {
-            event!(
+            event!(Level::ERROR, "failed to negotiate the publisher: {:?}", err,);
-                Level::ERROR,
-                "failed to negotiate the publisher: {:?}",
-                err,
-            );
             Err(EngineError::Rtc(err))
         } else {
             Ok(())
         }
     }
 
-    fn data_channel(&self, kind: data_packet::Kind) -> Arc<Mutex<DataChannel>> {
+    pub(super) fn data_channel(&self, kind: data_packet::Kind) -> Arc<Mutex<DataChannel>> {
         if kind == data_packet::Kind::Reliable {
             self.reliable_dc.clone()
         } else {
@@ -676,9 +574,3 @@ impl EngineInternal {
         }
     }
 }
-
-impl Debug for EngineInternal {
-    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(f, "EngineInternal")
-    }
-}

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

@@ -0,0 +1,153 @@
+use crate::lk_runtime::LKRuntime;
+use crate::proto::{data_packet, signal_response, DataPacket, JoinResponse, UserPacket};
+use crate::rtc_engine::engine_internal::EngineInternal;
+use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalOptions};
+use futures_util::{FutureExt, StreamExt};
+use lazy_static::lazy_static;
+use livekit_webrtc::data_channel::DataSendError;
+use livekit_webrtc::jsep::SdpParseError;
+use livekit_webrtc::rtc_error::RTCError;
+use prost::Message;
+use std::sync::{Arc, Weak};
+use std::time::Duration;
+use thiserror::Error;
+use tokio::sync::{mpsc, Mutex};
+use tokio::time;
+use tracing::{event, Level};
+
+mod engine_internal;
+
+lazy_static! {
+    // Share one LKRuntime across all RTCEngine instances
+    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
+}
+
+pub(crate) const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+pub(crate) const JOIN_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Error, Debug)]
+pub enum EngineError {
+    #[error("signal failure")]
+    Signal(#[from] SignalError),
+    #[error("internal webrtc failure")]
+    Rtc(#[from] RTCError),
+    #[error("failed to parse sdp")]
+    Parse(#[from] SdpParseError),
+    #[error("serde error")]
+    Serde(#[from] serde_json::Error),
+    #[error("failed to send data to the datachannel")]
+    Data(#[from] DataSendError),
+    #[error("connection error: {0}")]
+    Connection(String),
+    #[error("decode error")]
+    Decode(#[from] prost::DecodeError),
+    #[error("internal error: {0}")]
+    Internal(String), // Unexpected error
+}
+
+#[derive(Debug)]
+pub struct Packet {
+    pub data: UserPacket,
+    pub kind: data_packet::Kind,
+}
+
+#[derive(Debug)]
+pub enum EngineEvent {
+    DataReceived(Packet),
+}
+
+#[derive(Debug)]
+pub struct RTCEngine {
+    signal_client: Arc<SignalClient>,
+    internal: Arc<EngineInternal>,
+
+    #[allow(unused)]
+    lk_runtime: Arc<LKRuntime>, // Keep a reference while we're using the RTCEngine
+}
+
+#[tracing::instrument(skip(url, token))]
+pub async fn connect(
+    url: &str,
+    token: &str,
+    options: SignalOptions,
+) -> Result<RTCEngine, EngineError> {
+    // Acquire an existing/a new LKRuntime
+    let mut lk_runtime_ref = LK_RUNTIME.lock().await;
+    let mut lk_runtime = lk_runtime_ref.upgrade();
+
+    if lk_runtime.is_none() {
+        let new_runtime = Arc::new(LKRuntime::new());
+        *lk_runtime_ref = Arc::downgrade(&new_runtime);
+        lk_runtime = Some(new_runtime);
+    }
+    let lk_runtime = lk_runtime.unwrap();
+    let (signal_client, mut signal_events) = SignalClient::connect(url, token, options).await?;
+    let signal_client = Arc::new(signal_client);
+
+    let join_response = time::timeout(JOIN_RESPONSE_TIMEOUT, async move {
+        while let Some(event) = signal_events.next().await {
+            match event {
+                SignalEvent::Signal(signal_response::Message::Join(join)) => return join,
+                _ => {
+                    // Should we try a reconnect on close here?
+                    continue;
+                }
+            }
+        }
+
+        unreachable!();
+    })
+    .await
+    .map_err(|_| EngineError::Internal("failed to receive JoinResponse".to_string()))?;
+
+    event!(Level::DEBUG, "received JoinResponse: {:?}", join_response);
+
+    let (sender, receiver) = mpsc::channel(8);
+    let internal = Arc::new(EngineInternal::configure(
+        lk_runtime.clone(),
+        sender,
+        join_response.clone(),
+    )?);
+
+    if !join_response.subscriber_primary {
+        internal.publisher_pc.lock().await.negotiate().await?;
+    }
+
+    tokio::spawn({
+        let signal_client = signal_client.clone();
+        let internal = internal.clone();
+
+        async move {
+            internal.run(receiver, signal_client).await;
+        }
+    });
+
+    Ok(RTCEngine {
+        lk_runtime,
+        signal_client,
+        internal,
+    })
+}
+
+impl RTCEngine {
+    /// Send data to other participants in the Room
+    #[tracing::instrument]
+    pub async fn publish_data(
+        &mut self,
+        data: &DataPacket,
+        kind: data_packet::Kind,
+    ) -> Result<(), EngineError> {
+        self.internal.ensure_publisher_connected(kind).await?;
+        self.internal
+            .data_channel(kind)
+            .lock()
+            .await
+            .send(&data.encode_to_vec(), true)
+            .map_err(Into::into)
+    }
+
+    /// Return the last received JoinResponse
+    pub async fn join_response(&self) -> JoinResponse {
+        self.internal.join_response.lock().await.clone()
+    }
+}