Started Room, publisher negotiation, add RTCRuntime dependencies across webrtc instances

crates/livekit-core/src/lib.rs

@@ -3,8 +3,9 @@ pub mod proto {
 }
 
 mod lk_runtime;
+mod signal_client;
 mod pc_transport;
 mod rtc_engine;
-mod signal_client;
+mod local_participant;
 
 pub mod room;

crates/livekit-core/src/lk_runtime.rs

@@ -3,17 +3,21 @@ use log::trace;
 use livekit_webrtc::peer_connection_factory::PeerConnectionFactory;
 use livekit_webrtc::webrtc::RTCRuntime;
 
+/// SAFETY: The order of initialization and deletion is important for LKRuntime.
+/// See the C++ constructors & destructor of these fields
+
 pub struct LKRuntime {
-    pub rtc_runtime: RTCRuntime,
     pub pc_factory: PeerConnectionFactory,
+    pub rtc_runtime: RTCRuntime,
 }
 
 impl LKRuntime {
     pub fn new() -> Self {
         trace!("LKRuntime::new()");
+        let rtc_runtime = RTCRuntime::new();
         Self {
-            rtc_runtime: RTCRuntime::new(),
-            pc_factory: PeerConnectionFactory::new(),
+            pc_factory: PeerConnectionFactory::new(rtc_runtime.clone()),
+            rtc_runtime,
         }
     }
 }

crates/livekit-core/src/local_participant.rs

@@ -0,0 +1,47 @@
+use std::sync::Arc;
+
+use futures_util::TryFutureExt;
+use tokio::sync::Mutex;
+
+use crate::proto::{data_packet, DataPacket, ParticipantInfo, UserPacket};
+use crate::room::RoomError;
+use crate::rtc_engine::RTCEngine;
+
+pub struct LocalParticipant {
+    sid: String,
+    identity: String,
+    name: String,
+
+    engine: Arc<Mutex<RTCEngine>>,
+}
+
+impl LocalParticipant {
+    pub(crate) fn from(info: ParticipantInfo, engine: Arc<Mutex<RTCEngine>>) -> Self {
+        Self {
+            sid: info.sid,
+            identity: info.identity,
+            name: info.name,
+            engine,
+        }
+    }
+
+    pub(crate) fn update(info: ParticipantInfo) {
+        // TODO(theomonnom)
+    }
+
+    // TODO(theomonnom) Add the destinations parameter
+    pub async fn publish_data(&mut self, data: &[u8], kind: data_packet::Kind) -> Result<(), RoomError> {
+        let data = DataPacket {
+            kind: kind as i32,
+            value: Some(data_packet::Value::User(UserPacket {
+                participant_sid: self.sid.clone(),
+                payload: data.to_vec(),
+                destination_sids: vec![], // TODO(theomonnom)
+            })),
+        };
+
+        self.engine.lock().await.publish_data(&data, kind).await.map_err(Into::into)
+    }
+}
+
+

crates/livekit-core/src/pc_transport.rs

@@ -1,20 +1,18 @@
-use std::sync::Arc;
+use std::future::Future;
+use std::pin::Pin;
 use std::time::Duration;
 
 use log::{error, trace};
 
 use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
 use livekit_webrtc::peer_connection::{
-    PeerConnection, RTCOfferAnswerOptions, SignalingState,
+    IceConnectionState, PeerConnection, RTCOfferAnswerOptions, SignalingState,
 };
-use livekit_webrtc::peer_connection_factory::RTCConfiguration;
 use livekit_webrtc::rtc_error::RTCError;
 
-use crate::lk_runtime::LKRuntime;
-
 const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); // TODO(theomonnom)
 
-pub type OnOfferHandler = Box<dyn FnMut(SessionDescription) + Send>;
+pub type OnOfferHandler = Box<dyn (FnMut(SessionDescription) -> Pin<Box<dyn Future<Output=()> + Send + 'static>>) + Send + Sync>;
 
 pub struct PCTransport {
     peer_connection: PeerConnection,
@@ -25,16 +23,20 @@ pub struct PCTransport {
 }
 
 impl PCTransport {
-    pub fn new(lk_runtime: Arc<LKRuntime>, cfg: RTCConfiguration) -> Result<Self, RTCError> {
-        let peer_connection = lk_runtime.pc_factory.create_peer_connection(cfg)?;
-
-        Ok(Self {
+    pub fn new(peer_connection: PeerConnection) -> Self {
+        Self {
             peer_connection,
             pending_candidates: Vec::default(),
             on_offer_handler: None,
             restarting_ice: false,
             renegotiate: false,
-        })
+        }
+    }
+
+    pub fn is_connected(&self) -> bool {
+        self.peer_connection.ice_connection_state() == IceConnectionState::IceConnectionConnected
+            || self.peer_connection.ice_connection_state()
+            == IceConnectionState::IceConnectionCompleted
     }
 
     pub fn peer_connection(&mut self) -> &mut PeerConnection {
@@ -51,7 +53,9 @@ impl PCTransport {
             return Ok(());
         }
 
-        self.peer_connection.add_ice_candidate(ice_candidate).await?;
+        self.peer_connection
+            .add_ice_candidate(ice_candidate)
+            .await?;
         Ok(())
     }
 
@@ -116,7 +120,8 @@ impl PCTransport {
         self.peer_connection
             .set_local_description(offer.clone())
             .await?;
-        self.on_offer_handler.as_mut().unwrap()(offer);
+        self.on_offer_handler.as_mut().unwrap()(offer).await;
         Ok(())
     }
 }
+

crates/livekit-core/src/room.rs

@@ -1 +1,73 @@
-pub struct Room {}
+use std::sync::Arc;
+use std::time::Duration;
+use log::trace;
+
+use thiserror::Error;
+use tokio::sync::Mutex;
+use tokio::time::sleep;
+
+use crate::local_participant::LocalParticipant;
+use crate::proto::data_packet;
+use crate::proto::signal_request::Message::Mute;
+use crate::rtc_engine;
+use crate::rtc_engine::{EngineError, RTCEngine};
+
+#[derive(Error, Debug)]
+pub enum RoomError {
+    #[error("internal RTCEngine failure")]
+    Engine(#[from] EngineError),
+}
+
+pub struct Room {
+    sid: String,
+    name: String,
+    local_participant: LocalParticipant,
+    engine: Arc<Mutex<RTCEngine>>,
+}
+
+pub async fn connect(url: &str, token: &str) -> Result<Room, RoomError> {
+    let engine = rtc_engine::connect(url, token).await?;
+
+    engine.on_data(Box::new(|packet| {
+        trace!("This is a test");
+
+        Box::pin(async move {
+            trace!("Another test");
+        })
+    })).await;
+
+    let join = engine.join_response().await;
+    let engine = Arc::new(Mutex::new(engine));
+    let lp = LocalParticipant::from(join.participant.unwrap(), engine.clone());
+
+    let room_info = join.room.unwrap();
+    Ok(Room {
+        sid: room_info.sid,
+        name: room_info.name,
+        local_participant: lp,
+        engine,
+    })
+}
+
+impl Room {
+    pub fn local_participant(&mut self) -> &mut LocalParticipant {
+        &mut self.local_participant
+    }
+
+    pub fn sid(&self) -> &str {
+        &self.sid
+    }
+
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+}
+
+
+#[tokio::test]
+async fn test_test() {
+    env_logger::init();
+    let mut room = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap();
+    room.local_participant().publish_data(b"This is a test", data_packet::Kind::Reliable).await.unwrap();
+    //sleep(Duration::from_secs(60)).await;
+}

crates/livekit-core/src/rtc_engine.rs

@@ -1,16 +1,21 @@
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::{Arc, Weak};
-use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::time::Duration;
 
 use lazy_static::lazy_static;
 use log::{error, trace};
+use prost::Message;
 use thiserror::Error;
 use tokio::sync::{mpsc, Mutex};
-use tokio::time::sleep;
+use tokio::time;
 
-use livekit_webrtc::data_channel::{DataChannel, DataChannelInit};
+use livekit_webrtc::data_channel::{DataChannel, DataChannelInit, DataSendError, DataState};
 use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription};
-use livekit_webrtc::peer_connection::{PeerConnectionState, RTCOfferAnswerOptions};
+use livekit_webrtc::peer_connection::{
+    IceConnectionState, PeerConnectionState, RTCOfferAnswerOptions,
+};
 use livekit_webrtc::peer_connection_factory::{
     ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
 };
@@ -19,17 +24,35 @@ use livekit_webrtc::rtc_error::RTCError;
 use crate::{proto, signal_client};
 use crate::lk_runtime::LKRuntime;
 use crate::pc_transport::PCTransport;
-use crate::proto::{DataPacket, JoinResponse, signal_request, signal_response, SignalResponse, SignalTarget, TrickleRequest};
+use crate::proto::{
+    data_packet, DataPacket, JoinResponse, signal_request, signal_response, SignalTarget,
+    TrickleRequest, UserPacket,
+};
+use crate::proto::data_packet::Value;
 use crate::signal_client::{SignalClient, SignalError};
+use serde::{Deserialize, Serialize};
 
 const LOSSY_DC_LABEL: &str = "_lossy";
 const RELIABLE_DC_LABEL: &str = "_reliable";
+const MAX_ICE_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
 
 lazy_static! {
     // Share one LKRuntime across all RTCEngine instances
     static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
 }
 
+#[derive(Serialize, Deserialize)]
+struct IceCandidateJSON {
+    sdpMid: String,
+    sdpMLineIndex: i32,
+    candidate: String,
+}
+
+pub struct Packet {
+    pub data: UserPacket,
+    pub kind: data_packet::Kind,
+}
+
 #[derive(Error, Debug)]
 pub enum EngineError {
     #[error("signal failure")]
@@ -40,6 +63,14 @@ pub enum EngineError {
     Parse(#[from] SdpParseError),
     #[error("serde error")]
     Serde(#[from] serde_json::Error),
+    #[error("failed to send data to the datachannel")]
+    Data(#[from] DataSendError),
+    #[error("connection error: {0}")]
+    Connection(String),
+    #[error("decode error")]
+    Decode(#[from] prost::DecodeError),
+    #[error("internal error: {0}")]
+    Internal(String), // Unexpected error
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
@@ -55,11 +86,11 @@ enum PCState {
 pub enum EngineMessage {
     IceCandidate {
         ice_candidate: IceCandidate,
-        publisher: bool
+        publisher: bool,
     },
     ConnectionChange {
         state: PeerConnectionState,
-        primary: bool
+        primary: bool,
     },
     PrimaryDataChannel {
         data_channel: DataChannel,
@@ -70,27 +101,35 @@ pub enum EngineMessage {
     Data {
         data: Vec<u8>,
         binary: bool,
-        reliable: bool
-    }
+        reliable: bool,
+    },
 }
 
+pub type OnDataHandler =
+Box<dyn (FnMut(Packet) -> Pin<Box<dyn Future<Output=()> + Send + 'static>>) + Send + Sync>;
+
 struct EngineInternal {
     publisher_pc: Arc<Mutex<PCTransport>>,
     subscriber_pc: Arc<Mutex<PCTransport>>,
-
     lossy_dc: Arc<Mutex<DataChannel>>,
     reliable_dc: Arc<Mutex<DataChannel>>,
 
     msg_sender: mpsc::Sender<EngineMessage>,
+    join_response: Mutex<JoinResponse>,
+    pc_state: AtomicU8,
+    // PCState
+    has_published: AtomicBool,
 
-    pc_state: AtomicU8, // PCState
+    // Listeners
+    on_data_handler: Arc<Mutex<Option<OnDataHandler>>>,
 }
 
 pub struct RTCEngine {
+    signal_client: Arc<SignalClient>,
+    internal: Arc<EngineInternal>,
+
     #[allow(unused)]
     lk_runtime: Arc<LKRuntime>, // Keep a reference while we're using the RTCEngine
-    signal_client: Arc<SignalClient>,
-    internal: Arc<EngineInternal>
 }
 
 pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
@@ -109,7 +148,11 @@ pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
     if let signal_response::Message::Join(join_response) = signal_client.recv().await? {
         trace!("received join_response: {:?}", join_response);
         let (sender, receiver) = mpsc::channel(8);
-        let internal = Arc::new(RTCEngine::configure(lk_runtime.clone(), sender, join_response.clone())?);
+        let internal = Arc::new(RTCEngine::configure(
+            lk_runtime.clone(),
+            sender,
+            join_response.clone(),
+        )?);
 
         if !join_response.subscriber_primary {
             internal.publisher_pc.lock().await.negotiate().await?;
@@ -135,9 +178,93 @@ pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
 }
 
 impl RTCEngine {
+    /// Send data to other participants in the Room
+    pub async fn publish_data(
+        &mut self,
+        data: &DataPacket,
+        kind: data_packet::Kind,
+    ) -> Result<(), EngineError> {
+        self.ensure_publisher_connected(kind).await?;
 
-    async fn send_data() {
+        self.data_channel(kind)
+            .lock()
+            .await
+            .send(&data.encode_to_vec(), true)
+            .map_err(Into::into)
+    }
 
+    /// Return the last JoinResponse from the server
+    pub async fn join_response(&self) -> JoinResponse {
+        self.internal.join_response.lock().await.clone()
+    }
+
+    pub async fn on_data(&self, f: OnDataHandler) {
+        *self.internal.on_data_handler.lock().await = Some(f);
+    }
+
+    fn data_channel(&self, kind: data_packet::Kind) -> &Arc<Mutex<DataChannel>> {
+        if kind == data_packet::Kind::Reliable {
+            &self.internal.reliable_dc
+        } else {
+            &self.internal.lossy_dc
+        }
+    }
+
+    async fn ensure_publisher_connected(
+        &mut self,
+        kind: data_packet::Kind,
+    ) -> Result<(), EngineError> {
+        if !self.join_response().await.subscriber_primary {
+            return Ok(());
+        }
+
+        {
+            let mut publisher = self.internal.publisher_pc.lock().await;
+            if !publisher.is_connected()
+                && publisher.peer_connection().ice_connection_state()
+                != IceConnectionState::IceConnectionChecking
+            {
+                tokio::spawn({
+                    let rtc_internal = self.internal.clone();
+                    async move {
+                        let _ = Self::negotiate_publisher(rtc_internal).await;
+                    }
+                });
+            }
+        }
+
+        let dc = self.data_channel(kind);
+        {
+            let dc = self.data_channel(kind).lock().await;
+            if dc.state() == DataState::Open {
+                return Ok(());
+            }
+        }
+
+        let res = time::timeout(MAX_ICE_CONNECT_TIMEOUT, {
+            let internal = self.internal.clone();
+
+            async move {
+                let mut interval = time::interval(Duration::from_millis(50));
+
+                loop {
+                    if internal.publisher_pc.lock().await.is_connected() && dc.lock().await.state() == DataState::Open {
+                        break;
+                    }
+
+                    interval.tick().await;
+                }
+            }
+        })
+            .await;
+
+        if res.is_err() {
+            Err(EngineError::Connection(
+                "could not establish publisher connection".to_string(),
+            ))
+        } else {
+            Ok(())
+        }
     }
 
     fn send_request(msg: signal_request::Message, signal_client: Arc<SignalClient>) {
@@ -148,109 +275,229 @@ impl RTCEngine {
         });
     }
 
-    async fn handle_signal(signal: signal_response::Message, signal_client: &Arc<SignalClient>, rtc_internal: &Arc<EngineInternal>) -> Result<(), EngineError> {
+    async fn handle_signal(
+        signal: signal_response::Message,
+        signal_client: &Arc<SignalClient>,
+        rtc_internal: &Arc<EngineInternal>,
+    ) -> Result<(), EngineError> {
         match signal {
             signal_response::Message::Answer(answer) => {
+                trace!("received answer for publisher: {:?}", answer);
                 let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
-                rtc_internal.publisher_pc.lock().await.set_remote_description(sdp).await?;
-            },
+                rtc_internal
+                    .publisher_pc
+                    .lock()
+                    .await
+                    .set_remote_description(sdp)
+                    .await?;
+            }
             signal_response::Message::Offer(offer) => {
                 let sdp = SessionDescription::from(offer.r#type.parse().unwrap(), &offer.sdp)?;
                 let mut subscriber_pc = rtc_internal.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?;
+                let answer = subscriber_pc
+                    .peer_connection()
+                    .create_answer(RTCOfferAnswerOptions::default())
+                    .await?;
+                subscriber_pc
+                    .peer_connection()
+                    .set_local_description(answer.clone())
+                    .await?;
 
-                Self::send_request(signal_request::Message::Answer(proto::SessionDescription {
-                    r#type: "answer".to_string(),
-                    sdp: answer.to_string(),
-                }), signal_client.clone());
-            },
+                Self::send_request(
+                    signal_request::Message::Answer(proto::SessionDescription {
+                        r#type: "answer".to_string(),
+                        sdp: answer.to_string(),
+                    }),
+                    signal_client.clone(),
+                );
+            }
             signal_response::Message::Trickle(trickle) => {
-                let json: serde_json::Value = serde_json::from_str(&trickle.candidate_init)?;
-                let ice = IceCandidate::from(
-                    json["sdpMid"].as_str().unwrap(),
-                    json["sdpMLineIndex"].as_i64().unwrap().try_into().unwrap(),
-                    json["candidate"].as_str().unwrap()
-                )?;
+                let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?;
+                let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
+
+                trace!(
+                    "received ice_candidate: {:?} (publisher: {:?})",
+                    ice,
+                    trickle.target
+                );
 
                 if trickle.target == SignalTarget::Publisher as i32 {
-                    rtc_internal.publisher_pc.lock().await.add_ice_candidate(ice).await?;
+                    rtc_internal
+                        .publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
                 } else {
-                    rtc_internal.subscriber_pc.lock().await.add_ice_candidate(ice).await?;
+                    rtc_internal
+                        .subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
                 }
             }
-            _ => {},
+            _ => {}
         }
 
         Ok(())
     }
 
-    async fn handle_loop(mut receiver: mpsc::Receiver<EngineMessage>, signal_client: Arc<SignalClient>, rtc_internal: Arc<EngineInternal>) {
+    async fn handle_message(
+        msg: EngineMessage,
+        signal_client: &Arc<SignalClient>,
+        rtc_internal: &Arc<EngineInternal>,
+    ) -> Result<(), EngineError> {
+        match msg {
+            EngineMessage::IceCandidate {
+                ice_candidate,
+                publisher,
+            } => {
+                trace!(
+                    "sending ice_candidate: {:?} (publisher: {:?})",
+                    ice_candidate,
+                    publisher
+                );
+
+                let json = serde_json::to_string(&IceCandidateJSON {
+                    sdpMid: ice_candidate.sdp_mid(),
+                    sdpMLineIndex: ice_candidate.sdp_mline_index(),
+                    candidate: ice_candidate.candidate()
+                })?;
+
+                // Send the ice_candidate to the server
+                Self::send_request(
+                    signal_request::Message::Trickle(TrickleRequest {
+                        candidate_init: json,
+                        target: if publisher {
+                            SignalTarget::Publisher
+                        } else {
+                            SignalTarget::Subscriber
+                        } as i32,
+                    }),
+                    signal_client.clone(),
+                );
+            }
+            EngineMessage::ConnectionChange { state, primary } => {
+                if primary && state == PeerConnectionState::Connected {
+                    let old_state = rtc_internal.pc_state.load(Ordering::SeqCst);
+                    rtc_internal
+                        .pc_state
+                        .store(PCState::Connected as u8, Ordering::SeqCst);
+
+                    if old_state == PCState::New as u8 {
+                        // TODO(theomonnom) OnConnected
+                    }
+                } else if state == PeerConnectionState::Failed {
+                    rtc_internal
+                        .pc_state
+                        .store(PCState::Disconnected as u8, Ordering::SeqCst);
+
+                    // TODO(theomonnom) handle Disconnect
+                }
+            }
+            EngineMessage::PrimaryDataChannel { mut data_channel } => {
+                let reliable = data_channel.label() == RELIABLE_DC_LABEL;
+                Self::configure_dc(&mut data_channel, reliable, rtc_internal.msg_sender.clone());
+
+                trace!(
+                    "received and using subscriber datachannel (reliable: {:?})",
+                    reliable
+                );
+                if reliable {
+                    *rtc_internal.reliable_dc.lock().await = data_channel;
+                } else {
+                    *rtc_internal.lossy_dc.lock().await = data_channel;
+                }
+            }
+            EngineMessage::PublisherOffer { offer } => {
+                trace!("received publisher offer: {:?}", offer);
+                // Send the offer to the server
+                Self::send_request(
+                    signal_request::Message::Offer(proto::SessionDescription {
+                        r#type: "offer".to_string(),
+                        sdp: offer.to_string(),
+                    }),
+                    signal_client.clone(),
+                );
+            }
+            EngineMessage::Data {
+                data,
+                binary,
+                reliable: _,
+            } => {
+                if !binary {
+                    return Err(EngineError::Internal(
+                        "text message aren't supported by LiveKit".to_string(),
+                    ));
+                }
+
+                let data = DataPacket::decode(&*data)?;
+                match data.value.unwrap() {
+                    Value::User(user) => {
+                        let mut handler = rtc_internal.on_data_handler.lock().await;
+                        if let Some(f) = &mut *handler {
+                            f(Packet {
+                                data: user,
+                                kind: data_packet::Kind::from_i32(data.kind).unwrap(),
+                            }).await;
+                        }
+                    }
+                    Value::Speaker(_) => {
+                        // TODO(theomonnonm)
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn handle_loop(
+        mut receiver: mpsc::Receiver<EngineMessage>,
+        signal_client: Arc<SignalClient>,
+        rtc_internal: Arc<EngineInternal>,
+    ) {
         loop {
             tokio::select! {
                 Ok(signal) = signal_client.recv() => {
+                    trace!("received signal: {:?}", signal);
                     if let Err(err) = Self::handle_signal(signal, &signal_client, &rtc_internal).await {
                         error!("failed to handle signal: {:?}", err);
                     }
                 },
                 Some(msg) = receiver.recv() => {
-                    match msg {
-                        EngineMessage::IceCandidate { ice_candidate, publisher } => {
-                            trace!("received ice_candidate: {:?} (publisher: {:?})", ice_candidate, publisher);
-                            // Send the ice_candidate to the server
-                            Self::send_request(signal_request::Message::Trickle(TrickleRequest {
-                                candidate_init: ice_candidate.to_string(),
-                                target: if publisher {SignalTarget::Publisher} else {SignalTarget::Subscriber} as i32
-                            }), signal_client.clone());
-                        }
-                        EngineMessage::ConnectionChange { state, primary } => {
-                            if primary && state == PeerConnectionState::Connected {
-                                let old_state = rtc_internal.pc_state.load(Ordering::SeqCst);
-                                rtc_internal.pc_state.store(PCState::Connected as u8, Ordering::SeqCst);
-
-                                if old_state == PCState::New as u8 {
-                                    // TODO(theomonnom) OnConnected
-                                }
-                            } else if state == PeerConnectionState::Failed {
-                                rtc_internal.pc_state.store(PCState::Disconnected as u8, Ordering::SeqCst);
-                                // TODO(theomonnom) handle Disconnect
-                            }
-                        }
-                        EngineMessage::PrimaryDataChannel { mut data_channel } => {
-                            let reliable = data_channel.label() == RELIABLE_DC_LABEL;
-                            Self::configure_dc(&mut data_channel, reliable, rtc_internal.msg_sender.clone());
-
-                            trace!("received and using subscriber datachannel (reliable: {:?})", reliable);
-                            if reliable {
-                                *rtc_internal.reliable_dc.lock().await = data_channel;
-                            } else {
-                                *rtc_internal.lossy_dc.lock().await = data_channel;
-                            }
-                        }
-                        EngineMessage::PublisherOffer { offer } => {
-                            trace!("received publisher offer: {:?}", offer);
-                            // Send the offer to the server
-                            Self::send_request(signal_request::Message::Offer(proto::SessionDescription {
-                                r#type: "offer".to_string(),
-                                sdp: offer.to_string(),
-                            }), signal_client.clone());
-                        }
-                        EngineMessage::Data { data, binary, reliable } => {}
+                    if let Err(err) = Self::handle_message(msg, &signal_client, &rtc_internal).await {
+                        error!("failed to handle engine message: {:?}", err);
                     }
                 }
             }
         }
     }
 
+    async fn negotiate_publisher(rtc_internal: Arc<EngineInternal>) -> Result<(), EngineError> {
+        rtc_internal.has_published.store(true, Ordering::SeqCst);
+        if let Err(err) = rtc_internal.publisher_pc.lock().await.negotiate().await {
+            error!("failed to negotiate the publisher: {:?}", err);
+            Err(EngineError::Rtc(err))
+        } else {
+            Ok(())
+        }
+    }
+
     /// This function is called on connect & on reconnect
     /// It creates the PeerConnections, the DataChannels & the libwebrtc listeners
-    fn configure(lk_runtime: Arc<LKRuntime>, sender: mpsc::Sender<EngineMessage>, join: JoinResponse) -> Result<EngineInternal, EngineError> {
+    fn configure(
+        lk_runtime: Arc<LKRuntime>,
+        sender: mpsc::Sender<EngineMessage>,
+        join: JoinResponse,
+    ) -> Result<EngineInternal, EngineError> {
         let rtc_config = RTCConfiguration {
             ice_servers: {
                 let mut servers = vec![];
-                for is in join.ice_servers {
+                for is in join.ice_servers.clone() {
                     servers.push(ICEServer {
                         urls: is.urls,
                         username: is.username,
@@ -263,15 +510,20 @@ impl RTCEngine {
             ice_transport_type: IceTransportsType::All,
         };
 
-        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)?);
+        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)?);
 
         publisher_pc.peer_connection().on_ice_candidate(Box::new({
             let sender = sender.clone();
             move |ice_candidate| {
                 let _ = sender.blocking_send(EngineMessage::IceCandidate {
                     ice_candidate,
-                    publisher: true
+                    publisher: true,
                 });
             }
         }));
@@ -281,17 +533,21 @@ impl RTCEngine {
             move |ice_candidate| {
                 let _ = sender.blocking_send(EngineMessage::IceCandidate {
                     ice_candidate,
-                    publisher: false
+                    publisher: false,
                 });
             }
         }));
 
-        publisher_pc.on_offer(Box::new({
+        publisher_pc.on_offer({
             let sender = sender.clone();
-            move |offer| {
-                let _ = sender.blocking_send(EngineMessage::PublisherOffer {offer});
-            }
-        }));
+            Box::new(move |offer| {
+                let sender = sender.clone();
+
+                Box::pin(async move {
+                    let _ = sender.send(EngineMessage::PublisherOffer { offer }).await;
+                })
+            })
+        });
 
         let mut primary_pc = &mut publisher_pc;
         let mut secondary_pc = &mut subscriber_pc;
@@ -299,25 +555,24 @@ impl RTCEngine {
             primary_pc = &mut subscriber_pc;
             secondary_pc = &mut publisher_pc;
 
-            primary_pc.peer_connection().on_data_channel(Box::new({{
+            primary_pc.peer_connection().on_data_channel(Box::new({
                 let sender = sender.clone();
                 move |data_channel| {
-                    let _ = sender.blocking_send(EngineMessage::PrimaryDataChannel {data_channel});
-                }
-            }}));
-        }
-
-        primary_pc
-            .peer_connection()
-            .on_connection_change(Box::new({
-                let sender = sender.clone();
-                move |state| {
-                    let _ = sender.blocking_send(EngineMessage::ConnectionChange {
-                        state,
-                        primary: true
-                    });
+                    let _ =
+                        sender.blocking_send(EngineMessage::PrimaryDataChannel { data_channel });
                 }
             }));
+        }
+
+        primary_pc.peer_connection().on_connection_change(Box::new({
+            let sender = sender.clone();
+            move |state| {
+                let _ = sender.blocking_send(EngineMessage::ConnectionChange {
+                    state,
+                    primary: true,
+                });
+            }
+        }));
 
         secondary_pc
             .peer_connection()
@@ -326,7 +581,7 @@ impl RTCEngine {
                 move |state| {
                     let _ = sender.blocking_send(EngineMessage::ConnectionChange {
                         state,
-                        primary: false
+                        primary: false,
                     });
                 }
             }));
@@ -358,27 +613,26 @@ impl RTCEngine {
             subscriber_pc: Arc::new(Mutex::new(subscriber_pc)),
             lossy_dc: Arc::new(Mutex::new(lossy_dc)),
             reliable_dc: Arc::new(Mutex::new(reliable_dc)),
+            msg_sender: sender,
+            join_response: Mutex::new(join),
             pc_state: AtomicU8::new(PCState::New as u8),
-            msg_sender: sender
+            has_published: AtomicBool::new(false),
+            on_data_handler: Default::default(),
         })
     }
 
     /// Map the libwebrtc listeners to a mpsc channel
-    fn configure_dc(data_channel: &mut DataChannel, reliable: bool, sender: mpsc::Sender<EngineMessage>) {
+    fn configure_dc(
+        data_channel: &mut DataChannel,
+        reliable: bool,
+        sender: mpsc::Sender<EngineMessage>,
+    ) {
         data_channel.on_message(Box::new(move |data, binary| {
             let _ = sender.blocking_send(EngineMessage::Data {
                 data: data.to_vec(),
                 reliable,
-                binary
+                binary,
             });
         }));
     }
 }
-
-#[tokio::test]
-async fn test_test() {
-    env_logger::init();
-    let engine = connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NzEyMzk4NjAsImlzcyI6IkFQSXpLYkFTaUNWYWtnSiIsIm5hbWUiOiJ0ZXN0IiwibmJmIjoxNjY0MDM5ODYwLCJzdWIiOiJ0ZXN0IiwidmlkZW8iOnsicm9vbUFkbWluIjp0cnVlLCJyb29tQ3JlYXRlIjp0cnVlLCJyb29tSm9pbiI6dHJ1ZX19.0Bee2jI2cSZveAbZ8MLc-ADoMYQ4l8IRxcAxpXAS6a8").await.unwrap();
-
-    sleep(Duration::from_secs(60)).await;
-}