Tiny Room prototype

Minimalist room poc

Cargo.lock

@@ -475,6 +475,7 @@ dependencies = [
  "prost 0.11.0",
  "prost-build",
  "prost-types 0.11.1",
+ "serde",
  "serde_json",
  "thiserror",
  "tokio",
@@ -912,6 +913,20 @@ name = "serde"
 version = "1.0.145"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.145"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "81fa1584d3d1bcacd84c277a0dfe21f5b0f6accf4a23d04d4c6d61f1af522b4c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
 
 [[package]]
 name = "serde_json"

crates/livekit-core/Cargo.toml

@@ -4,6 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+serde = { version = "1", features = ["derive"] }
 serde_json = "1.0"
 log = "0.4"
 tokio-tungstenite = { version = "0.17.2", features = ["native-tls"] }

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(rtc_runtime.clone()),
-            pc_factory: PeerConnectionFactory::new(),
+            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> {
+    pub fn new(peer_connection: PeerConnection) -> Self {
-        let peer_connection = lk_runtime.pc_factory.create_peer_connection(cfg)?;
+        Self {
-
-        Ok(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,17 +1,21 @@
-use std::sync::{Arc, Mutex, Weak};
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::{Arc, Weak};
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::time::Duration;
 
 use lazy_static::lazy_static;
 use log::{error, trace};
-use prost::Message as ProstMessage;
+use prost::Message;
 use thiserror::Error;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, Mutex};
-use tokio::time::sleep;
+use tokio::time;
-use tokio_tungstenite::tungstenite::protocol::frame::coding::Data;
 
-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,
 };
@@ -21,12 +25,33 @@ use crate::{proto, signal_client};
 use crate::lk_runtime::LKRuntime;
 use crate::pc_transport::PCTransport;
 use crate::proto::{
-    DataPacket, JoinResponse, signal_request, signal_response, SignalTarget, TrickleRequest,
+    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 {
@@ -38,9 +63,17 @@ 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(PartialEq, Debug, Copy, Clone)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
 enum PCState {
     New,
     Connected,
@@ -49,219 +82,422 @@ enum PCState {
     Closed,
 }
 
-lazy_static! {
+#[derive(Debug)]
-    // Share one LKRuntime across all RTCEngine instances
+pub enum EngineMessage {
-    static ref LK_RUNTIME: Mutex<Weak<LKRuntime>> = Mutex::new(Weak::new());
+    IceCandidate {
+        ice_candidate: IceCandidate,
+        publisher: bool,
+    },
+    ConnectionChange {
+        state: PeerConnectionState,
+        primary: bool,
+    },
+    PrimaryDataChannel {
+        data_channel: DataChannel,
+    },
+    PublisherOffer {
+        offer: SessionDescription,
+    },
+    Data {
+        data: Vec<u8>,
+        binary: bool,
+        reliable: bool,
+    },
 }
 
-enum EngineMessage {}
+pub type OnDataHandler =
+Box<dyn (FnMut(Packet) -> Pin<Box<dyn Future<Output=()> + Send + 'static>>) + Send + Sync>;
 
-struct PeerInternal {
+struct EngineInternal {
-    publisher_pc: PCTransport,
+    publisher_pc: Arc<Mutex<PCTransport>>,
-    subscriber_pc: PCTransport,
+    subscriber_pc: Arc<Mutex<PCTransport>>,
+    lossy_dc: Arc<Mutex<DataChannel>>,
+    reliable_dc: Arc<Mutex<DataChannel>>,
 
-    lossy_dc: DataChannel,
+    msg_sender: mpsc::Sender<EngineMessage>,
-    reliable_dc: DataChannel,
+    join_response: Mutex<JoinResponse>,
+    pc_state: AtomicU8,
+    // PCState
+    has_published: AtomicBool,
 
-    pub_ice_rx: mpsc::Receiver<IceCandidate>,
+    // Listeners
-    sub_ice_rx: mpsc::Receiver<IceCandidate>,
+    on_data_handler: Arc<Mutex<Option<OnDataHandler>>>,
-
-    pub_offer_rx: mpsc::Receiver<SessionDescription>,
-
-    primary_connection_state_rx: mpsc::Receiver<PeerConnectionState>,
-    secondary_connection_state_rx: mpsc::Receiver<PeerConnectionState>,
-
-    lossy_data_rx: mpsc::Receiver<DataPacket>,
-    reliable_data_rx: mpsc::Receiver<DataPacket>,
-
-    sub_dc_rx: mpsc::Receiver<DataChannel>,
-
-    pc_state: PCState,
 }
 
-struct RTCInternal {
+pub struct RTCEngine {
-    #[allow(unused)]
-    lk_runtime: Arc<LKRuntime>,
     signal_client: Arc<SignalClient>,
-    pc_internal: PeerInternal,
+    internal: Arc<EngineInternal>,
+
+    #[allow(unused)]
+    lk_runtime: Arc<LKRuntime>, // Keep a reference while we're using the RTCEngine
 }
 
-impl RTCInternal {
+pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
-    async fn connect(url: &str, token: &str) -> Result<Self, EngineError> {
+    // Acquire an existing/a new LKRuntime
-        let mut lk_runtime = None;
+    let mut lk_runtime_ref = LK_RUNTIME.lock().await;
-        {
+    let mut lk_runtime = lk_runtime_ref.upgrade();
-            // Acquire an existing/a new LKRuntime
-            let mut lk_runtime_ref = LK_RUNTIME.lock().unwrap();
-            lk_runtime = lk_runtime_ref.upgrade();
 
-            if lk_runtime.is_none() {
+    if lk_runtime.is_none() {
-                let new_runtime = Arc::new(LKRuntime::new());
+        let new_runtime = Arc::new(LKRuntime::new());
-                *lk_runtime_ref = Arc::downgrade(&new_runtime);
+        *lk_runtime_ref = Arc::downgrade(&new_runtime);
-                lk_runtime = Some(new_runtime);
+        lk_runtime = Some(new_runtime);
-            }
+    }
+    let lk_runtime = lk_runtime.unwrap();
+    let signal_client = Arc::new(signal_client::connect(url, token).await?);
+
+    if let 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(),
+        )?);
+
+        if !join_response.subscriber_primary {
+            internal.publisher_pc.lock().await.negotiate().await?;
         }
-        let lk_runtime = lk_runtime.unwrap();
-        let signal_client = Arc::new(signal_client::connect(url, token).await?);
 
-        trace!("waiting join_response..");
+        tokio::spawn({
-        if let signal_response::Message::Join(join) = signal_client.recv().await? {
+            let signal_client = signal_client.clone();
-            trace!("configuring peer_connections: {:?}", join);
+            let internal = internal.clone();
-            let mut pc_internal = Self::configure(lk_runtime.clone(), join.clone())?;
 
-            if !join.subscriber_primary {
+            async move {
-                pc_internal.publisher_pc.negotiate().await?;
+                RTCEngine::handle_loop(receiver, signal_client, internal).await;
             }
+        });
 
-            Ok(Self {
+        Ok(RTCEngine {
-                lk_runtime,
+            lk_runtime,
-                signal_client,
+            signal_client,
-                pc_internal,
+            internal,
-            })
+        })
+    } else {
+        panic!("the first received message isn't a JoinResponse");
+    }
+}
+
+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?;
+
+        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 {
-            panic!("the first received message isn't a JoinResponse");
+            &self.internal.lossy_dc
         }
     }
 
-    fn request_signal(&mut self, msg: signal_request::Message) {
+    async fn ensure_publisher_connected(
-        tokio::spawn({
+        &mut self,
-            let sc = self.signal_client.clone();
+        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 {
-                if let Err(err) = sc.send(msg).await {
+                let mut interval = time::interval(Duration::from_millis(50));
-                    error!("failed to send signal: {:?}", err);
+
+                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>) {
+        tokio::spawn(async move {
+            if let Err(err) = signal_client.send(msg).await {
+                error!("failed to send signal: {:?}", err);
+            }
         });
     }
 
-    async fn handle_signal(&mut self, signal: signal_response::Message) -> 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)?;
-                self.pc_internal.publisher_pc.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)?;
-                self.pc_internal.subscriber_pc.set_remote_description(sdp).await?;
+                let mut subscriber_pc = rtc_internal.subscriber_pc.lock().await;
-                let answer = self.pc_internal.subscriber_pc.peer_connection().create_answer(RTCOfferAnswerOptions::default()).await?;
-                self.pc_internal.subscriber_pc.peer_connection().set_local_description(answer.clone()).await?;
 
-                self.request_signal(signal_request::Message::Answer(proto::SessionDescription {
+                subscriber_pc.set_remote_description(sdp).await?;
-                    r#type: "answer".to_string(),
+                let answer = subscriber_pc
-                    sdp: answer.to_string(),
+                    .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(),
+                );
+            }
             signal_response::Message::Trickle(trickle) => {
-                let json: serde_json::Value = serde_json::from_str(&trickle.candidate_init)?;
+                let json: IceCandidateJSON = serde_json::from_str(&trickle.candidate_init)?;
-                let ice = IceCandidate::from(
+                let ice = IceCandidate::from(&json.sdpMid, json.sdpMLineIndex, &json.candidate)?;
-                    json["sdpMid"].as_str().unwrap(),
+
-                    json["sdpMLineIndex"].as_i64().unwrap().try_into().unwrap(),
+                trace!(
-                    json["candidate"].as_str().unwrap()
+                    "received ice_candidate: {:?} (publisher: {:?})",
-                )?;
+                    ice,
+                    trickle.target
+                );
 
                 if trickle.target == SignalTarget::Publisher as i32 {
-                    self.pc_internal.publisher_pc.add_ice_candidate(ice).await?;
+                    rtc_internal
+                        .publisher_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
                 } else {
-                    self.pc_internal.subscriber_pc.add_ice_candidate(ice).await?;
+                    rtc_internal
+                        .subscriber_pc
+                        .lock()
+                        .await
+                        .add_ice_candidate(ice)
+                        .await?;
                 }
             }
-            _ => {},
+            _ => {}
         }
 
         Ok(())
     }
 
-    async fn run(&mut self) {
+    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) = self.signal_client.recv() => {
+                Ok(signal) = signal_client.recv() => {
-                    if let Err(err) = self.handle_signal(signal).await {
+                    trace!("received signal: {:?}", signal);
+                    if let Err(err) = Self::handle_signal(signal, &signal_client, &rtc_internal).await {
                         error!("failed to handle signal: {:?}", err);
                     }
                 },
-                Some(ice_candidate) = self.pc_internal.pub_ice_rx.recv() => {
+                Some(msg) = receiver.recv() => {
-                    self.request_signal(signal_request::Message::Trickle(TrickleRequest {
+                    if let Err(err) = Self::handle_message(msg, &signal_client, &rtc_internal).await {
-                        candidate_init: ice_candidate.to_string(),
+                        error!("failed to handle engine message: {:?}", err);
-                        target: SignalTarget::Publisher as i32
-                    }));
-                },
-                Some(ice_candidate) = self.pc_internal.sub_ice_rx.recv() => {
-                    self.request_signal(signal_request::Message::Trickle(TrickleRequest {
-                        candidate_init: ice_candidate.to_string(),
-                        target: SignalTarget::Subscriber as i32
-                    }));
-                },
-                Some(sdp) = self.pc_internal.pub_offer_rx.recv() => {
-                    trace!("received publisher offer: {:?}", sdp);
-                    self.request_signal(signal_request::Message::Offer(proto::SessionDescription {
-                        r#type: "offer".to_string(),
-                        sdp: sdp.to_string(),
-                    }));
-                },
-                Some(state) = self.pc_internal.primary_connection_state_rx.recv() => {
-                    if state == PeerConnectionState::Connected {
-                        let old_state = self.pc_internal.pc_state;
-                        self.pc_internal.pc_state = PCState::Connected;
-
-                        if old_state == PCState::New {
-                            // TODO(theomonnom) OnConnected
-                        }
-                    } else if state == PeerConnectionState::Failed {
-                        self.pc_internal.pc_state = PCState::Disconnected;
-                        // TODO(theomonnom) Handle Disconnect
-                    }
-                },
-                Some(state) = self.pc_internal.secondary_connection_state_rx.recv() => {
-                    if state == PeerConnectionState::Failed {
-                        self.pc_internal.pc_state = PCState::Disconnected;
-                        // TODO(theomonnom) Handle Disconnect
-                    }
-                },
-                Some(data) = self.pc_internal.lossy_data_rx.recv() => {
-
-                },
-                Some(data) = self.pc_internal.reliable_data_rx.recv() => {
-
-                },
-                Some(mut dc) = self.pc_internal.sub_dc_rx.recv() => {
-                    // Subscriber DataChannels
-                    // Only received when the subscriber_primary is enabled
-                    trace!("using subscriber data channels");
-
-                    let (data_tx, data_rx) = mpsc::channel(8);
-                    Self::configure_dc(&mut dc, data_tx);
-
-                    if dc.label() == RELIABLE_DC_LABEL {
-                        self.pc_internal.reliable_dc = dc;
-                        self.pc_internal.reliable_data_rx = data_rx;
-                    } else {
-                        self.pc_internal.lossy_dc = dc;
-                        self.pc_internal.lossy_data_rx = data_rx;
                     }
                 }
             }
         }
     }
 
-    fn configure_dc(data_channel: &mut DataChannel, data_tx: mpsc::Sender<DataPacket>) {
+    async fn negotiate_publisher(rtc_internal: Arc<EngineInternal>) -> Result<(), EngineError> {
-        let label = data_channel.label();
+        rtc_internal.has_published.store(true, Ordering::SeqCst);
-        data_channel.on_message(Box::new(move |data, _| {
+        if let Err(err) = rtc_internal.publisher_pc.lock().await.negotiate().await {
-            if let Ok(data) = DataPacket::decode(data) {
+            error!("failed to negotiate the publisher: {:?}", err);
-                let _ = data_tx.blocking_send(data);
+            Err(EngineError::Rtc(err))
-            } else {
+        } else {
-                trace!("{} - failed to decode DataPacket", label);
+            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<PeerInternal, EngineError> {
+    ) -> Result<EngineInternal, EngineError> {
-        let cfg = RTCConfiguration {
+        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,
@@ -274,59 +510,80 @@ impl RTCInternal {
             ice_transport_type: IceTransportsType::All,
         };
 
-        // Create the PeerConnections
+        let mut publisher_pc = PCTransport::new(
-        let mut publisher_pc = PCTransport::new(lk_runtime.clone(), cfg.clone())?;
+            lk_runtime
-        let mut subscriber_pc = PCTransport::new(lk_runtime, cfg)?;
+                .pc_factory
+                .create_peer_connection(rtc_config.clone())?,
+        );
+        let mut subscriber_pc =
+            PCTransport::new(lk_runtime.pc_factory.create_peer_connection(rtc_config)?);
 
-        let (pub_ice_tx, pub_ice_rx) = mpsc::channel(8);
+        publisher_pc.peer_connection().on_ice_candidate(Box::new({
-        let (sub_ice_tx, sub_ice_rx) = mpsc::channel(8);
+            let sender = sender.clone();
-        let (pub_offer_tx, pub_offer_rx) = mpsc::channel(8);
+            move |ice_candidate| {
-        let (primary_connection_state_tx, primary_connection_state_rx) = mpsc::channel(8);
+                let _ = sender.blocking_send(EngineMessage::IceCandidate {
-        let (secondary_connection_state_tx, secondary_connection_state_rx) = mpsc::channel(8);
+                    ice_candidate,
-        let (lossy_data_tx, lossy_data_rx) = mpsc::channel(8);
+                    publisher: true,
-        let (reliable_data_tx, reliable_data_rx) = mpsc::channel(8);
+                });
-        let (sub_dc_tx, sub_dc_rx) = mpsc::channel(8);
+            }
-
-        publisher_pc
-            .peer_connection()
-            .on_ice_candidate(Box::new(move |ice_candidate| {
-                trace!("publisher - on_ice_candidate: {:?}", ice_candidate);
-                let _ = pub_ice_tx.blocking_send(ice_candidate);
-            }));
-
-        subscriber_pc
-            .peer_connection()
-            .on_ice_candidate(Box::new(move |ice_candidate| {
-                trace!("subscriber - on_ice_candidate: {:?}", ice_candidate);
-                let _ = sub_ice_tx.blocking_send(ice_candidate);
-            }));
-
-        publisher_pc.on_offer(Box::new(move |offer| {
-            trace!("publisher - on_offer: {:?}", offer);
-            let _ = pub_offer_tx.blocking_send(offer); // TODO(theomonnom) Don't use blocking_send here
         }));
 
+        subscriber_pc.peer_connection().on_ice_candidate(Box::new({
+            let sender = sender.clone();
+            move |ice_candidate| {
+                let _ = sender.blocking_send(EngineMessage::IceCandidate {
+                    ice_candidate,
+                    publisher: false,
+                });
+            }
+        }));
+
+        publisher_pc.on_offer({
+            let sender = sender.clone();
+            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;
         if join.subscriber_primary {
             primary_pc = &mut subscriber_pc;
             secondary_pc = &mut publisher_pc;
 
-            primary_pc.peer_connection().on_data_channel(Box::new(move |dc| {
+            primary_pc.peer_connection().on_data_channel(Box::new({
-                let _ = sub_dc_tx.blocking_send(dc);
+                let sender = sender.clone();
+                move |data_channel| {
+                    let _ =
+                        sender.blocking_send(EngineMessage::PrimaryDataChannel { data_channel });
+                }
             }));
         }
 
-        primary_pc
+        primary_pc.peer_connection().on_connection_change(Box::new({
-            .peer_connection()
+            let sender = sender.clone();
-            .on_connection_change(Box::new(move |state| {
+            move |state| {
-                let _ = primary_connection_state_tx.blocking_send(state);
+                let _ = sender.blocking_send(EngineMessage::ConnectionChange {
-            }));
+                    state,
+                    primary: true,
+                });
+            }
+        }));
 
         secondary_pc
             .peer_connection()
-            .on_connection_change(Box::new(move |state| {
+            .on_connection_change(Box::new({
-                let _ = secondary_connection_state_tx.blocking_send(state);
+                let sender = sender.clone();
+                move |state| {
+                    let _ = sender.blocking_send(EngineMessage::ConnectionChange {
+                        state,
+                        primary: false,
+                    });
+                }
             }));
 
         // Note that when subscriber_primary feature is enabled,
@@ -348,72 +605,34 @@ impl RTCInternal {
             },
         )?;
 
-        Self::configure_dc(&mut lossy_dc, lossy_data_tx);
+        Self::configure_dc(&mut lossy_dc, true, sender.clone());
-        Self::configure_dc(&mut reliable_dc, reliable_data_tx);
+        Self::configure_dc(&mut reliable_dc, false, sender.clone());
 
-        Ok(PeerInternal {
+        Ok(EngineInternal {
-            publisher_pc,
+            publisher_pc: Arc::new(Mutex::new(publisher_pc)),
-            subscriber_pc,
+            subscriber_pc: Arc::new(Mutex::new(subscriber_pc)),
-            lossy_dc,
+            lossy_dc: Arc::new(Mutex::new(lossy_dc)),
-            reliable_dc,
+            reliable_dc: Arc::new(Mutex::new(reliable_dc)),
-            pub_ice_rx,
+            msg_sender: sender,
-            sub_ice_rx,
+            join_response: Mutex::new(join),
-            pub_offer_rx,
+            pc_state: AtomicU8::new(PCState::New as u8),
-            primary_connection_state_rx,
+            has_published: AtomicBool::new(false),
-            secondary_connection_state_rx,
+            on_data_handler: Default::default(),
-            lossy_data_rx,
-            reliable_data_rx,
-            sub_dc_rx,
-            pc_state: PCState::New,
         })
     }
-}
 
-pub struct RTCEngine {}
+    /// Map the libwebrtc listeners to a mpsc channel
-
+    fn configure_dc(
-/// Initialize the SignalClient & the PeerConnections
+        data_channel: &mut DataChannel,
-pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
+        reliable: bool,
-    let mut rtc_internal = RTCInternal::connect(url, token).await?;
+        sender: mpsc::Sender<EngineMessage>,
-    tokio::spawn(async move {
+    ) {
-        rtc_internal.run().await
+        data_channel.on_message(Box::new(move |data, binary| {
-    });
+            let _ = sender.blocking_send(EngineMessage::Data {
-
+                data: data.to_vec(),
-    Ok(RTCEngine{})
+                reliable,
-}
+                binary,
-
+            });
-impl RTCEngine {
+        }));
-    async fn rtc_handle() {
-        loop {}
     }
 }
-
-#[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;
-
-}
-
-/*sync fn handle_rtc(mut signal_receiver: broadcast::Receiver<Message>) {
-    loop {
-        let msg = match signal_receiver.recv().await {
-            Ok(msg) => msg,
-            Err(error) => {
-                error!("Failed to receive SignalResponse: {:?}", error);
-                continue;
-            }
-        };
-
-        match msg {
-            Message::Join(join) => {}
-            Message::Trickle(trickle) => {}
-            Message::Answer(answer) => {}
-            Message::Offer(offer) => {}
-            _ => {}
-        }
-    }
-}*/

crates/livekit-webrtc/libwebrtc-sys/include/livekit/data_channel.h

@@ -10,6 +10,7 @@
 #include "api/data_channel_interface.h"
 #include "rust/cxx.h"
 #include "rust_types.h"
+#include "webrtc.h"
 
 namespace livekit {
 using NativeDataChannelInit = webrtc::DataChannelInit;
@@ -18,15 +19,18 @@ class NativeDataChannelObserver;
 class DataChannel {
  public:
   explicit DataChannel(
+      std::shared_ptr<RTCRuntime> rtc_runtime,
       rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel);
 
   void register_observer(NativeDataChannelObserver& observer);
   void unregister_observer();
   bool send(const DataBuffer& buffer);
   rust::String label() const;
+  DataState state() const;
   void close();
 
  private:
+  std::shared_ptr<RTCRuntime> rtc_runtime_;
   rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel_;
 };
 

crates/livekit-webrtc/libwebrtc-sys/include/livekit/jsep.h

@@ -18,7 +18,11 @@ class IceCandidate {
  public:
   explicit IceCandidate(
       std::unique_ptr<webrtc::IceCandidateInterface> ice_candidate);
-
+  
+  rust::String sdp_mid() const;
+  int sdp_mline_index() const;
+  rust::String candidate() const; // TODO(theomonnom) Return livekit::Candidate instead of rust::String
+  
   rust::String stringify() const;
   std::unique_ptr<webrtc::IceCandidateInterface> release();
 

crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection.h

@@ -12,6 +12,7 @@
 #include "jsep.h"
 #include "rust/cxx.h"
 #include "rust_types.h"
+#include "webrtc.h"
 
 namespace livekit {
 class NativeAddIceCandidateObserver;
@@ -19,6 +20,7 @@ class NativeAddIceCandidateObserver;
 class PeerConnection {
  public:
   explicit PeerConnection(
+      std::shared_ptr<RTCRuntime> rtc_runtime,
       rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection);
 
   void create_offer(NativeCreateSdpObserverHandle& observer,
@@ -38,9 +40,11 @@ class PeerConnection {
   std::unique_ptr<SessionDescription> remote_description() const;
   SignalingState signaling_state() const;
   IceGatheringState ice_gathering_state() const;
+  IceConnectionState ice_connection_state() const;
   void close();
 
  private:
+  std::shared_ptr<RTCRuntime> rtc_runtime_;
   rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
 };
 
@@ -65,7 +69,7 @@ create_native_add_ice_candidate_observer(
 
 class NativePeerConnectionObserver : public webrtc::PeerConnectionObserver {
  public:
-  explicit NativePeerConnectionObserver(
+  explicit NativePeerConnectionObserver(std::shared_ptr<RTCRuntime> rtc_runtime,
       rust::Box<PeerConnectionObserverWrapper> observer);
 
   void OnSignalingChange(
@@ -126,11 +130,13 @@ class NativePeerConnectionObserver : public webrtc::PeerConnectionObserver {
   void OnInterestingUsage(int usage_pattern) override;
 
  private:
+  std::shared_ptr<RTCRuntime> rtc_runtime_;
   rust::Box<PeerConnectionObserverWrapper> observer_;
 };
 
 std::unique_ptr<NativePeerConnectionObserver>
 create_native_peer_connection_observer(
+    std::shared_ptr<RTCRuntime> rtc_runtime,
     rust::Box<PeerConnectionObserverWrapper> observer);
 }  // namespace livekit
 

crates/livekit-webrtc/libwebrtc-sys/include/livekit/peer_connection_factory.h

@@ -8,6 +8,7 @@
 #include "api/peer_connection_interface.h"
 #include "peer_connection.h"
 #include "rust_types.h"
+#include "webrtc.h"
 
 namespace livekit {
 using NativeRTCConfiguration =
@@ -15,21 +16,19 @@ using NativeRTCConfiguration =
 
 class PeerConnectionFactory {
  public:
-  PeerConnectionFactory();
+  explicit PeerConnectionFactory(std::shared_ptr<RTCRuntime> rtc_runtime);
+  ~PeerConnectionFactory();
 
   std::unique_ptr<PeerConnection> create_peer_connection(
       std::unique_ptr<NativeRTCConfiguration> config,
       NativePeerConnectionObserver& observer) const;
 
  private:
-  std::unique_ptr<rtc::Thread> network_thread_;
+  std::shared_ptr<RTCRuntime> rtc_runtime_;
-  std::unique_ptr<rtc::Thread> worker_thread_;
-  std::unique_ptr<rtc::Thread> signaling_thread_;
-
   rtc::scoped_refptr<webrtc::PeerConnectionFactoryInterface> peer_factory_;
 };
 
-std::unique_ptr<PeerConnectionFactory> create_peer_connection_factory();
+std::unique_ptr<PeerConnectionFactory> create_peer_connection_factory(std::shared_ptr<RTCRuntime> rtc_runtime);
 std::unique_ptr<NativeRTCConfiguration> create_rtc_configuration(
     RTCConfiguration conf);
 }  // namespace livekit

crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h

@@ -22,6 +22,7 @@ enum class SignalingState;
 enum class IceConnectionState;
 enum class IceGatheringState;
 enum class SdpType;
+enum class DataState;
 struct SdpParseError;
 struct RTCOfferAnswerOptions;
 struct RTCError;

crates/livekit-webrtc/libwebrtc-sys/include/livekit/webrtc.h

@@ -22,13 +22,20 @@ class RTCRuntime {
   RTCRuntime(const RTCRuntime&) = delete;
   RTCRuntime& operator=(const RTCRuntime&) = delete;
 
+  rtc::Thread* network_thread() const;
+  rtc::Thread* worker_thread() const;
+  rtc::Thread* signaling_thread() const;
+
  private:
+  std::unique_ptr<rtc::Thread> network_thread_;
+  std::unique_ptr<rtc::Thread> worker_thread_;
+  std::unique_ptr<rtc::Thread> signaling_thread_;
 #ifdef WEBRTC_WIN
   rtc::WinsockInitializer winsock_;
 #endif
 };
 
-std::unique_ptr<RTCRuntime> create_rtc_runtime();
+std::shared_ptr<RTCRuntime> create_rtc_runtime();
 
 }  // namespace livekit
 

crates/livekit-webrtc/libwebrtc-sys/src/data_channel.cpp

@@ -11,8 +11,9 @@
 namespace livekit {
 
 DataChannel::DataChannel(
+    std::shared_ptr<RTCRuntime> rtc_runtime,
     rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel)
-    : data_channel_(std::move(data_channel)) {}
+    : rtc_runtime_(std::move(rtc_runtime)), data_channel_(std::move(data_channel)) {}
 
 void DataChannel::register_observer(NativeDataChannelObserver& observer) {
   data_channel_->RegisterObserver(&observer);
@@ -31,6 +32,10 @@ rust::String DataChannel::label() const {
   return data_channel_->label();
 }
 
+DataState DataChannel::state() const {
+  return static_cast<DataState>(data_channel_->state());
+}
+
 void DataChannel::close() {
   return data_channel_->Close();
 }

crates/livekit-webrtc/libwebrtc-sys/src/data_channel.rs

@@ -69,6 +69,7 @@ pub mod ffi {
         fn unregister_observer(self: Pin<&mut DataChannel>);
         fn send(self: Pin<&mut DataChannel>, data: &DataBuffer) -> bool;
         fn label(self: &DataChannel) -> String;
+        fn state(self: &DataChannel) -> DataState;
         fn close(self: Pin<&mut DataChannel>);
 
         fn create_data_channel_init(init: DataChannelInit) -> UniquePtr<NativeDataChannelInit>;

crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp

@@ -12,7 +12,7 @@
 
 namespace livekit {
 
-const std::string& serialize_sdp_error(webrtc::SdpParseError error) {
+std::string serialize_sdp_error(webrtc::SdpParseError error) {
   std::stringstream ss;
   ss << std::hex << std::setfill('0');
   ss << std::setw(8) << (uint32_t)error.line.length();
@@ -25,6 +25,18 @@ IceCandidate::IceCandidate(
     std::unique_ptr<webrtc::IceCandidateInterface> ice_candidate)
     : ice_candidate_(std::move(ice_candidate)) {}
 
+rust::String IceCandidate::sdp_mid() const {
+  return ice_candidate_->sdp_mid();
+}
+
+int IceCandidate::sdp_mline_index() const {
+  return ice_candidate_->sdp_mline_index();
+}
+
+rust::String IceCandidate::candidate() const {
+  return stringify();
+}
+
 rust::String IceCandidate::stringify() const {
   std::string str;
   ice_candidate_->ToString(&str);

crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs

@@ -49,6 +49,9 @@ pub mod ffi {
         type NativeSetLocalSdpObserverHandle;
         type NativeSetRemoteSdpObserverHandle;
 
+        fn sdp_mid(self: &IceCandidate) -> String;
+        fn sdp_mline_index(self: &IceCandidate) -> i32;
+        fn candidate(self: &IceCandidate) -> String;
         fn stringify(self: &IceCandidate) -> String;
 
         fn stringify(self: &SessionDescription) -> String;

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.cpp

@@ -24,8 +24,10 @@ toNativeOfferAnswerOptions(const RTCOfferAnswerOptions& options) {
 }
 
 PeerConnection::PeerConnection(
+    std::shared_ptr<RTCRuntime> rtc_runtime,
     rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection)
-    : peer_connection_(std::move(peer_connection)) {}
+    : rtc_runtime_(std::move(rtc_runtime)),
+      peer_connection_(std::move(peer_connection)) {}
 
 void PeerConnection::create_offer(
     NativeCreateSdpObserverHandle& observer_handle,
@@ -65,7 +67,7 @@ std::unique_ptr<DataChannel> PeerConnection::create_data_channel(
     throw std::runtime_error(serialize_error(to_error(result.error())));
   }
 
-  return std::make_unique<DataChannel>(result.value());
+  return std::make_unique<DataChannel>(rtc_runtime_, result.value());
 }
 
 void PeerConnection::add_ice_candidate(
@@ -97,14 +99,19 @@ SignalingState PeerConnection::signaling_state() const {
 }
 
 IceGatheringState PeerConnection::ice_gathering_state() const {
-  return static_cast<IceGatheringState>(peer_connection_->ice_gathering_state());
+  return static_cast<IceGatheringState>(
+      peer_connection_->ice_gathering_state());
+}
+
+IceConnectionState PeerConnection::ice_connection_state() const {
+  return static_cast<IceConnectionState>(
+      peer_connection_->ice_connection_state());
 }
 
 void PeerConnection::close() {
   peer_connection_->Close();
 }
 
-
 // AddIceCandidateObserver
 
 NativeAddIceCandidateObserver::NativeAddIceCandidateObserver(
@@ -124,8 +131,9 @@ create_native_add_ice_candidate_observer(
 // PeerConnectionObserver
 
 NativePeerConnectionObserver::NativePeerConnectionObserver(
+    std::shared_ptr<RTCRuntime> rtc_runtime,
     rust::Box<PeerConnectionObserverWrapper> observer)
-    : observer_(std::move(observer)) {}
+    : rtc_runtime_(std::move(rtc_runtime)), observer_(std::move(observer)) {}
 
 void NativePeerConnectionObserver::OnSignalingChange(
     webrtc::PeerConnectionInterface::SignalingState new_state) {
@@ -144,7 +152,7 @@ void NativePeerConnectionObserver::OnRemoveStream(
 
 void NativePeerConnectionObserver::OnDataChannel(
     rtc::scoped_refptr<webrtc::DataChannelInterface> data_channel) {
-  observer_->on_data_channel(std::make_unique<DataChannel>(data_channel));
+  observer_->on_data_channel(std::make_unique<DataChannel>(rtc_runtime_, data_channel));
 }
 
 void NativePeerConnectionObserver::OnRenegotiationNeeded() {
@@ -255,7 +263,9 @@ void NativePeerConnectionObserver::OnInterestingUsage(int usage_pattern) {
 
 std::unique_ptr<NativePeerConnectionObserver>
 create_native_peer_connection_observer(
+    std::shared_ptr<RTCRuntime> rtc_runtime,
     rust::Box<PeerConnectionObserverWrapper> observer) {
-  return std::make_unique<NativePeerConnectionObserver>(std::move(observer));
+  return std::make_unique<NativePeerConnectionObserver>(rtc_runtime,
+                                                        std::move(observer));
 }
 }  // namespace livekit

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs

@@ -109,6 +109,7 @@ pub mod ffi {
         type NativeSetRemoteSdpObserverHandle = crate::jsep::ffi::NativeSetRemoteSdpObserverHandle;
         type NativeDataChannelInit = crate::data_channel::ffi::NativeDataChannelInit;
         type SessionDescription = crate::jsep::ffi::SessionDescription;
+        type RTCRuntime = crate::webrtc::ffi::RTCRuntime;
 
         type NativeAddIceCandidateObserver;
         type NativePeerConnectionObserver;
@@ -166,9 +167,12 @@ pub mod ffi {
 
         fn ice_gathering_state(self: &PeerConnection) -> IceGatheringState;
 
+        fn ice_connection_state(self: &PeerConnection) -> IceConnectionState;
+
         fn close(self: Pin<&mut PeerConnection>);
 
         fn create_native_peer_connection_observer(
+            rtc_runtime: SharedPtr<RTCRuntime>,
             observer: Box<PeerConnectionObserverWrapper>,
         ) -> UniquePtr<NativePeerConnectionObserver>;
 

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp

@@ -4,6 +4,8 @@
 
 #include "livekit/peer_connection_factory.h"
 
+#include <utility>
+
 #include "api/audio_codecs/builtin_audio_decoder_factory.h"
 #include "api/audio_codecs/builtin_audio_encoder_factory.h"
 #include "api/rtc_event_log/rtc_event_log_factory.h"
@@ -16,24 +18,15 @@
 
 namespace livekit {
 
-PeerConnectionFactory::PeerConnectionFactory() {
+PeerConnectionFactory::PeerConnectionFactory(
-  rtc::LogMessage::LogToDebug(rtc::LS_INFO);
+    std::shared_ptr<RTCRuntime> rtc_runtime)
+    : rtc_runtime_(std::move(rtc_runtime)) {
   RTC_LOG(LS_INFO) << "PeerConnectionFactory::PeerConnectionFactory()";
 
-  network_thread_ = rtc::Thread::CreateWithSocketServer();
-  network_thread_->SetName("network_thread", &network_thread_);
-  network_thread_->Start();
-  worker_thread_ = rtc::Thread::Create();
-  worker_thread_->SetName("worker_thread", &worker_thread_);
-  worker_thread_->Start();
-  signaling_thread_ = rtc::Thread::Create();
-  signaling_thread_->SetName("signaling_thread", &signaling_thread_);
-  signaling_thread_->Start();
-
   webrtc::PeerConnectionFactoryDependencies dependencies;
-  dependencies.network_thread = network_thread_.get();
+  dependencies.network_thread = rtc_runtime_->network_thread();
-  dependencies.worker_thread = worker_thread_.get();
+  dependencies.worker_thread = rtc_runtime_->worker_thread();
-  dependencies.signaling_thread = signaling_thread_.get();
+  dependencies.signaling_thread = rtc_runtime_->signaling_thread();
   dependencies.task_queue_factory = webrtc::CreateDefaultTaskQueueFactory();
   dependencies.event_log_factory = std::make_unique<webrtc::RtcEventLogFactory>(
       dependencies.task_queue_factory.get());
@@ -56,6 +49,10 @@ PeerConnectionFactory::PeerConnectionFactory() {
   }
 }
 
+PeerConnectionFactory::~PeerConnectionFactory() {
+  RTC_LOG(LS_INFO) << "PeerConnectionFactory::~PeerConnectionFactory()";
+}
+
 std::unique_ptr<PeerConnection> PeerConnectionFactory::create_peer_connection(
     std::unique_ptr<webrtc::PeerConnectionInterface::RTCConfiguration> config,
     NativePeerConnectionObserver& observer) const {
@@ -67,11 +64,11 @@ std::unique_ptr<PeerConnection> PeerConnectionFactory::create_peer_connection(
     throw std::runtime_error(serialize_error(to_error(result.error())));
   }
 
-  return std::make_unique<PeerConnection>(result.value());
+  return std::make_unique<PeerConnection>(rtc_runtime_, result.value());
 }
 
-std::unique_ptr<PeerConnectionFactory> create_peer_connection_factory() {
+std::unique_ptr<PeerConnectionFactory> create_peer_connection_factory(std::shared_ptr<RTCRuntime> rtc_runtime) {
-  return std::make_unique<PeerConnectionFactory>();
+  return std::make_unique<PeerConnectionFactory>(std::move(rtc_runtime));
 }
 
 std::unique_ptr<NativeRTCConfiguration> create_rtc_configuration(

crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs

@@ -43,8 +43,9 @@ pub mod ffi {
             crate::peer_connection::ffi::NativePeerConnectionObserver;
         type PeerConnectionFactory;
         type NativeRTCConfiguration;
+        type RTCRuntime = crate::webrtc::ffi::RTCRuntime;
 
-        fn create_peer_connection_factory() -> UniquePtr<PeerConnectionFactory>;
+        fn create_peer_connection_factory(runtime: SharedPtr<RTCRuntime>) -> UniquePtr<PeerConnectionFactory>;
         fn create_rtc_configuration(conf: RTCConfiguration) -> UniquePtr<NativeRTCConfiguration>;
 
         /// SAFETY

crates/livekit-webrtc/libwebrtc-sys/src/webrtc.cpp

@@ -8,8 +8,19 @@
 
 namespace livekit {
 RTCRuntime::RTCRuntime() {
+  rtc::LogMessage::LogToDebug(rtc::LS_INFO);
   RTC_LOG(LS_INFO) << "RTCRuntime()";
   RTC_CHECK(rtc::InitializeSSL()) << "Failed to InitializeSSL()";
+
+  network_thread_ = rtc::Thread::CreateWithSocketServer();
+  network_thread_->SetName("network_thread", &network_thread_);
+  network_thread_->Start();
+  worker_thread_ = rtc::Thread::Create();
+  worker_thread_->SetName("worker_thread", &worker_thread_);
+  worker_thread_->Start();
+  signaling_thread_ = rtc::Thread::Create();
+  signaling_thread_->SetName("signaling_thread", &signaling_thread_);
+  signaling_thread_->Start();
 }
 
 RTCRuntime::~RTCRuntime() {
@@ -17,7 +28,19 @@ RTCRuntime::~RTCRuntime() {
   RTC_CHECK(rtc::CleanupSSL()) << "Failed to CleanupSSL()";
 }
 
-std::unique_ptr<RTCRuntime> create_rtc_runtime() {
+rtc::Thread* RTCRuntime::network_thread() const {
-  return std::make_unique<RTCRuntime>();
+  return network_thread_.get();
+}
+
+rtc::Thread* RTCRuntime::worker_thread() const {
+  return worker_thread_.get();
+}
+
+rtc::Thread* RTCRuntime::signaling_thread() const {
+  return signaling_thread_.get();
+}
+
+std::shared_ptr<RTCRuntime> create_rtc_runtime() {
+  return std::make_shared<RTCRuntime>();
 }
 }  // namespace livekit

crates/livekit-webrtc/libwebrtc-sys/src/webrtc.rs

@@ -7,7 +7,7 @@ pub mod ffi {
 
         type RTCRuntime;
 
-        fn create_rtc_runtime() -> UniquePtr<RTCRuntime>;
+        fn create_rtc_runtime() -> SharedPtr<RTCRuntime>;
     }
 }
 

crates/livekit-webrtc/src/data_channel.rs

@@ -1,11 +1,12 @@
-use std::fmt::{Debug, Formatter};
+use std::error::Error;
+use std::fmt::{Debug, Display, Formatter};
 use std::sync::{Arc, Mutex};
-
 use cxx::UniquePtr;
 use log::trace;
 
 use libwebrtc_sys::data_channel as sys_dc;
-pub use sys_dc::ffi::Priority;
+
+pub use sys_dc::ffi::{Priority, DataState};
 
 pub struct DataChannel {
     cxx_handle: UniquePtr<sys_dc::ffi::DataChannel>,
@@ -21,6 +22,17 @@ impl Debug for DataChannel {
     }
 }
 
+#[derive(Debug)]
+pub struct DataSendError;
+
+impl Display for DataSendError {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "failed to send data to the DataChannel")
+    }
+}
+
+impl Error for DataSendError { }
+
 impl DataChannel {
     pub(crate) fn new(cxx_handle: UniquePtr<sys_dc::ffi::DataChannel>) -> Self {
         let mut observer = Box::new(InternalDataChannelObserver::default());
@@ -44,19 +56,28 @@ impl DataChannel {
         dc
     }
 
-    pub fn send(&mut self, data: &[u8], binary: bool) -> bool {
+    pub fn send(&mut self, data: &[u8], binary: bool) -> Result<(), DataSendError> {
         let buffer = sys_dc::ffi::DataBuffer {
             ptr: data.as_ptr(),
             len: data.len(),
             binary,
         };
-        self.cxx_handle.pin_mut().send(&buffer)
+
+        self.cxx_handle
+            .pin_mut()
+            .send(&buffer)
+            .then_some(())
+            .ok_or(DataSendError {})
     }
 
     pub fn label(&self) -> String {
         self.cxx_handle.label()
     }
 
+    pub fn state(&self) -> DataState {
+        self.cxx_handle.state()
+    }
+
     pub fn close(&mut self) {
         self.cxx_handle.pin_mut().close();
     }
@@ -69,7 +90,7 @@ impl DataChannel {
         *self.observer.on_message_handler.lock().unwrap() = Some(handler);
     }
 
-    pub fn on_buffer(&mut self, handler: OnBufferedAmountChangeHandler) {
+    pub fn on_buffered_amount_change(&mut self, handler: OnBufferedAmountChangeHandler) {
         *self
             .observer
             .on_buffered_amount_change_handler

crates/livekit-webrtc/src/jsep.rs

@@ -26,6 +26,18 @@ impl IceCandidate {
     pub(crate) fn release(self) -> UniquePtr<sys_jsep::ffi::IceCandidate> {
         self.cxx_handle
     }
+
+    pub fn sdp_mid(&self) -> String {
+        self.cxx_handle.sdp_mid()
+    }
+
+    pub fn sdp_mline_index(&self) -> i32 {
+        self.cxx_handle.sdp_mline_index()
+    }
+
+    pub fn candidate(&self) -> String {
+        self.cxx_handle.candidate()
+    }
 }
 
 impl ToString for IceCandidate {

crates/livekit-webrtc/src/peer_connection.rs

@@ -25,6 +25,7 @@ pub struct PeerConnection {
     observer: Box<InternalObserver>,
 
     // Keep alive for C++
+    #[allow(unused)]
     native_observer: UniquePtr<sys_pc::ffi::NativePeerConnectionObserver>,
 }
 
@@ -174,6 +175,10 @@ impl PeerConnection {
         self.cxx_handle.ice_gathering_state()
     }
 
+    pub fn ice_connection_state(&self) -> IceConnectionState {
+        self.cxx_handle.ice_connection_state()
+    }
+
     pub fn close(&mut self) {
         self.cxx_handle.pin_mut().close();
     }

crates/livekit-webrtc/src/peer_connection_factory.rs

@@ -8,15 +8,18 @@ pub use sys_factory::ffi::{
 
 use crate::peer_connection::{InternalObserver, PeerConnection};
 use crate::rtc_error::RTCError;
+use crate::webrtc::RTCRuntime;
 
 pub struct PeerConnectionFactory {
     cxx_handle: UniquePtr<sys_factory::ffi::PeerConnectionFactory>,
+    rtc_runtime: RTCRuntime,
 }
 
 impl PeerConnectionFactory {
-    pub fn new() -> Self {
+    pub fn new(rtc_runtime: RTCRuntime) -> Self {
         Self {
-            cxx_handle: sys_factory::ffi::create_peer_connection_factory(),
+            cxx_handle: sys_factory::ffi::create_peer_connection_factory(rtc_runtime.clone().release()),
+            rtc_runtime,
         }
     }
 
@@ -28,8 +31,8 @@ impl PeerConnectionFactory {
 
         unsafe {
             let mut observer = Box::new(InternalObserver::default());
-            let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer(
+            let mut native_observer = sys_pc::ffi::create_native_peer_connection_observer(self.rtc_runtime.clone().release(),
-                Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)),
+                                                                                          Box::new(sys_pc::PeerConnectionObserverWrapper::new(&mut *observer)),
             );
 
             let res = self

crates/livekit-webrtc/src/webrtc.rs

@@ -1,9 +1,10 @@
-use cxx::UniquePtr;
+use cxx::{SharedPtr};
 
 use libwebrtc_sys::webrtc as sys_rtc;
 
+#[derive(Clone)]
 pub struct RTCRuntime {
-    cxx_handle: UniquePtr<sys_rtc::ffi::RTCRuntime>,
+    cxx_handle: SharedPtr<sys_rtc::ffi::RTCRuntime>,
 }
 
 impl RTCRuntime {
@@ -12,4 +13,8 @@ impl RTCRuntime {
             cxx_handle: sys_rtc::ffi::create_rtc_runtime(),
         }
     }
+
+    pub(crate) fn release(self) -> SharedPtr<sys_rtc::ffi::RTCRuntime> {
+        self.cxx_handle
+    }
 }