rtc_session WIP

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

@@ -19,6 +19,8 @@ use livekit_webrtc::peer_connection::{
 };
 use livekit_webrtc::peer_connection_factory::RTCConfiguration;
 
+use super::rtc_session::{RTCSession, SessionInfo};
+use super::{rtc_events, EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult};
 use crate::proto::data_packet::Value;
 use crate::proto::{
     data_packet, signal_request, signal_response, DataPacket, JoinResponse, ParticipantUpdate,
@@ -28,52 +30,28 @@ 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 super::{rtc_events, 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 EngineInternal {
     lk_runtime: Arc<LKRuntime>,
     session: AsyncRwLock<RTCSession>,
-    signal_client: SignalClient,
+    signal_client: Arc<SignalClient>,
     reconnecting: AtomicBool,
     closed: AtomicBool,
-    has_published: AtomicBool,
-    pc_state: AtomicU8, // Casted to PCState enum
     engine_emitter: EngineEmitter,
 }
 
-#[derive(Serialize, Deserialize)]
-#[allow(non_snake_case)]
-struct IceCandidateJSON {
-    sdpMid: String,
-    sdpMLineIndex: i32,
-    candidate: String,
-}
-
 impl EngineInternal {
     #[tracing::instrument]
     pub async fn connect(
@@ -93,13 +71,6 @@ impl EngineInternal {
             }
         }
         let lk_runtime = lk_runtime.unwrap();
-
-        // Connect to the SignalClient
-        let signal_client = SignalClient::default();
-        let mut signal_events = signal_client.connect(url, token, options.clone()).await?;
-        let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
-        debug!("received JoinResponse: {:?}", join_response);
-
         // Configure the PeerConnections/RTCSession
         let (engine_emitter, engine_events) = mpsc::channel(8);
         let session_info = SessionInfo {
@@ -112,7 +83,6 @@ impl EngineInternal {
         let rtc_session = AsyncRwLock::new(rtc_session);
         let rtc_internal = Arc::new(Self {
             lk_runtime,
-            info: Mutex::new(session_info),
             session: rtc_session,
             signal_client,
             reconnecting: AtomicBool::new(false),
@@ -165,23 +135,27 @@ impl EngineInternal {
             } => {
                 trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate);
 
-                let json = serde_json::to_string(&IceCandidateJSON {
-                    sdpMid: ice_candidate.sdp_mid(),
-                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
-                    candidate: ice_candidate.candidate(),
-                })?;
-
                 self.signal_client
                     .send(signal_request::Message::Trickle(TrickleRequest {
-                        candidate_init: json,
+                        candidate_init: serde_json::to_string(&IceCandidateJSON {
+                            sdpMid: ice_candidate.sdp_mid(),
+                            sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                            candidate: ice_candidate.candidate(),
+                        })?,
                         target: target as i32,
                     }))
                     .await;
             }
             RTCEvent::ConnectionChange { state, target } => {
                 trace!("connection change, {:?} {:?}", state, target);
-                let subscriber_primary = self.info.lock().join_response.subscriber_primary;
+                let is_primary = self
-                let is_primary = subscriber_primary && target == SignalTarget::Subscriber;
+                    .session
+                    .read()
+                    .await
+                    .info()
+                    .join_response
+                    .subscriber_primary
+                    && target == SignalTarget::Subscriber;
 
                 if is_primary && state == PeerConnectionState::Connected {
                     let old_state = self
@@ -189,7 +163,6 @@ impl EngineInternal {
                         .swap(PCState::Connected as u8, Ordering::SeqCst);
                     if old_state == PCState::New as u8 {
                         let _ = self.engine_emitter.send(EngineEvent::Connected).await;
-                        // First time connected
                     }
                 } else if state == PeerConnectionState::Failed {
                     self.pc_state
@@ -203,11 +176,7 @@ impl EngineInternal {
                 target,
             } => {
                 if target == SignalTarget::Subscriber {
-                    if data_channel.label() == RELIABLE_DC_LABEL {
+                    self.session.read().await.use_data_channel(data_channel);
-                        *self.session.read().await.sub_reliable_dc.lock() = Some(data_channel);
-                    } else {
-                        *self.session.read().await.sub_lossy_dc.lock() = Some(data_channel);
-                    }
                 }
             }
             RTCEvent::Offer { offer, target } => {
@@ -259,75 +228,13 @@ impl EngineInternal {
     }
 
     async fn handle_signal(self: &Arc<Self>, event: signal_response::Message) -> EngineResult<()> {
+        self.session
+            .read()
+            .await
+            .on_signal_event(self.signal_client.clone(), event.clone())
+            .await?;
+
         match event {
-            signal_response::Message::Answer(answer) => {
-                trace!("received answer from the publisher: {:?}", answer);
-
-                let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
-                self.session
-                    .read()
-                    .await
-                    .publisher_pc
-                    .lock()
-                    .await
-                    .set_remote_description(answer)
-                    .await?;
-            }
-            signal_response::Message::Offer(offer) => {
-                trace!("received offer for the subscriber: {:?}", offer);
-                // Handle the subscriber offer & send an answer to livekit-server
-                // We always get an offer from the server when connecting
-
-                let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
-                let answer = self
-                    .session
-                    .read()
-                    .await
-                    .subscriber_pc
-                    .lock()
-                    .await
-                    .create_anwser(offer, RTCOfferAnswerOptions::default())
-                    .await?;
-
-                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_candidate =
-                    IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
-                let signal_target = SignalTarget::from_i32(trickle.target).unwrap();
-                trace!(
-                    "received ice_candidate {:?} {:?}",
-                    signal_target,
-                    ice_candidate
-                );
-
-                if trickle.target == SignalTarget::Publisher as i32 {
-                    self.session
-                        .read()
-                        .await
-                        .publisher_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice_candidate)
-                        .await?;
-                } else {
-                    self.session
-                        .read()
-                        .await
-                        .subscriber_pc
-                        .lock()
-                        .await
-                        .add_ice_candidate(ice_candidate)
-                        .await?;
-                }
-            }
             signal_response::Message::Update(update) => {
                 let _ = self
                     .engine_emitter
@@ -340,79 +247,6 @@ impl EngineInternal {
         Ok(())
     }
 
-    /// Start publisher negotiation
-    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)
-    }
-
-    /// Ensure the Publisher PC is connected, if not, start the negotiation
-    /// This is required when sending data to the server
-    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
-        }
-    }
 }
 
 /// Reconnection logic impl, LiveKit handles reconnection in two ways:

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

@@ -70,6 +70,10 @@ impl PCTransport {
         self.restarting_ice = true;
     }
 
+    pub fn close(&mut self) {
+        self.peer_connection.close();
+    }
+
     #[tracing::instrument(level = Level::DEBUG)]
     pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> {
         if self.peer_connection.remote_description().is_none() {

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

@@ -28,16 +28,14 @@ pub enum RTCEvent {
     },
     DataChannel {
         data_channel: DataChannel,
-        target: SignalTarget,
     },
+    // TODO (theomonnom): Move Offer to PCTransport
     Offer {
         offer: SessionDescription,
-        target: SignalTarget,
     },
     AddTrack {
         rtp_receiver: RtpReceiver,
         streams: Vec<MediaStream>,
-        target: SignalTarget,
     },
     Data {
         data: Vec<u8>,

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

@@ -1,38 +1,138 @@
-#[derive(Debug, Clone, Default)]
+use livekit_webrtc::media_stream::{MediaStream, MediaStreamTrackHandle};
-pub struct SessionInfo {
+use livekit_webrtc::rtp_receiver::RtpReceiver;
-    url: String,
+use parking_lot::{Mutex, RwLock};
-    token: String,
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
-    options: SignalOptions,
+use std::sync::{Arc, Weak};
-    join_response: JoinResponse,
+use std::time::Duration;
+use tokio::task::JoinHandle;
+
+use tokio::sync::{broadcast, mpsc, watch, Mutex as AsyncMutex, RwLock as AsyncRwLock};
+use tokio::time::sleep;
+
+use lazy_static::lazy_static;
+use prost::Message;
+use serde::{Deserialize, Serialize};
+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::peer_connection::{
+    IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
+};
+use livekit_webrtc::peer_connection_factory::RTCConfiguration;
+
+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 super::{rtc_events, EngineEmitter, EngineError, EngineEvent, EngineEvents, EngineResult};
+
+pub const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+pub const LOSSY_DC_LABEL: &str = "_lossy";
+pub const RELIABLE_DC_LABEL: &str = "_reliable";
+
+pub type SessionEmitter = mpsc::UnboundedSender<SessionEvent>;
+pub type SessionEvents = mpsc::UnboundedReceiver<SessionEvent>;
+
+#[derive(Debug)]
+pub enum SessionEvent {
+    Data {
+        data: Vec<u8>,
+    },
+    MediaTrack {
+        track: MediaStreamTrackHandle,
+        stream: MediaStream,
+        receiver: RtpReceiver,
+    },
+    Close {
+        reason: String,
+    },
 }
 
-/// This struct holds a WebRTC session
+#[repr(u8)]
-/// The session changes at every reconnection
+pub enum PCState {
+    New,
+    Connected,
+    Disconnected,
+    Reconnecting,
+    Closed,
+}
+
+#[derive(Serialize, Deserialize)]
+#[allow(non_snake_case)]
+struct IceCandidateJSON {
+    sdpMid: String,
+    sdpMLineIndex: i32,
+    candidate: String,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct SessionInfo {
+    pub url: String,
+    pub token: String,
+    pub options: SignalOptions,
+    pub join_response: JoinResponse,
+}
+
+/// Fields shared with engine_task and signal_task
 #[derive(Debug)]
-pub struct RTCSession {
+struct SessionInner {
     info: SessionInfo,
+    signal_client: Arc<SignalClient>,
+    pc_state: AtomicU8, // PCState
+    has_published: AtomicBool,
+
     publisher_pc: AsyncMutex<PCTransport>,
     subscriber_pc: AsyncMutex<PCTransport>,
 
     // Publisher data channels
-    // Used to send data to other participants ( The SFU forwards the messages )
+    // used to send data to other participants ( The SFU forwards the messages )
     lossy_dc: DataChannel,
     reliable_dc: DataChannel,
 
-    // Subscriber data channels
+    // Keep a strong reference to the subscriber datachannels,
-    // 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>>,
+    subscriber_dc: Mutex<Vec<DataChannel>>,
-    sub_lossy_dc: Mutex<Option<DataChannel>>,
+
+    emitter: SessionEmitter,
+}
+
+/// This struct holds a WebRTC session
+/// The session changes at every reconnection
+///
+/// RTCSession is also responsable for the signaling and the negotation
+#[derive(Debug)]
+pub struct RTCSession {
+    lk_runtime: Arc<LKRuntime>,
+    inner: Arc<SessionInner>,
+    close_emitter: watch::Sender<bool>, // false = is_running
+    signal_task: JoinHandle<()>,
+    engine_task: JoinHandle<()>,
 }
 
 impl RTCSession {
-    pub fn new(
+    pub async fn connect(
+        url: &str,
+        token: &str,
+        options: SignalOptions,
         lk_runtime: Arc<LKRuntime>,
-        session_info: SessionInfo,
+    ) -> EngineResult<(Self, SessionEvents)> {
-    ) -> EngineResult<(Self, RTCEvents)> {
+        // Connect to the SignalClient
-        let (rtc_emitter, events) = mpsc::unbounded_channel();
+        let (signal_client, mut signal_events) = SignalClient::new();
-        let rtc_config = RTCConfiguration::from(session_info.join_response);
+        let signal_client = Arc::new(signal_client);
+        signal_client.connect(url, token, options.clone()).await?;
+        let join_response = signal_client::utils::next_join_response(&mut signal_events).await?;
+        debug!("received JoinResponse: {:?}", join_response);
+
+        let (rtc_emitter, rtc_events) = mpsc::unbounded_channel();
+        let rtc_config = RTCConfiguration::from(join_response.clone());
 
         let mut publisher_pc = PCTransport::new(
             lk_runtime
@@ -70,19 +170,366 @@ impl RTCSession {
         rtc_events::forward_dc_events(&mut lossy_dc, rtc_emitter.clone());
         rtc_events::forward_dc_events(&mut reliable_dc, rtc_emitter.clone());
 
-        Ok((
+        let session_info = SessionInfo {
-            Self {
+            url: url.to_owned(),
-                info: session_info,
+            token: token.to_owned(),
-                publisher_pc: AsyncMutex::new(publisher_pc),
+            options,
-                subscriber_pc: AsyncMutex::new(subscriber_pc),
+            join_response,
-                sub_lossy_dc: Default::default(),
+        };
-                sub_reliable_dc: Default::default(),
+
-                lossy_dc,
+        // Start tasks
-                reliable_dc,
+        let (session_emitter, session_events) = mpsc::unbounded_channel();
-            },
+        let (close_emitter, close_receiver) = watch::channel(false);
-            events,
+
-        ))
+        let inner = Arc::new(SessionInner {
+            info: session_info,
+            pc_state: AtomicU8::new(PCState::New as u8),
+            has_published: Default::default(),
+            signal_client,
+            publisher_pc: AsyncMutex::new(publisher_pc),
+            subscriber_pc: AsyncMutex::new(subscriber_pc),
+            lossy_dc,
+            reliable_dc,
+            subscriber_dc: Default::default(),
+            emitter: session_emitter,
+        });
+
+        // Start session tasks
+        let signal_task = tokio::spawn(
+            inner
+                .clone()
+                .signal_task(signal_events, close_receiver.clone()),
+        );
+        let engine_task = tokio::spawn(
+            inner
+                .clone()
+                .engine_task(rtc_events, close_receiver.clone()),
+        );
+
+        let session = Self {
+            lk_runtime,
+            inner,
+            close_emitter,
+            signal_task,
+            engine_task,
+        };
+
+        if !join_response.subscriber_primary {
+            inner.negotiate_publisher().await?;
+        }
+
+        Ok((session, session_events))
     }
 
-    
+    /// Close the PeerConnections and the SignalClient
+    pub async fn close(self) {
+        // Close the tasks
+        self.close_emitter.send(true);
+        self.engine_task.await;
+        self.signal_task.await;
+        self.inner.close().await;
+    }
+
+    pub async fn wait_pc_connection(&self) -> EngineResult<()> {
+        let wait_connected = async move {
+            while self.inner.pc_state.load(Ordering::Acquire) != PCState::Connected as u8 {
+                tokio::task::yield_now().await;
+            }
+        };
+
+        tokio::select! {
+            _ = wait_connected => Ok(()),
+            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
+                let err = EngineError::Connection("wait_pc_connection timed out".to_string());
+                Err(err)
+            }
+        }
+    }
+}
+
+impl RTCSession {
+    pub fn info(&self) -> &SessionInfo {
+        &self.inner.info
+    }
+
+    pub fn publisher(&self) -> &AsyncMutex<PCTransport> {
+        &self.inner.publisher_pc
+    }
+
+    pub fn subscriber(&self) -> &AsyncMutex<PCTransport> {
+        &self.inner.subscriber_pc
+    }
+
+    pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
+        &self.inner.data_channel(kind)
+    }
+}
+
+impl SessionInner {
+    async fn engine_task(
+        self: Arc<Self>,
+        mut rtc_events: RTCEvents,
+        mut close_receiver: watch::Receiver<bool>,
+    ) {
+        loop {
+            tokio::select! {
+                res = rtc_events.recv() => {
+                    if let Some(event) = res {
+                        if let Err(err) = self.on_rtc_event(event).await {
+                            error!("failed to handle rtc event: {:?}", err);
+                        }
+                    } else {
+                        panic!("rtc_events has been closed unexpectedly");
+                    }
+                },
+                 _ = close_receiver.changed() => {
+                    break;
+                }
+            }
+        }
+    }
+
+    async fn signal_task(
+        self: Arc<Self>,
+        mut signal_events: SignalEvents,
+        mut close_receiver: watch::Receiver<bool>,
+    ) {
+        loop {
+            tokio::select! {
+                res = signal_events.recv() => {
+                    if let Some(signal) = res {
+                        match signal {
+                            SignalEvent::Open => {}
+                            SignalEvent::Signal(signal) => {
+                                if let Err(err) = self.on_signal_event(signal).await {
+                                    error!("failed to handle signal: {:?}", err);
+                                }
+                            }
+                            SignalEvent::Close => {
+                                self.on_session_disconnected("SignalClient closed");
+                            }
+                        }
+                    } else {
+                        panic!("signal_events has been closed unexpectedly");
+                    }
+
+                },
+                _ = close_receiver.changed() => {
+                    break;
+                }
+            }
+        }
+    }
+
+    async fn on_signal_event(&self, event: signal_response::Message) -> EngineResult<()> {
+        match event {
+            signal_response::Message::Answer(answer) => {
+                trace!("received publisher answer: {:?}", answer);
+                let answer = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
+                self.publisher_pc
+                    .lock()
+                    .await
+                    .set_remote_description(answer)
+                    .await?;
+            }
+            signal_response::Message::Offer(offer) => {
+                trace!("received subscriber offer: {:?}", offer);
+                let offer = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
+                let answer = self
+                    .subscriber_pc
+                    .lock()
+                    .await
+                    .create_anwser(offer, RTCOfferAnswerOptions::default())
+                    .await?;
+
+                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) => {
+                let target = SignalTarget::from_i32(trickle.target).unwrap();
+                let ice_candidate = {
+                    let json = serde_json::from_str::<IceCandidateJSON>(&trickle.candidate_init)?;
+                    IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?
+                };
+
+                trace!("received ice_candidate {:?} {:?}", target, ice_candidate);
+
+                if target == SignalTarget::Publisher {
+                    self.publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice_candidate)
+                        .await?;
+                } else {
+                    self.subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice_candidate)
+                        .await?;
+                }
+            }
+            _ => {}
+        }
+
+        Ok(())
+    }
+
+    async fn on_rtc_event(&self, event: RTCEvent) -> EngineResult<()> {
+        match event {
+            RTCEvent::IceCandidate {
+                ice_candidate,
+                target,
+            } => {
+                self.signal_client
+                    .send(signal_request::Message::Trickle(TrickleRequest {
+                        candidate_init: serde_json::to_string(&IceCandidateJSON {
+                            sdpMid: ice_candidate.sdp_mid(),
+                            sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                            candidate: ice_candidate.candidate(),
+                        })?,
+                        target: target as i32,
+                    }))
+                    .await;
+            }
+            RTCEvent::ConnectionChange { state, target } => {
+                trace!("connection change, {:?} {:?}", state, target);
+                let is_primary = self.info.join_response.subscriber_primary
+                    && target == SignalTarget::Subscriber;
+
+                if is_primary && state == PeerConnectionState::Connected {
+                    let old_state = self
+                        .pc_state
+                        .swap(PCState::Connected as u8, Ordering::SeqCst);
+                    if old_state == PCState::New as u8 {
+                        let _ = self.engine_emitter.send(EngineEvent::Connected).await;
+                    }
+                } else if state == PeerConnectionState::Failed {
+                    self.pc_state
+                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
+
+                    self.on_session_disconnected("pc_state failed");
+                }
+            }
+            RTCEvent::DataChannel { data_channel } => {
+                self.subscriber_dc.lock().push(data_channel);
+            }
+            RTCEvent::Offer { offer } => {
+                // Send the publisher offer to the server
+                self.signal_client
+                    .send(signal_request::Message::Offer(proto::SessionDescription {
+                        r#type: "offer".to_string(),
+                        sdp: offer.to_string(),
+                    }))
+                    .await;
+            }
+            RTCEvent::AddTrack {
+                rtp_receiver,
+                streams,
+            } => {
+                let _ = self
+                    .engine_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(())
+    }
+
+    /// Called when the SignalClient or one of the PeerConnection has lost the connection
+    /// The RTCEngine may try a reconnect.
+    fn on_session_disconnected(&self, reason: &str) {
+        let _ = self.emitter.send(SessionEvent::Close {
+            reason: reason.to_owned(),
+        });
+    }
+
+    async fn close(&self) {
+        self.signal_client.close().await;
+        self.publisher_pc.lock().await.close();
+        self.subscriber_pc.lock().await.close();
+    }
+
+    /// Start publisher negotiation
+    async fn negotiate_publisher(&self) -> EngineResult<()> {
+        self.has_published.store(true, Ordering::Release);
+        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)
+    }
+
+    /// Ensure the Publisher PC is connected, if not, start the negotiation
+    /// This is required when sending data to the server
+    async fn ensure_publisher_connected(&self, kind: data_packet::Kind) -> EngineResult<()> {
+        if !self.info.join_response.subscriber_primary {
+            return Ok(());
+        }
+
+        if !self.publisher_pc.lock().await.is_connected()
+            && self
+                .publisher_pc
+                .lock()
+                .await
+                .peer_connection()
+                .ice_connection_state()
+                != IceConnectionState::IceConnectionChecking
+        {
+            let _ = self.negotiate_publisher().await;
+        }
+
+        let dc = self.data_channel(kind);
+        if dc.state() == DataState::Open {
+            return Ok(());
+        }
+
+        // Wait until the PeerConnection is connected
+        let wait_connected = async {
+            while self.publisher_pc.lock().await.is_connected() && dc.state() == DataState::Open {
+                tokio::task::yield_now().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)
+            }
+        }
+    }
+
+    fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
+        if kind == data_packet::Kind::Reliable {
+            &self.reliable_dc
+        } else {
+            &self.lossy_dc
+        }
+    }
 }

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

@@ -61,23 +61,34 @@ impl Default for SignalOptions {
     }
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug)]
 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(
+    pub async fn connect(
         &self,
         url: &str,
         token: &str,
         options: SignalOptions,
-    ) -> SignalResult<SignalEvents> {
+    ) -> SignalResult<()> {
-        let (emitter, events) = mpsc::channel(8);
+        let stream = SignalStream::connect(url, token, options, self.emitter.clone()).await?;
-        let stream = SignalStream::connect(url, token, options, emitter).await?;
         *self.stream.write() = Some(stream);
-        Ok(events)
+        Ok(())
     }
 
     #[instrument(level = Level::DEBUG)]