refactored RTCEngine

2022-12-20 23:05:31 +01:00
parent 2e3c87f232
commit 1ed925dd38
8 changed files with 339 additions and 501 deletions
@@ -1,372 +0,0 @@
 use parking_lot::{Mutex, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::{Arc, Weak};
 use std::time::Duration;
 use tokio::sync::{mpsc, 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 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,
    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};
 //
 // TODO(theomonnom): Smarter retry intervals
 pub(crate) const RECONNECT_ATTEMPTS: u32 = 10;
 pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300);
 lazy_static! {
    // Share one LKRuntime across all RTCEngine instances
    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
 }
 #[derive(Debug)]
 pub struct EngineInternal {
    lk_runtime: Arc<LKRuntime>,
    session: AsyncRwLock<RTCSession>,
    signal_client: Arc<SignalClient>,
    reconnecting: AtomicBool,
    closed: AtomicBool,
    engine_emitter: EngineEmitter,
 }
 impl EngineInternal {
    #[tracing::instrument]
    pub async fn connect(
        url: &str,
        token: &str,
        options: SignalOptions,
    ) -> EngineResult<(Arc<Self>, EngineEvents)> {
        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);
            }
        }
        let lk_runtime = lk_runtime.unwrap();
        // Configure the PeerConnections/RTCSession
        let (engine_emitter, engine_events) = mpsc::channel(8);
        let session_info = SessionInfo {
            url: url.to_owned(),
            token: token.to_owned(),
            join_response: join_response.clone(),
            options,
        };
        let (rtc_session, rtc_events) = RTCSession::new(lk_runtime.clone(), session_info.clone())?;
        let rtc_session = AsyncRwLock::new(rtc_session);
        let rtc_internal = Arc::new(Self {
            lk_runtime,
            session: rtc_session,
            signal_client,
            reconnecting: AtomicBool::new(false),
            closed: AtomicBool::new(false),
            has_published: AtomicBool::new(false),
            pc_state: AtomicU8::new(PCState::New as u8),
            engine_emitter,
        });
        // Start tasks
        tokio::spawn(rtc_internal.clone().signal_task(signal_events));
        tokio::spawn(rtc_internal.clone().engine_task(rtc_events));
        if !join_response.subscriber_primary {
            rtc_internal.negotiate_publisher().await?;
        }
        Ok((rtc_internal, 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,
            } => {
                trace!("sending ice_candidate ({:?}) - {:?}", target, ice_candidate);
                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
                    .session
                    .read()
                    .await
                    .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.handle_disconnected();
                }
            }
            RTCEvent::DataChannel {
                data_channel,
                target,
            } => {
                if target == SignalTarget::Subscriber {
                    self.session.read().await.use_data_channel(data_channel);
                }
            }
            RTCEvent::Offer { offer, target } => {
                if target == SignalTarget::Publisher {
                    // 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,
                target,
            } => {
                if target == SignalTarget::Subscriber {
                    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(())
    }
    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::Update(update) => {
                let _ = self
                    .engine_emitter
                    .send(EngineEvent::ParticipantUpdate(update))
                    .await;
            }
            _ => {}
        }
        Ok(())
    }
 }
 /// Reconnection logic impl, LiveKit handles reconnection in two ways:
 ///  - If the connection is recoverable, the client performs an ICE Restart [`try_resume_connection()`]
 ///  - Othwerwise, a full reconnect is performed. See [`try_restart_connection()`]
 impl EngineInternal {
    /// Called every time the PeerConnection or the SignalClient is closed
    /// We first try to resume the connection, if it fails, we start a full reconnect.
    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 _ = self.engine_emitter.send(EngineEvent::Restarting).await;
                }
                info!("restarting connection... attempt: {}", i);
                if let Err(err) = self.try_restart_connection().await {
                    error!("restarting connection failed: {}", err);
                } else {
                    return;
                }
            } else {
                if i == 0 {
                    let _ = self.engine_emitter.send(EngineEvent::Resuming).await;
                }
                info!("resuming connection... attempt: {}", i);
                if let Err(err) = self.try_resume_connection().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
    }
    /// Try to recover the connection by doing a full reconnect.
    /// It creates a new RTCSession
    async fn try_restart_connection(self: &Arc<Self>) -> EngineResult<()> {
        Ok(())
    }
    /// Try to recover the connection by doing an ICE restart.
    async fn try_resume_connection(self: &Arc<Self>) -> EngineResult<()> {
        let mut session_info = self.info.lock();
        info.options.sid = self
            .session
            .read()
            .await
            .join_response
            .lock()
            .participant
            .as_ref()
            .unwrap()
            .sid
            .clone();
        self.signal_client.close().await;
        self.signal_client
            .connect(&info.url, &info.token.clone(), info.options.clone())
            .await?;
        self.engine_emitter.send(EngineEvent::SignalResumed).await;
        self.session
            .read()
            .as_ref()
            .unwrap()
            .subscriber_pc
            .lock()
            .await
            .prepare_ice_restart();
        if self
            .session
            .read()
            .as_ref()
            .unwrap()
            .has_published
            .load(Ordering::SeqCst)
        {
            self.session
                .read()
                .as_ref()
                .unwrap()
                .publisher_pc
                .lock()
                .await
                .create_and_send_offer(RTCOfferAnswerOptions {
                    ice_restart: true,
                    ..Default::default()
                })
                .await?;
        }
        self.session
            .read()
            .as_ref()
            .unwrap()
            .wait_pc_connection()
            .await?;
        self.signal_client.flush_queue().await;
        self.engine_emitter.send(EngineEvent::Resumed);
        Ok(())
    }
 }
@@ -1,28 +1,22 @@
-use parking_lot::Mutex;
+use livekit_webrtc::data_channel::DataSendError;
-use std::error;
+use livekit_webrtc::jsep::SdpParseError;
 use livekit_webrtc::media_stream::MediaStream;
 use livekit_webrtc::rtc_error::RTCError;
 use livekit_webrtc::rtp_receiver::RtpReceiver;
 use parking_lot::{Mutex, RwLock};
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::{Arc, Weak};
 use std::time::Duration;
-
+use thiserror::Error;
-use tokio::sync::{mpsc, Mutex as AsyncMutex};
+use tokio::task::JoinHandle;
 use tokio_tungstenite::tungstenite::Error;
 use lazy_static::lazy_static;
 use prost::Message;
-use serde::{Deserialize, Serialize};
+use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex, RwLock as AsyncRwLock};
 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::{
@@ -34,7 +28,8 @@ 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;
+use self::rtc_session::{RTCSession, SessionEvent, SessionEvents};
 mod lk_runtime;
 mod pc_transport;
 mod rtc_events;
@@ -79,94 +74,247 @@ pub enum EngineEvent {
    Restarted,
 }
 //
 // TODO(theomonnom): Smarter retry intervals
 pub(crate) const RECONNECT_ATTEMPTS: u32 = 10;
 pub(crate) const RECONNECT_INTERVAL: Duration = Duration::from_millis(300);
 lazy_static! {
    // Share one LKRuntime across all RTCEngine instances
    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
 }
 /// Represents a running RTCSession with the ability to close the session
 /// and the engine_task
 #[derive(Debug)]
 struct EngineHandle {
    session: RTCSession,
    engine_task: JoinHandle<()>,
    close_sender: oneshot::Sender<()>,
 }
 #[derive(Debug)]
 struct EngineInner {
    running_handle: RwLock<Option<EngineHandle>>,
    reconnecting: AtomicBool,
    opened: AtomicBool,
    engine_emitter: EngineEmitter,
 }
 #[derive(Debug)]
 pub struct RTCEngine {
-    engine_inner: Arc<EngineInternal>,
+    lk_runtime: Arc<LKRuntime>,
    inner: Arc<EngineInner>,
 }
 impl RTCEngine {
-    pub fn new() -> Self {
+    pub fn new() -> (Self, EngineEvents) {
        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::new());
+                let new_runtime = Arc::new(LKRuntime::default());
                *lk_runtime_ref = Arc::downgrade(&new_runtime);
                lk_runtime = Some(new_runtime);
            }
        }
-        let (signal_client, mut signal_events) = SignalClient::new();
+        let (engine_emitter, engine_events) = mpsc::channel(8);
        let inner = Arc::new(EngineInner {
            running_handle: Default::default(),
            reconnecting: Default::default(),
            opened: Default::default(),
            engine_emitter,
        });
-        Self { lk_runtime }
+        (
-    }
+            Self {
-
+                lk_runtime: lk_runtime.unwrap(),
-    #[tracing::instrument(skip(url, token))]
+                inner,
-    pub(crate) async fn connect(
+            },
-        url: &str,
+            engine_events,
-        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?;
        debug!("received JoinResponse: {:?}", join_response);
        let (engine_inner, rtc_events) =
            Self::configure_engine(lk_runtime.clone(), join_response.clone())?;
        let engine_inner = Arc::new(engine_inner);
        let signal_client = Arc::new(signal_client);
        let (emitter, events) = mpsc::channel(8);
        tokio::spawn(Self::signal_task(
            signal_client.clone(),
            engine_inner.clone(),
            signal_events,
            emitter.clone(),
        ));
        tokio::spawn(Self::engine_task(
            signal_client.clone(),
            engine_inner.clone(),
            rtc_events,
            emitter.clone(),
        ));
        if !join_response.subscriber_primary {
            engine_inner.negotiate_publisher().await?;
        }
        let rtc_engine = Self {
            signal_client,
            engine_inner,
            lk_runtime,
        };
        Ok((rtc_engine, events))
    }
    #[tracing::instrument]
    pub async fn connect(
        &self,
        url: &str,
        token: &str,
        options: SignalOptions,
    ) -> EngineResult<()> {
        let (session_emitter, session_events) = mpsc::unbounded_channel();
        let session = RTCSession::connect(
            url,
            token,
            options,
            self.lk_runtime.clone(),
            session_emitter,
        )
        .await?;
        let (close_sender, close_receiver) = oneshot::channel();
        let engine_task = tokio::spawn(
            self.inner
                .clone()
                .engine_task(session_events, close_receiver),
        );
        self.inner.opened.store(true, Ordering::SeqCst);
        *self.inner.running_handle.write() = Some(EngineHandle {
            session,
            engine_task,
            close_sender,
        });
        Ok(())
    }
    #[tracing::instrument]
    pub async fn close(&self) {
        self.inner.opened.store(false, Ordering::SeqCst);
        self.inner.close();
    }
    #[tracing::instrument(skip(data))]
    pub async fn publish_data(
        &self,
        data: &DataPacket,
        kind: data_packet::Kind,
    ) -> Result<(), EngineError> {
-        self.engine_inner.ensure_publisher_connected(kind).await?;
+        self.inner.wait_reconnection().await?;
-        self.engine_inner
+        self.inner
-            .data_channel(kind)
+            .running_handle
-            .lock()
+            .read()
-            .send(&data.encode_to_vec(), true)
+            .as_ref()
-            .map_err(Into::into)
+            .unwrap()
            .session
            .publish_data(data, kind)
            .await?;
        Ok(())
    }
-    pub fn join_response(&self) -> JoinResponse {
+    pub fn join_response(&self) -> Option<JoinResponse> {
-        self.engine_inner.join_response.lock().clone()
+        if let Some(handle) = self.inner.running_handle.read().as_ref() {
            Some(handle.session.info().join_response.clone())
        } else {
            None
        }
    }
 }
-    fn close(&self) {
+impl EngineInner {
-        // TODO
+    async fn engine_task(
        self: Arc<Self>,
        mut session_events: SessionEvents,
        mut close_receiver: oneshot::Receiver<()>,
    ) {
        loop {
            tokio::select! {
                res = session_events.recv() => {
                    if let Some(event) = res {
                        if let Err(err) = self.on_session_event(event).await {
                            error!("failed to handle session event: {:?}", err);
                        }
                    } else {
                        panic!("rtc_sessions has been closed unexpectedly");
                    }
                },
                 _ = &mut close_receiver => {
                    break;
                }
            }
        }
    }
    async fn on_session_event(&self, event: SessionEvent) -> EngineResult<()> {
        Ok(())
    }
    async fn close(&self) {
        if let Some(handle) = self.running_handle.write().take() {
            handle.session.close().await;
            let _ = handle.close_sender.send(());
            handle.engine_task.await;
        }
    }
    async fn wait_reconnection(&self) -> EngineResult<()> {
        if !self.opened.load(Ordering::SeqCst) {
            Err(EngineError::Connection("not opened".to_owned()))?
        }
        while self.reconnecting.load(Ordering::Acquire) {
            tokio::task::yield_now().await;
        }
        if self.running_handle.read().is_none() {
            Err(EngineError::Connection("reconnection failed".to_owned()))?
        }
        Ok(())
    }
    /// Called every time the PeerConnection or the SignalClient is closed
    /// We first try to resume the connection, if it fails, we start a full reconnect.
    async fn handle_disconnected(&self) {
        if !self.opened.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 _ = self.engine_emitter.send(EngineEvent::Restarting).await;
                }
                info!("restarting connection... attempt: {}", i);
                if let Err(err) = self.try_restart_connection().await {
                    error!("restarting connection failed: {}", err);
                } else {
                    let _ = self.engine_emitter.send(EngineEvent::Restarted).await;
                    return;
                }
            } else {
                if i == 0 {
                    let _ = self.engine_emitter.send(EngineEvent::Resuming).await;
                }
                info!("resuming connection... attempt: {}", i);
                if let Err(err) = self.try_resume_connection().await {
                    error!("resuming connection failed: {}", err);
                    if let EngineError::Signal(_) = err {
                        full_reconnect = true;
                    }
                } else {
                    let _ = self.engine_emitter.send(EngineEvent::Resumed).await;
                    return;
                }
            }
            tokio::time::sleep(RECONNECT_INTERVAL).await;
        }
        error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
        self.reconnecting.store(false, Ordering::SeqCst);
        // TODO DISCONNECT
    }
    /// Try to recover the connection by doing a full reconnect.
    /// It creates a new RTCSession
    async fn try_restart_connection(&self) -> EngineResult<()> {
        Ok(())
    }
    /// Try to restart the current session
    async fn try_resume_connection(&self) -> EngineResult<()> {
        Ok(())
    }
 }
@@ -76,14 +76,15 @@ impl PCTransport {
    #[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() {
+        if self.peer_connection.remote_description().is_some() && !self.restarting_ice {
-            self.pending_candidates.push(ice_candidate);
+            self.peer_connection
                .add_ice_candidate(ice_candidate)
                .await?;
            return Ok(());
        }
-        self.peer_connection
+        self.pending_candidates.push(ice_candidate);
            .add_ice_candidate(ice_candidate)
            .await?;
        Ok(())
    }
@@ -54,6 +54,7 @@ pub enum SessionEvent {
    Close {
        reason: String,
    },
    Connected,
 }
 #[repr(u8)]
@@ -81,7 +82,7 @@ pub struct SessionInfo {
    pub join_response: JoinResponse,
 }
-/// Fields shared with engine_task and signal_task
+/// Fields shared with rtc_task and signal_task
 #[derive(Debug)]
 struct SessionInner {
    info: SessionInfo,
@@ -114,7 +115,7 @@ pub struct RTCSession {
    inner: Arc<SessionInner>,
    close_emitter: watch::Sender<bool>, // false = is_running
    signal_task: JoinHandle<()>,
-    engine_task: JoinHandle<()>,
+    rtc_task: JoinHandle<()>,
 }
 impl RTCSession {
@@ -123,7 +124,8 @@ impl RTCSession {
        token: &str,
        options: SignalOptions,
        lk_runtime: Arc<LKRuntime>,
-    ) -> EngineResult<(Self, SessionEvents)> {
+        session_emitter: SessionEmitter,
    ) -> EngineResult<Self> {
        // Connect to the SignalClient
        let (signal_client, mut signal_events) = SignalClient::new();
        let signal_client = Arc::new(signal_client);
@@ -165,6 +167,7 @@ impl RTCSession {
            },
        )?;
        // Forward events received in the Signaling Thread to our rtc channel
        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());
@@ -178,7 +181,6 @@ impl RTCSession {
        };
        // Start tasks
        let (session_emitter, session_events) = mpsc::unbounded_channel();
        let (close_emitter, close_receiver) = watch::channel(false);
        let inner = Arc::new(SessionInner {
@@ -200,50 +202,42 @@ impl RTCSession {
                .clone()
                .signal_task(signal_events, close_receiver.clone()),
        );
-        let engine_task = tokio::spawn(
+        let rtc_task = tokio::spawn(inner.clone().rtc_task(rtc_events, close_receiver.clone()));
            inner
                .clone()
                .engine_task(rtc_events, close_receiver.clone()),
        );
        let session = Self {
            lk_runtime,
-            inner,
+            inner: inner.clone(),
            close_emitter,
            signal_task,
-            engine_task,
+            rtc_task,
        };
-        if !join_response.subscriber_primary {
+        if !inner.info.join_response.subscriber_primary {
            inner.negotiate_publisher().await?;
        }
-        Ok((session, session_events))
+        Ok(session)
    }
    /// Close the PeerConnections and the SignalClient
    pub async fn close(self) {
        // Close the tasks
        self.close_emitter.send(true);
-        self.engine_task.await;
+        self.rtc_task.await;
        self.signal_task.await;
        self.inner.close().await;
    }
-    pub async fn wait_pc_connection(&self) -> EngineResult<()> {
+    pub async fn publish_data(
-        let wait_connected = async move {
+        &self,
-            while self.inner.pc_state.load(Ordering::Acquire) != PCState::Connected as u8 {
+        data: &DataPacket,
-                tokio::task::yield_now().await;
+        kind: data_packet::Kind,
-            }
+    ) -> Result<(), EngineError> {
-        };
+        self.inner.publish_data(data, kind).await
    }
-        tokio::select! {
+    pub async fn wait_pc_connectiom(&self) -> EngineResult<()> {
-            _ = wait_connected => Ok(()),
+        self.inner.wait_pc_connection().await
            _ = sleep(MAX_ICE_CONNECT_TIMEOUT) => {
                let err = EngineError::Connection("wait_pc_connection timed out".to_string());
                Err(err)
            }
        }
    }
 }
@@ -260,13 +254,17 @@ impl RTCSession {
        &self.inner.subscriber_pc
    }
    pub fn signal_client(&self) -> &Arc<SignalClient> {
        &self.inner.signal_client
    }
    pub fn data_channel(&self, kind: data_packet::Kind) -> &DataChannel {
        &self.inner.data_channel(kind)
    }
 }
 impl SessionInner {
-    async fn engine_task(
+    async fn rtc_task(
        self: Arc<Self>,
        mut rtc_events: RTCEvents,
        mut close_receiver: watch::Receiver<bool>,
@@ -405,7 +403,7 @@ impl SessionInner {
                        .pc_state
                        .swap(PCState::Connected as u8, Ordering::SeqCst);
                    if old_state == PCState::New as u8 {
-                        let _ = self.engine_emitter.send(EngineEvent::Connected).await;
+                        let _ = self.emitter.send(SessionEvent::Connected);
                    }
                } else if state == PeerConnectionState::Failed {
                    self.pc_state
@@ -428,15 +426,17 @@ impl SessionInner {
            }
            RTCEvent::AddTrack {
                rtp_receiver,
-                streams,
+                mut streams,
            } => {
-                let _ = self
+                if !streams.is_empty() {
-                    .engine_emitter
+                    let _ = self.emitter.send(SessionEvent::MediaTrack {
-                    .send(EngineEvent::AddTrack {
+                        track: rtp_receiver.track(),
-                        rtp_receiver,
+                        stream: streams.remove(0),
-                        streams,
+                        receiver: rtp_receiver,
-                    })
+                    });
-                    .await;
+                } else {
                    warn!("AddTrack event with no streams");
                }
            }
            RTCEvent::Data { data, binary } => {
                if !binary {
@@ -474,6 +474,68 @@ impl SessionInner {
        self.subscriber_pc.lock().await.close();
    }
    #[tracing::instrument]
    async fn publish_data(
        &self,
        data: &DataPacket,
        kind: data_packet::Kind,
    ) -> Result<(), EngineError> {
        self.ensure_publisher_connected(kind).await?;
        self.data_channel(kind)
            .send(&data.encode_to_vec(), true)
            .map_err(Into::into)
    }
    /// Try to restart the session by doing an ICE Restart (The SignalClient is also restarted)
    /// This reconnection if more seemless than the full reconnection implemented in ['RTCEngine']
    async fn restart_session(&self) -> EngineResult<()> {
        self.signal_client.close().await;
        let mut options = self.info.options.clone();
        options.sid = self.info.join_response.participant.clone().unwrap().sid;
        options.reconnect = true;
        self.signal_client
            .connect(&self.info.url, &self.info.token, options)
            .await?;
        self.subscriber_pc.lock().await.prepare_ice_restart();
        if self.has_published.load(Ordering::Acquire) {
            self.publisher_pc
                .lock()
                .await
                .create_and_send_offer(RTCOfferAnswerOptions {
                    ice_restart: true,
                    ..Default::default()
                })
                .await?;
        }
        self.wait_pc_connection().await?;
        self.signal_client.flush_queue().await;
        Ok(())
    }
    // Wait for PCState to become PCState::Connected
    // Timeout after ['MAX_ICE_CONNECT_TIMEOUT']
    async fn wait_pc_connection(&self) -> EngineResult<()> {
        let wait_connected = async move {
            while self.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)
            }
        }
    }
    /// Start publisher negotiation
    async fn negotiate_publisher(&self) -> EngineResult<()> {
        self.has_published.store(true, Ordering::Release);
@@ -24,10 +24,10 @@ class DataChannel {
  void register_observer(NativeDataChannelObserver& observer);
  void unregister_observer();
-  bool send(const DataBuffer& buffer);
+  bool send(const DataBuffer& buffer) const;
  rust::String label() const;
  DataState state() const;
-  void close();
+  void close() const;
 private:
  std::shared_ptr<RTCRuntime> rtc_runtime_;
@@ -24,7 +24,7 @@ void DataChannel::unregister_observer() {
  data_channel_->UnregisterObserver();
 }
-bool DataChannel::send(const DataBuffer& buffer) {
+bool DataChannel::send(const DataBuffer& buffer) const {
  return data_channel_->Send(webrtc::DataBuffer{
      rtc::CopyOnWriteBuffer(buffer.ptr, buffer.len), buffer.binary});
 }
@@ -37,7 +37,7 @@ DataState DataChannel::state() const {
  return static_cast<DataState>(data_channel_->state());
 }
-void DataChannel::close() {
+void DataChannel::close() const {
  return data_channel_->Close();
 }
@@ -68,10 +68,10 @@ pub mod ffi {
        );
        fn unregister_observer(self: Pin<&mut DataChannel>);
-        fn send(self: Pin<&mut DataChannel>, data: &DataBuffer) -> bool;
+        fn send(self:  &DataChannel, data: &DataBuffer) -> bool;
        fn label(self: &DataChannel) -> String;
        fn state(self: &DataChannel) -> DataState;
-        fn close(self: Pin<&mut DataChannel>);
+        fn close(self: &DataChannel);
        fn create_data_channel_init(init: DataChannelInit) -> UniquePtr<NativeDataChannelInit>;
        fn create_native_data_channel_observer(
@@ -58,7 +58,7 @@ impl DataChannel {
        dc
    }
-    pub fn send(&mut self, data: &[u8], binary: bool) -> Result<(), DataSendError> {
+    pub fn send(&self, data: &[u8], binary: bool) -> Result<(), DataSendError> {
        let buffer = sys_dc::ffi::DataBuffer {
            ptr: data.as_ptr(),
            len: data.len(),
@@ -66,7 +66,6 @@ impl DataChannel {
        };
        self.cxx_handle
            .pin_mut()
            .send(&buffer)
            .then_some(())
            .ok_or(DataSendError {})
@@ -80,8 +79,8 @@ impl DataChannel {
        self.cxx_handle.state()
    }
-    pub fn close(&mut self) {
+    pub fn close(&self) {
-        self.cxx_handle.pin_mut().close();
+        self.cxx_handle.close();
    }
    pub fn on_state_change(&mut self, handler: OnStateChangeHandler) {