wip

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

@@ -0,0 +1,637 @@
+use parking_lot::{Mutex, RwLock};
+use std::error;
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
+use std::sync::{Arc, Weak};
+use std::time::Duration;
+
+use tokio::sync::{mpsc, Mutex as AsyncMutex};
+
+use lazy_static::lazy_static;
+use prost::Message;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::time::sleep;
+use tracing::{debug, error, info, trace, warn};
+
+use crate::{proto, signal_client};
+use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState};
+use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription};
+use livekit_webrtc::media_stream::MediaStream;
+use livekit_webrtc::peer_connection::{
+    IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
+};
+use livekit_webrtc::peer_connection_factory::RTCConfiguration;
+use livekit_webrtc::rtc_error::RTCError;
+use livekit_webrtc::rtp_receiver::RtpReceiver;
+
+use crate::proto::data_packet::Value;
+use crate::proto::{
+    data_packet, signal_request, signal_response, DataPacket, JoinResponse, ParticipantUpdate,
+    SignalTarget, TrickleRequest,
+};
+use crate::rtc_engine::lk_runtime::LKRuntime;
+use crate::rtc_engine::pc_transport::PCTransport;
+use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents};
+use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions};
+use std::cell::RefCell;
+
+use super::{rtc_events, EngineEvents};
+use super::{EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult};
+//
+// TODO(theomonnom): Smarter retry intervals
+pub(crate) const RECONNECT_ATTEMPTS: u32 = 10;
+pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300);
+
+pub(crate) const LOSSY_DC_LABEL: &str = "_lossy";
+pub(crate) const RELIABLE_DC_LABEL: &str = "_reliable";
+pub(crate) 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());
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum PCState {
+    New,
+    Connected,
+    Disconnected,
+    Reconnecting,
+    Closed,
+}
+
+#[derive(Debug)]
+pub struct SessionInfo {
+    url: String,
+    token: String,
+    options: SignalOptions,
+}
+
+#[derive(Debug)]
+pub struct EngineInternal {
+    lk_runtime: Arc<LKRuntime>,
+    signal_client: Arc<SignalClient>,
+    session: Arc<RwLock<Option<RTCSession>>>,
+    reconnecting: AtomicBool,
+    closed: AtomicBool,
+    engine_emitter: EngineEmitter,
+}
+
+/// This struct holds a WebRTC session
+/// The session changes at every reconnection
+#[derive(Debug)]
+pub struct RTCSession {
+    join_response: Mutex<JoinResponse>,
+    has_published: AtomicBool,
+    pc_state: AtomicU8, // Casted to PCState enum
+
+    publisher_pc: AsyncMutex<PCTransport>,
+    subscriber_pc: AsyncMutex<PCTransport>,
+
+    // Publisher data channels
+    // Used to send data to other participants ( The SFU forwards the messages )
+    lossy_dc: Mutex<DataChannel>,
+    reliable_dc: Mutex<DataChannel>,
+
+    // Subscriber data channels
+    // These fields are never used, we just keep a strong reference to them,
+    // so we can receive data from other participants
+    sub_reliable_dc: Mutex<Option<DataChannel>>,
+    sub_lossy_dc: Mutex<Option<DataChannel>>,
+}
+
+#[derive(Serialize, Deserialize)]
+#[allow(non_snake_case)]
+struct IceCandidateJSON {
+    sdpMid: String,
+    sdpMLineIndex: i32,
+    candidate: String,
+}
+
+impl RTCSession {
+    pub fn configure(
+        lk_runtime: Arc<LKRuntime>,
+        join_response: JoinResponse,
+    ) -> EngineResult<(Self, RTCEvents)> {
+        let (rtc_emitter, events) = mpsc::unbounded_channel();
+        let rtc_config = RTCConfiguration::from(join_response.clone());
+
+        let mut publisher_pc = PCTransport::new(
+            lk_runtime
+                .pc_factory
+                .create_peer_connection(rtc_config.clone())?,
+            SignalTarget::Publisher,
+        );
+
+        let mut subscriber_pc = PCTransport::new(
+            lk_runtime
+                .pc_factory
+                .create_peer_connection(rtc_config.clone())?,
+            SignalTarget::Subscriber,
+        );
+
+        let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(
+            LOSSY_DC_LABEL,
+            DataChannelInit {
+                ordered: true,
+                max_retransmits: Some(0),
+                ..DataChannelInit::default()
+            },
+        )?;
+
+        let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(
+            RELIABLE_DC_LABEL,
+            DataChannelInit {
+                ordered: true,
+                ..DataChannelInit::default()
+            },
+        )?;
+
+        rtc_events::forward_pc_events(&mut publisher_pc, rtc_emitter.clone());
+        rtc_events::forward_pc_events(&mut subscriber_pc, rtc_emitter.clone());
+        rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone());
+        rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter.clone());
+
+        Ok((
+            Self {
+                join_response: Mutex::new(join_response),
+                has_published: AtomicBool::new(false),
+                pc_state: AtomicU8::new(PCState::New as u8),
+                publisher_pc: AsyncMutex::new(publisher_pc),
+                subscriber_pc: AsyncMutex::new(subscriber_pc),
+                lossy_dc: Mutex::new(lossy_dc),
+                reliable_dc: Mutex::new(reliable_dc),
+                sub_lossy_dc: Mutex::new(None),
+                sub_reliable_dc: Mutex::new(None),
+            },
+            events,
+        ))
+    }
+
+    async fn negotiate_publisher(&self) -> EngineResult<()> {
+        self.has_published.store(true, Ordering::SeqCst);
+        let res = self.publisher_pc.lock().await.negotiate().await;
+        if let Err(err) = &res {
+            error!("failed to negotiate the publisher: {:?}", err);
+        }
+        res.map_err(Into::into)
+    }
+
+    async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> {
+        if !self.join_response.lock().subscriber_primary {
+            return Ok(());
+        }
+
+        let publisher = &self.publisher_pc;
+        {
+            let mut publisher = publisher.lock().await;
+            if !publisher.is_connected()
+                && publisher.peer_connection().ice_connection_state()
+                    != IceConnectionState::IceConnectionChecking
+            {
+                let _ = self.negotiate_publisher().await;
+            }
+        }
+
+        let dc = self.data_channel(kind);
+        if dc.lock().state() == DataState::Open {
+            return Ok(());
+        }
+
+        // Wait until the PeerConnection is connected
+        let wait_connected = async move {
+            while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open {
+                sleep(Duration::from_millis(50)).await;
+            }
+        };
+
+        tokio::select! {
+            _ = wait_connected => Ok(()),
+            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
+                let err = EngineError::Connection("could not establish publisher connection: timeout".to_string());
+                error!(error = ?err);
+                Err(err)
+            }
+        }
+    }
+
+    async fn wait_pc_connection(&self) -> EngineResult<()> {
+        let wait_connected = async move {
+            while self.pc_state.load(Ordering::SeqCst) != PCState::Connected as u8 {
+                sleep(Duration::from_millis(50)).await;
+            }
+        };
+
+        tokio::select! {
+            _ = wait_connected => Ok(()),
+            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
+                let err = EngineError::Connection("wait_pc_connection timed out".to_string());
+                Err(err)
+            }
+        }
+    }
+
+    fn data_channel(&self, kind: data_packet::Kind) -> &Mutex<DataChannel> {
+        if kind == data_packet::Kind::Reliable {
+            &self.reliable_dc
+        } else {
+            &self.lossy_dc
+        }
+    }
+}
+
+impl Default for EngineInternal {
+    fn default() -> Self {
+        let mut lk_runtime = None;
+        {
+            let mut lk_runtime_ref = LK_RUNTIME.lock();
+            lk_runtime = lk_runtime_ref.upgrade();
+
+            if lk_runtime.is_none() {
+                let new_runtime = Arc::new(LKRuntime::default());
+                *lk_runtime_ref = Arc::downgrade(&new_runtime);
+                lk_runtime = Some(new_runtime);
+            }
+        }
+
+        Self {
+            lk_runtime: lk_runtime.unwrap(),
+            ..Default::default()
+        }
+    }
+}
+
+impl EngineInternal {
+    #[tracing::instrument]
+    pub async fn connect(
+        self: Arc<Self>,
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> EngineResult<EngineEvents> {
+        let mut signal_events = self.signal_client.connect(url, token, options).await?;
+        let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
+        debug!("received JoinResponse: {:?}", join_response);
+
+        let (session, rtc_events) =
+            RTCSession::configure(self.lk_runtime.clone(), join_response.clone())?;
+        let session = Arc::new(RwLock::new(Some(session)));
+
+        let (engine_emitter, engine_events) = mpsc::channel(8);
+
+        tokio::spawn(self.clone().signal_task(signal_events));
+        tokio::spawn(self.clone().engine_task(rtc_events));
+
+        if !join_response.subscriber_primary {
+            session.read().unwrap().negotiate_publisher().await?;
+        }
+
+        Ok(engine_events)
+    }
+
+    async fn engine_task(self: Arc<Self>, mut rtc_events: RTCEvents) {
+        while let Some(event) = rtc_events.recv().await {
+            if let Err(err) = self.handle_rtc(event).await {
+                error!("failed to handle rtc event: {:?}", err);
+            }
+        }
+    }
+
+    async fn signal_task(self: Arc<Self>, mut signal_events: SignalEvents) {
+        while let Some(signal) = signal_events.recv().await {
+            match signal {
+                SignalEvent::Open => {}
+                SignalEvent::Signal(signal) => {
+                    if let Err(err) = self.handle_signal(signal).await {
+                        error!("failed to handle signal: {:?}", err);
+                    }
+                }
+                SignalEvent::Close => {
+                    self.handle_disconnected();
+                }
+            }
+        }
+    }
+
+    async fn handle_rtc(self: &Arc<Self>, event: RTCEvent) -> EngineResult<()> {
+        match event {
+            RTCEvent::IceCandidate {
+                ice_candidate,
+                target,
+            } => {
+                let json = serde_json::to_string(&IceCandidateJSON {
+                    sdpMid: ice_candidate.sdp_mid(),
+                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                    candidate: ice_candidate.candidate(),
+                })?;
+
+                trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate);
+
+                tokio::spawn(async move {
+                    signal_client
+                        .send(signal_request::Message::Trickle(TrickleRequest {
+                            candidate_init: json,
+                            target: target as i32,
+                        }))
+                        .await;
+                });
+            }
+            RTCEvent::ConnectionChange { state, target } => {
+                trace!("connection change, {:?} {:?}", state, target);
+                let subscriber_primary = session
+                    .read()
+                    .unwrap()
+                    .join_response
+                    .lock()
+                    .subscriber_primary;
+
+                let is_primary = subscriber_primary && target == SignalTarget::Subscriber;
+
+                if is_primary && state == PeerConnectionState::Connected {
+                    let old_state = session
+                        .read()
+                        .unwrap()
+                        .pc_state
+                        .swap(PCState::Connected as u8, Ordering::SeqCst);
+                    if old_state == PCState::New as u8 {
+                        let _ = emitter.send(EngineEvent::Connected).await; // First time connected
+                    }
+                } else if state == PeerConnectionState::Failed {
+                    session
+                        .read()
+                        .unwrap()
+                        .pc_state
+                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
+
+                    Self::handle_disconnected(signal_client, engine_inner, emitter);
+                }
+            }
+            RTCEvent::DataChannel {
+                data_channel,
+                target,
+            } => {
+                if target == SignalTarget::Subscriber {
+                    if data_channel.label() == RELIABLE_DC_LABEL {
+                        *session.read().unwrap().sub_reliable_dc.lock() = Some(data_channel);
+                    } else {
+                        *session.read().unwrap().sub_lossy_dc.lock() = Some(data_channel);
+                    }
+                }
+            }
+            RTCEvent::Offer { offer, target } => {
+                if target == SignalTarget::Publisher {
+                    // Send the publisher offer to the server
+                    tokio::spawn(async move {
+                        signal_client
+                            .send(signal_request::Message::Offer(proto::SessionDescription {
+                                r#type: "offer".to_string(),
+                                sdp: offer.to_string(),
+                            }))
+                            .await;
+                    });
+                }
+            }
+            RTCEvent::AddTrack {
+                rtp_receiver,
+                streams,
+                target,
+            } => {
+                if target == SignalTarget::Subscriber {
+                    let _ = emitter
+                        .send(EngineEvent::AddTrack {
+                            rtp_receiver,
+                            streams,
+                        })
+                        .await;
+                }
+            }
+            RTCEvent::Data { data, binary } => {
+                if !binary {
+                    Err(EngineError::Internal(
+                        "text messages aren't supported".to_string(),
+                    ))?;
+                }
+
+                let data = DataPacket::decode(&*data)?;
+                match data.value.unwrap() {
+                    Value::User(user) => {
+                        // TODO(theomonnom) Send event
+                    }
+                    Value::Speaker(_) => {
+                        // TODO(theomonnonm)
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn handle_signal(self: &Arc<Self>, event: signal_response::Message) -> EngineResult<()> {
+        match event {
+            signal_response::Message::Answer(answer) => {
+                trace!("received answer from the publisher: {:?}", answer);
+
+                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
+                self.session
+                    .read()
+                    .unwrap()
+                    .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
+                trace!("received offer for the subscriber: {:?}", offer);
+                let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
+
+                let subscriber_pc = self
+                    .session
+                    .read()
+                    .as_ref()
+                    .unwrap()
+                    .subscriber_pc
+                    .lock()
+                    .await;
+
+                subscriber_pc.set_remote_description(sdp).await?;
+                let answer = subscriber_pc
+                    .peer_connection()
+                    .create_answer(RTCOfferAnswerOptions::default())
+                    .await?;
+                subscriber_pc
+                    .peer_connection()
+                    .set_local_description(answer.clone())
+                    .await?;
+
+                tokio::spawn(async move {
+                    self.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)?;
+
+                trace!(
+                    "received ice_candidate {:?} - {:?}",
+                    SignalTarget::from_i32(trickle.target).unwrap(),
+                    ice
+                );
+
+                if trickle.target == SignalTarget::Publisher as i32 {
+                    self.session
+                        .read()
+                        .unwrap()
+                        .publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
+                } else {
+                    self.session
+                        .read()
+                        .unwrap()
+                        .subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
+                }
+            }
+            signal_response::Message::Update(update) => {
+                let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await;
+            }
+            _ => {}
+        }
+
+        Ok(())
+    }
+}
+
+/// Reconnection Logic for the RTCEngine, it is responsable for: TODO
+impl EngineInternal {
+    async fn handle_disconnected(self: &Arc<Self>) {
+        if self.closed.load(Ordering::SeqCst) || self.reconnecting.load(Ordering::SeqCst) {
+            return;
+        }
+
+        self.reconnecting.store(true, Ordering::SeqCst);
+        warn!("RTCEngine disconnected unexpectedly, reconnecting...");
+
+        let mut full_reconnect = false;
+        for i in 0..RECONNECT_ATTEMPTS {
+            if full_reconnect {
+                if i == 0 {
+                    let _ = emitter.send(EngineEvent::Restarting).await;
+                }
+
+                info!("restarting connection... attempt: {}", i);
+                if let Err(err) = Self::try_restart_connection(
+                    signal_client.clone(),
+                    engine_inner.clone(),
+                    emitter.clone(),
+                )
+                .await
+                {
+                    error!("restarting connection failed: {}", err);
+                } else {
+                    return;
+                }
+            } else {
+                if i == 0 {
+                    let _ = emitter.send(EngineEvent::Resuming).await;
+                }
+
+                info!("resuming connection... attempt: {}", i);
+                if let Err(err) = Self::try_resume_connection(
+                    signal_client.clone(),
+                    engine_inner.clone(),
+                    emitter.clone(),
+                )
+                .await
+                {
+                    error!("resuming connection failed: {}", err);
+                    if let EngineError::Signal(_) = err {
+                        full_reconnect = true;
+                    }
+                } else {
+                    return;
+                }
+            }
+
+            tokio::time::sleep(RECONNECT_INTERVAL).await;
+        }
+        error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
+        self.reconnecting.store(false, Ordering::SeqCst);
+
+        // TODO DISCONNECT
+    }
+
+    async fn try_restart_connection(
+        self: Arc<Self>,
+        signal_client: Arc<SignalClient>,
+        emitter: EngineEmitter,
+    ) -> EngineResult<()> {
+        Ok(())
+    }
+
+    async fn try_resume_connection(
+        self: Arc<Self>,
+        signal_client: Arc<SignalClient>,
+        emitter: EngineEmitter,
+    ) -> EngineResult<()> {
+        let mut options = engine_inner.options.lock().clone();
+        options.sid = engine_inner
+            .join_response
+            .lock()
+            .participant
+            .as_ref()
+            .unwrap()
+            .sid
+            .clone();
+
+        signal_client
+            .reconnect(
+                &engine_inner.url,
+                &engine_inner.token.lock().clone(),
+                options,
+            )
+            .await?;
+
+        let _ = emitter.send(EngineEvent::SignalResumed).await;
+
+        engine_inner
+            .subscriber_pc
+            .lock()
+            .await
+            .prepare_ice_restart();
+
+        if engine_inner.has_published.load(Ordering::SeqCst) {
+            engine_inner
+                .publisher_pc
+                .lock()
+                .await
+                .create_and_send_offer(RTCOfferAnswerOptions {
+                    ice_restart: true,
+                    ..Default::default()
+                })
+                .await?;
+        }
+
+        Self::wait_pc_connection(engine_inner).await?;
+        signal_client.flush_queue().await;
+
+        let _ = emitter.send(EngineEvent::Resumed);
+
+        Ok(())
+    }
+}

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

@@ -1,6 +1,6 @@
 use std::fmt::{Debug, Formatter};
 
-use tracing::{event, Level};
+use tracing::trace;
 
 use livekit_webrtc::peer_connection_factory::PeerConnectionFactory;
 use livekit_webrtc::webrtc::RTCRuntime;
@@ -19,9 +19,9 @@ impl Debug for LKRuntime {
     }
 }
 
-impl LKRuntime {
-    pub fn new() -> Self {
-        event!(Level::TRACE, "LKRuntime::new()");
+impl Default for LKRuntime {
+    fn default() -> Self {
+        trace!("LKRuntime::default()");
         let rtc_runtime = RTCRuntime::new();
         Self {
             pc_factory: PeerConnectionFactory::new(rtc_runtime.clone()),
@@ -32,6 +32,6 @@ impl LKRuntime {
 
 impl Drop for LKRuntime {
     fn drop(&mut self) {
-        event!(Level::TRACE, "LKRuntime::drop()");
+        trace!("LKRuntime::drop()");
     }
 }

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

@@ -34,44 +34,15 @@ use crate::rtc_engine::pc_transport::PCTransport;
 use crate::rtc_engine::rtc_events::{RTCEmitter, RTCEvent, RTCEvents};
 use crate::signal_client::{SignalClient, SignalError, SignalEvent, SignalEvents, SignalOptions};
 
+mod engine_internal;
 mod lk_runtime;
 mod pc_transport;
 mod rtc_events;
 
-lazy_static! {
-    // Share one LKRuntime across all RTCEngine instances
-    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
-}
-
 pub(crate) type EngineEmitter = mpsc::Sender<EngineEvent>;
 pub(crate) type EngineEvents = mpsc::Receiver<EngineEvent>;
 pub(crate) type EngineResult<T> = Result<T, EngineError>;
 
-// TODO(theomonnom): Smarter retry intervals
-pub(crate) const RECONNECT_ATTEMPTS: u32 = 10;
-pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300);
-
-pub(crate) const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
-pub(crate) const LOSSY_DC_LABEL: &str = "_lossy";
-pub(crate) const RELIABLE_DC_LABEL: &str = "_reliable";
-
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-pub(crate) enum PCState {
-    New,
-    Connected,
-    Disconnected,
-    Reconnecting,
-    Closed,
-}
-
-#[derive(Serialize, Deserialize)]
-#[allow(non_snake_case)]
-struct IceCandidateJSON {
-    sdpMid: String,
-    sdpMLineIndex: i32,
-    candidate: String,
-}
-
 #[derive(Error, Debug)]
 pub enum EngineError {
     #[error("signal failure: {0}")]
@@ -93,7 +64,7 @@ pub enum EngineError {
 }
 
 #[derive(Debug)]
-pub(crate) enum EngineEvent {
+pub enum EngineEvent {
     ParticipantUpdate(ParticipantUpdate),
     AddTrack {
         rtp_receiver: RtpReceiver,
@@ -107,115 +78,13 @@ pub(crate) enum EngineEvent {
     Restarted,
 }
 
-#[derive(Debug)]
-struct EngineInner {
-    // Join infornation
-    url: String,
-    token: Mutex<String>, // The token is refreshed periodically
-    options: Mutex<SignalOptions>,
-    join_response: Mutex<JoinResponse>,
-
-    has_published: AtomicBool,
-    pc_state: AtomicU8, // Casted to PCState enum
-    reconnecting: AtomicBool,
-
-    publisher_pc: AsyncMutex<PCTransport>,
-    subscriber_pc: AsyncMutex<PCTransport>,
-
-    // Publisher data channels
-    // Used to send data to other participants ( The SFU forward the messages )
-    lossy_dc: Mutex<DataChannel>,
-    reliable_dc: Mutex<DataChannel>,
-    // Subscriber data channels
-    // These fields are never used, we just keep a strong reference to them,
-    // so we can receive data from other participants
-    sub_reliable_dc: Mutex<Option<DataChannel>>,
-    sub_lossy_dc: Mutex<Option<DataChannel>>,
-
-    closed: AtomicBool,
-}
-
 #[derive(Debug)]
 pub struct RTCEngine {
-    signal_client: Arc<SignalClient>,
-    engine_inner: Arc<EngineInner>,
-
-    #[allow(unused)]
-    lk_runtime: Arc<LKRuntime>, // Keep a reference while we're using the RTCEngine
-}
-
-impl EngineInner {
-    async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> {
-        if !self.join_response.lock().subscriber_primary {
-            return Ok(());
-        }
-
-        let publisher = &self.publisher_pc;
-        {
-            let mut publisher = publisher.lock().await;
-            if !publisher.is_connected()
-                && publisher.peer_connection().ice_connection_state()
-                    != IceConnectionState::IceConnectionChecking
-            {
-                let _ = self.negotiate_publisher().await;
-            }
-        }
-
-        let dc = self.data_channel(kind);
-        if dc.lock().state() == DataState::Open {
-            return Ok(());
-        }
-
-        // Wait until the PeerConnection is connected
-        let wait_connected = async move {
-            while publisher.lock().await.is_connected() && dc.lock().state() == DataState::Open {
-                sleep(Duration::from_millis(50)).await;
-            }
-        };
-
-        tokio::select! {
-            _ = wait_connected => Ok(()),
-            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
-                let err = EngineError::Connection("could not establish publisher connection: timeout".to_string());
-                error!(error = ?err);
-                Err(err)
-            }
-        }
-    }
-
-    async fn negotiate_publisher(&self) -> EngineResult<()> {
-        self.has_published.store(true, Ordering::SeqCst);
-        if let Err(err) = self.publisher_pc.lock().await.negotiate().await {
-            error!("failed to negotiate the publisher: {:?}", err);
-            Err(err)?
-        } else {
-            Ok(())
-        }
-    }
-
-    fn data_channel(&self, kind: data_packet::Kind) -> &Mutex<DataChannel> {
-        if kind == data_packet::Kind::Reliable {
-            &self.reliable_dc
-        } else {
-            &self.lossy_dc
-        }
-    }
+    engine_inner: Arc<EngineInternal>,
 }
 
 impl RTCEngine {
     pub fn new() -> Self {
-
-		Self {
-
-		}
-	}
-
-    #[tracing::instrument(skip(url, token))]
-    pub(crate) async fn connect(
-        url: &str,
-        token: &str,
-        options: SignalOptions,
-    ) -> EngineResult<(RTCEngine, EngineEvents)> {
         let mut lk_runtime = None;
         {
             let mut lk_runtime_ref = LK_RUNTIME.lock();
@@ -228,7 +97,17 @@ impl RTCEngine {
             }
         }
 
-        let lk_runtime = lk_runtime.unwrap();
+        let (signal_client, mut signal_events) = SignalClient::new();
+
+        Self { lk_runtime }
+    }
+
+    #[tracing::instrument(skip(url, token))]
+    pub(crate) async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
+    ) -> EngineResult<(RTCEngine, EngineEvents)> {
         let (signal_client, mut signal_events) = SignalClient::connect(url, token, options).await?;
 
         let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
@@ -286,513 +165,7 @@ impl RTCEngine {
         self.engine_inner.join_response.lock().clone()
     }
 
-    async fn engine_task(
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        mut rtc_events: RTCEvents,
-        emitter: EngineEmitter,
-    ) {
-        while let Some(event) = rtc_events.recv().await {
-            if let Err(err) = Self::handle_rtc(
-                event,
-                signal_client.clone(),
-                engine_inner.clone(),
-                emitter.clone(),
-            )
-            .await
-            {
-                error!("failed to handle rtc event: {:?}", err);
-            }
-        }
-    }
-
-    async fn signal_task(
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        mut signal_events: SignalEvents,
-        emitter: EngineEmitter,
-    ) {
-        while let Some(signal) = signal_events.recv().await {
-            match signal {
-                SignalEvent::Open => {}
-                SignalEvent::Signal(signal) => {
-                    if let Err(err) = Self::handle_signal(
-                        signal,
-                        signal_client.clone(),
-                        engine_inner.clone(),
-                        emitter.clone(),
-                    )
-                    .await
-                    {
-                        error!("failed to handle signal: {:?}", err);
-                    }
-                }
-                SignalEvent::Close => {
-                    Self::handle_disconnected(
-                        signal_client.clone(),
-                        engine_inner.clone(),
-                        emitter.clone(),
-                    );
-                }
-            }
-        }
-    }
-
-    async fn handle_rtc(
-        event: RTCEvent,
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        emitter: EngineEmitter,
-    ) -> EngineResult<()> {
-        match event {
-            RTCEvent::IceCandidate {
-                ice_candidate,
-                target,
-            } => {
-                let json = serde_json::to_string(&IceCandidateJSON {
-                    sdpMid: ice_candidate.sdp_mid(),
-                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
-                    candidate: ice_candidate.candidate(),
-                })?;
-
-                trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate);
-
-                tokio::spawn(async move {
-                    signal_client
-                        .send(signal_request::Message::Trickle(TrickleRequest {
-                            candidate_init: json,
-                            target: target as i32,
-                        }))
-                        .await;
-                });
-            }
-            RTCEvent::ConnectionChange { state, target } => {
-                trace!("connection change, {:?} {:?}", state, target);
-                let subscriber_primary = engine_inner.join_response.lock().subscriber_primary;
-                let is_primary = subscriber_primary && target == SignalTarget::Subscriber;
-
-                if is_primary && state == PeerConnectionState::Connected {
-                    let old_state = engine_inner
-                        .pc_state
-                        .swap(PCState::Connected as u8, Ordering::SeqCst);
-                    if old_state == PCState::New as u8 {
-                        let _ = emitter.send(EngineEvent::Connected).await; // First time connected
-                    }
-                } else if state == PeerConnectionState::Failed {
-                    engine_inner
-                        .pc_state
-                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
-
-                    Self::handle_disconnected(signal_client, engine_inner, emitter);
-                }
-            }
-            RTCEvent::DataChannel {
-                data_channel,
-                target,
-            } => {
-                if target == SignalTarget::Subscriber {
-                    if data_channel.label() == RELIABLE_DC_LABEL {
-                        *engine_inner.sub_reliable_dc.lock() = Some(data_channel);
-                    } else {
-                        *engine_inner.sub_lossy_dc.lock() = Some(data_channel);
-                    }
-                }
-            }
-            RTCEvent::Offer { offer, target } => {
-                if target == SignalTarget::Publisher {
-                    // Send the publisher offer to the server
-                    tokio::spawn(async move {
-                        signal_client
-                            .send(signal_request::Message::Offer(proto::SessionDescription {
-                                r#type: "offer".to_string(),
-                                sdp: offer.to_string(),
-                            }))
-                            .await;
-                    });
-                }
-            }
-            RTCEvent::AddTrack {
-                rtp_receiver,
-                streams,
-                target,
-            } => {
-                if target == SignalTarget::Subscriber {
-                    let _ = emitter
-                        .send(EngineEvent::AddTrack {
-                            rtp_receiver,
-                            streams,
-                        })
-                        .await;
-                }
-            }
-            RTCEvent::Data { data, binary } => {
-                if !binary {
-                    Err(EngineError::Internal(
-                        "text messages aren't supported".to_string(),
-                    ))?;
-                }
-
-                let data = DataPacket::decode(&*data)?;
-                match data.value.unwrap() {
-                    Value::User(user) => {
-                        // TODO(theomonnom) Send event
-                    }
-                    Value::Speaker(_) => {
-                        // TODO(theomonnonm)
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn handle_signal(
-        event: signal_response::Message,
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        emitter: EngineEmitter,
-    ) -> EngineResult<()> {
-        match event {
-            signal_response::Message::Answer(answer) => {
-                trace!("received answer from the publisher: {:?}", answer);
-
-                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
-                engine_inner
-                    .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
-                trace!("received offer for the subscriber: {:?}", offer);
-                let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
-
-                engine_inner
-                    .subscriber_pc
-                    .lock()
-                    .await
-                    .set_remote_description(sdp)
-                    .await?;
-                let answer = engine_inner
-                    .subscriber_pc
-                    .lock()
-                    .await
-                    .peer_connection()
-                    .create_answer(RTCOfferAnswerOptions::default())
-                    .await?;
-                engine_inner
-                    .subscriber_pc
-                    .lock()
-                    .await
-                    .peer_connection()
-                    .set_local_description(answer.clone())
-                    .await?;
-
-                tokio::spawn(async move {
-                    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)?;
-
-                trace!(
-                    "received ice_candidate {:?} - {:?}",
-                    SignalTarget::from_i32(trickle.target).unwrap(),
-                    ice
-                );
-
-                if trickle.target == SignalTarget::Publisher as i32 {
-                    engine_inner
-                        .publisher_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice)
-                        .await?;
-                } else {
-                    engine_inner
-                        .subscriber_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice)
-                        .await?;
-                }
-            }
-            signal_response::Message::Update(update) => {
-                let _ = emitter.send(EngineEvent::ParticipantUpdate(update)).await;
-            }
-            _ => {}
-        }
-
-        Ok(())
-    }
-
-    async fn handle_disconnected(
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        emitter: EngineEmitter,
-    ) {
-        if engine_inner.closed.load(Ordering::SeqCst)
-            || engine_inner.reconnecting.load(Ordering::SeqCst)
-        {
-            return;
-        }
-
-        engine_inner.reconnecting.store(true, Ordering::SeqCst);
-        warn!("RTCEngine disconnected unexpectedly, reconnecting...");
-
-        let mut full_reconnect = false;
-        for i in 0..RECONNECT_ATTEMPTS {
-            if full_reconnect {
-                if i == 0 {
-                    let _ = emitter.send(EngineEvent::Restarting).await;
-                }
-
-                info!("restarting connection... attempt: {}", i);
-                if let Err(err) = Self::try_restart_connection(
-                    signal_client.clone(),
-                    engine_inner.clone(),
-                    emitter.clone(),
-                )
-                .await
-                {
-                    error!("restarting connection failed: {}", err);
-                } else {
-                    return;
-                }
-            } else {
-                if i == 0 {
-                    let _ = emitter.send(EngineEvent::Resuming).await;
-                }
-
-                info!("resuming connection... attempt: {}", i);
-                if let Err(err) = Self::try_resume_connection(
-                    signal_client.clone(),
-                    engine_inner.clone(),
-                    emitter.clone(),
-                )
-                .await
-                {
-                    error!("resuming connection failed: {}", err);
-                    if let EngineError::Signal(_) = err {
-                        full_reconnect = true;
-                    }
-                } else {
-                    return;
-                }
-            }
-
-            tokio::time::sleep(RECONNECT_INTERVAL).await;
-        }
-        error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
-        engine_inner.reconnecting.store(false, Ordering::SeqCst);
-
-        // TODO DISCONNECT
-    }
-
-    async fn try_restart_connection(
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        emitter: EngineEmitter,
-    ) -> EngineResult<()> {
-        Ok(())
-    }
-
-    async fn try_resume_connection(
-        signal_client: Arc<SignalClient>,
-        engine_inner: Arc<EngineInner>,
-        emitter: EngineEmitter,
-    ) -> EngineResult<()> {
-        let mut options = engine_inner.options.lock().clone();
-        options.sid = engine_inner
-            .join_response
-            .lock()
-            .participant
-            .as_ref()
-            .unwrap()
-            .sid
-            .clone();
-
-        signal_client
-            .reconnect(
-                &engine_inner.url,
-                &engine_inner.token.lock().clone(),
-                options,
-            )
-            .await?;
-
-        let _ = emitter.send(EngineEvent::SignalResumed).await;
-
-        engine_inner
-            .subscriber_pc
-            .lock()
-            .await
-            .prepare_ice_restart();
-
-        if engine_inner.has_published.load(Ordering::SeqCst) {
-            engine_inner
-                .publisher_pc
-                .lock()
-                .await
-                .create_and_send_offer(RTCOfferAnswerOptions {
-                    ice_restart: true,
-                    ..Default::default()
-                })
-                .await?;
-        }
-
-        Self::wait_pc_connection(engine_inner).await?;
-        signal_client.flush_queue().await;
-
-        let _ = emitter.send(EngineEvent::Resumed);
-
-        Ok(())
-    }
-
-    async fn wait_pc_connection(engine_inner: Arc<EngineInner>) -> EngineResult<()> {
-        let wait_connected = async move {
-            while engine_inner.pc_state.load(Ordering::SeqCst) != PCState::Connected as u8 {
-                sleep(Duration::from_millis(50)).await;
-            }
-        };
-
-        tokio::select! {
-            _ = wait_connected => Ok(()),
-            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
-                let err = EngineError::Connection("wait_pc_connection timed out".to_string());
-                Err(err)
-            }
-        }
-    }
-
     fn close(&self) {
         // TODO
     }
-
-    fn configure_engine(
-        lk_runtime: Arc<LKRuntime>,
-        join_response: JoinResponse,
-    ) -> EngineResult<(EngineInner, RTCEvents)> {
-        let (rtc_emitter, events) = mpsc::unbounded_channel();
-        let rtc_config = RTCConfiguration::from(join_response.clone());
-
-        let mut publisher_pc = PCTransport::new(
-            lk_runtime
-                .pc_factory
-                .create_peer_connection(rtc_config.clone())?,
-        );
-
-        let mut subscriber_pc = PCTransport::new(
-            lk_runtime
-                .pc_factory
-                .create_peer_connection(rtc_config.clone())?,
-        );
-
-        let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(
-            LOSSY_DC_LABEL,
-            DataChannelInit {
-                ordered: true,
-                max_retransmits: Some(0),
-                ..DataChannelInit::default()
-            },
-        )?;
-
-        let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(
-            RELIABLE_DC_LABEL,
-            DataChannelInit {
-                ordered: true,
-                ..DataChannelInit::default()
-            },
-        )?;
-
-        publisher_pc
-            .peer_connection()
-            .on_ice_candidate(rtc_events::on_ice_candidate(
-                SignalTarget::Publisher,
-                rtc_emitter.clone(),
-            ));
-        subscriber_pc
-            .peer_connection()
-            .on_ice_candidate(rtc_events::on_ice_candidate(
-                SignalTarget::Subscriber,
-                rtc_emitter.clone(),
-            ));
-
-        publisher_pc.on_offer(rtc_events::on_offer(
-            SignalTarget::Publisher,
-            rtc_emitter.clone(),
-        ));
-        subscriber_pc.on_offer(rtc_events::on_offer(
-            SignalTarget::Subscriber,
-            rtc_emitter.clone(),
-        ));
-
-        publisher_pc
-            .peer_connection()
-            .on_data_channel(rtc_events::on_data_channel(
-                SignalTarget::Publisher,
-                rtc_emitter.clone(),
-            ));
-        subscriber_pc
-            .peer_connection()
-            .on_data_channel(rtc_events::on_data_channel(
-                SignalTarget::Subscriber,
-                rtc_emitter.clone(),
-            ));
-
-        publisher_pc
-            .peer_connection()
-            .on_add_track(rtc_events::on_add_track(
-                SignalTarget::Publisher,
-                rtc_emitter.clone(),
-            ));
-        subscriber_pc
-            .peer_connection()
-            .on_add_track(rtc_events::on_add_track(
-                SignalTarget::Subscriber,
-                rtc_emitter.clone(),
-            ));
-
-        publisher_pc
-            .peer_connection()
-            .on_connection_change(rtc_events::on_connection_change(
-                SignalTarget::Publisher,
-                rtc_emitter.clone(),
-            ));
-        subscriber_pc
-            .peer_connection()
-            .on_connection_change(rtc_events::on_connection_change(
-                SignalTarget::Subscriber,
-                rtc_emitter.clone(),
-            ));
-
-        lossy_dc.on_message(rtc_events::on_message(rtc_emitter.clone()));
-        reliable_dc.on_message(rtc_events::on_message(rtc_emitter.clone()));
-
-        Ok((
-            EngineInner {
-                has_published: AtomicBool::new(false),
-                join_response: Mutex::new(join_response),
-                pc_state: AtomicU8::new(PCState::New as u8),
-                publisher_pc: AsyncMutex::new(publisher_pc),
-                subscriber_pc: AsyncMutex::new(subscriber_pc),
-                lossy_dc: Mutex::new(lossy_dc),
-                reliable_dc: Mutex::new(reliable_dc),
-                sub_lossy_dc: Mutex::new(None),
-                sub_reliable_dc: Mutex::new(None),
-                closed: AtomicBool::new(false),
-            },
-            events,
-        ))
-    }
 }

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

@@ -11,6 +11,8 @@ use livekit_webrtc::peer_connection::{
 };
 use livekit_webrtc::rtc_error::RTCError;
 
+use crate::proto::SignalTarget;
+
 const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150);
 
 pub type OnOfferHandler = Box<
@@ -20,6 +22,7 @@ pub type OnOfferHandler = Box<
 >;
 
 pub(crate) struct PCTransport {
+    signal_target: SignalTarget,
     peer_connection: PeerConnection,
     pending_candidates: Vec<IceCandidate>,
     on_offer_handler: Option<OnOfferHandler>,
@@ -34,8 +37,9 @@ impl Debug for PCTransport {
 }
 
 impl PCTransport {
-    pub fn new(peer_connection: PeerConnection) -> Self {
+    pub fn new(peer_connection: PeerConnection, signal_target: SignalTarget) -> Self {
         Self {
+            signal_target,
             peer_connection,
             pending_candidates: Vec::default(),
             on_offer_handler: None,
@@ -54,6 +58,10 @@ impl PCTransport {
         &mut self.peer_connection
     }
 
+    pub fn signal_target(&self) -> SignalTarget {
+        self.signal_target.clone()
+    }
+
     pub fn on_offer(&mut self, handler: OnOfferHandler) {
         self.on_offer_handler = Some(handler);
     }

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

@@ -11,11 +11,13 @@ use tokio::sync::mpsc;
 use crate::proto::SignalTarget;
 use crate::rtc_engine::pc_transport::OnOfferHandler;
 
-pub(super) type RTCEmitter = mpsc::UnboundedSender<RTCEvent>;
-pub(super) type RTCEvents = mpsc::UnboundedReceiver<RTCEvent>;
+use super::pc_transport::PCTransport;
+
+pub type RTCEmitter = mpsc::UnboundedSender<RTCEvent>;
+pub type RTCEvents = mpsc::UnboundedReceiver<RTCEvent>;
 
 #[derive(Debug)]
-pub(super) enum RTCEvent {
+pub enum RTCEvent {
     IceCandidate {
         ice_candidate: IceCandidate,
         target: SignalTarget,
@@ -43,19 +45,16 @@ pub(super) enum RTCEvent {
     },
 }
 
-/// Handlers used to forward event to a channel
+/// Handlers used to forward events to a channel
 /// Every callback here is called on the signaling thread
 
-pub(super) fn on_connection_change(
-    target: SignalTarget,
-    emitter: RTCEmitter,
-) -> OnConnectionChangeHandler {
+fn on_connection_change(target: SignalTarget, emitter: RTCEmitter) -> OnConnectionChangeHandler {
     Box::new(move |state| {
         let _ = emitter.send(RTCEvent::ConnectionChange { state, target });
     })
 }
 
-pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler {
+fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnIceCandidateHandler {
     Box::new(move |ice_candidate| {
         let _ = emitter.send(RTCEvent::IceCandidate {
             ice_candidate,
@@ -64,7 +63,7 @@ pub(super) fn on_ice_candidate(target: SignalTarget, emitter: RTCEmitter) -> OnI
     })
 }
 
-pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler {
+fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHandler {
     Box::new(move |offer| {
         let _ = emitter.send(RTCEvent::Offer { offer, target });
 
@@ -72,7 +71,7 @@ pub(super) fn on_offer(target: SignalTarget, emitter: RTCEmitter) -> OnOfferHand
     })
 }
 
-pub(super) fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler {
+fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDataChannelHandler {
     Box::new(move |mut data_channel| {
         data_channel.on_message(on_message(emitter.clone()));
 
@@ -83,7 +82,7 @@ pub(super) fn on_data_channel(target: SignalTarget, emitter: RTCEmitter) -> OnDa
     })
 }
 
-pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler {
+fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTrackHandler {
     Box::new(move |rtp_receiver, streams| {
         let _ = emitter.send(RTCEvent::AddTrack {
             rtp_receiver,
@@ -93,7 +92,34 @@ pub(super) fn on_add_track(target: SignalTarget, emitter: RTCEmitter) -> OnAddTr
     })
 }
 
-pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
+pub fn forward_pc_events(transport: &mut PCTransport, rtc_emitter: RTCEmitter) {
+    transport
+        .peer_connection()
+        .on_ice_candidate(on_ice_candidate(
+            transport.signal_target(),
+            rtc_emitter.clone(),
+        ));
+
+    transport.peer_connection().on_data_channel(on_data_channel(
+        transport.signal_target(),
+        rtc_emitter.clone(),
+    ));
+
+    transport
+        .peer_connection()
+        .on_add_track(on_add_track(transport.signal_target(), rtc_emitter.clone()));
+
+    transport
+        .peer_connection()
+        .on_connection_change(on_connection_change(
+            transport.signal_target(),
+            rtc_emitter.clone(),
+        ));
+
+    transport.on_offer(on_offer(transport.signal_target(), rtc_emitter.clone()));
+}
+
+fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
     Box::new(move |data, binary| {
         let _ = emitter.send(RTCEvent::Data {
             data: data.to_vec(),
@@ -101,3 +127,7 @@ pub(super) fn on_message(emitter: RTCEmitter) -> OnMessageHandler {
         });
     })
 }
+
+pub fn forward_dc_events(dc: &mut DataChannel, rtc_emitter: RTCEmitter) {
+    dc.on_message(on_message(rtc_emitter.clone()));
+}

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

@@ -61,34 +61,23 @@ impl Default for SignalOptions {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct SignalClient {
     stream: RwLock<Option<SignalStream>>,
-    emitter: SignalEmitter,
 }
 
 impl SignalClient {
-    pub fn new() -> (Self, SignalEvents) {
-        let (emitter, events) = mpsc::channel(8);
-        (
-            Self {
-                stream: Default::default(),
-                emitter,
-            },
-            events,
-        )
-    }
-
     #[instrument(level = Level::DEBUG, skip(url, token, options))]
     pub(crate) async fn connect(
         &self,
         url: &str,
         token: &str,
         options: SignalOptions,
-    ) -> SignalResult<()> {
-        let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?;
+    ) -> SignalResult<SignalEvents> {
+        let (emitter, events) = mpsc::channel(8);
+        let stream = SignalStream::connect(url, token, options, emitter).await?;
         *self.stream.write() = Some(stream);
-        Ok(())
+        Ok(events)
     }
 
     #[instrument(level = Level::DEBUG)]
@@ -142,14 +131,15 @@ impl From<JoinResponse> for RTCConfiguration {
 pub mod utils {
     use crate::proto::{signal_response, JoinResponse};
     use crate::signal_client::{SignalError, SignalEvent, SignalResult, JOIN_RESPONSE_TIMEOUT};
-    use tokio::sync::mpsc;
     use tokio::time::timeout;
     use tokio_tungstenite::tungstenite::Error as WsError;
     use tracing::{event, instrument, Level};
 
+    use super::SignalEvents;
+
     #[instrument(level = Level::DEBUG, skip(receiver))]
     pub(crate) async fn next_join_response(
-        receiver: &mut mpsc::Receiver<SignalEvent>,
+        receiver: &mut SignalEvents,
     ) -> SignalResult<JoinResponse> {
         let join = async {
             while let Some(event) = receiver.recv().await {

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

@@ -77,9 +77,8 @@ impl SignalStream {
         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));
+        let write_handle = tokio::spawn(Self::write_task(internal_rx, ws_writer, emitter.clone()));
+        let read_handle = tokio::spawn(Self::read_task(internal_tx.clone(), ws_reader, emitter));
 
         Ok(Self {
             internal_tx,
@@ -119,7 +118,7 @@ impl SignalStream {
 
     /// This task is used to send messages to the websocket
     /// It is also responsible for closing the connection
-    async fn handle_write(
+    async fn write_task(
         mut internal_rx: mpsc::Receiver<InternalMessage>,
         mut ws_writer: SplitSink<WebSocket, Message>,
         emitter: SignalEmitter,
@@ -170,7 +169,7 @@ impl SignalStream {
     /// and dispatch them through the EventEmitter.
     ///
     /// It can also send messages to [handle_write] task ( Used e.g. answer to pings )
-    async fn handle_read(
+    async fn read_task(
         internal_tx: mpsc::Sender<InternalMessage>,
         mut ws_reader: SplitStream<WebSocket>,
         emitter: SignalEmitter,