subscriber prototype

Cargo.lock

@@ -475,6 +475,7 @@ dependencies = [
  "prost 0.11.0",
  "prost-build",
  "prost-types 0.11.1",
+ "serde_json",
  "thiserror",
  "tokio",
  "tokio-tungstenite",
@@ -855,6 +856,12 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "ryu"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4501abdff3ae82a1c1b477a17252eb69cee9e66eb915c1abaa4f44d873df9f09"
+
 [[package]]
 name = "schannel"
 version = "0.1.20"
@@ -900,6 +907,23 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "serde"
+version = "1.0.145"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "728eb6351430bccb993660dfffc5a72f91ccc1295abaa8ce19b27ebe4f75568b"
+
+[[package]]
+name = "serde_json"
+version = "1.0.85"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44"
+dependencies = [
+ "itoa",
+ "ryu",
+ "serde",
+]
+
 [[package]]
 name = "sha-1"
 version = "0.10.0"

crates/livekit-core/Cargo.toml

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

crates/livekit-core/src/lib.rs

@@ -2,9 +2,9 @@ pub mod proto {
     include!(concat!(env!("OUT_DIR"), "/livekit.rs"));
 }
 
-mod rtc_engine;
-mod signal_client;
 mod lk_runtime;
 mod pc_transport;
+mod rtc_engine;
+mod signal_client;
 
 pub mod room;

crates/livekit-core/src/pc_transport.rs

@@ -4,7 +4,9 @@ use std::time::Duration;
 use log::{error, trace};
 
 use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
-use livekit_webrtc::peer_connection::{PeerConnection, RTCOfferAnswerOptions, SdpError, SignalingState};
+use livekit_webrtc::peer_connection::{
+    PeerConnection, RTCOfferAnswerOptions, SignalingState,
+};
 use livekit_webrtc::peer_connection_factory::RTCConfiguration;
 use livekit_webrtc::rtc_error::RTCError;
 
@@ -12,7 +14,7 @@ use crate::lk_runtime::LKRuntime;
 
 const NEGOTIATION_FREQUENCY: Duration = Duration::from_millis(150); // TODO(theomonnom)
 
-pub type OnOfferHandler = Box<dyn FnMut(SessionDescription)>;
+pub type OnOfferHandler = Box<dyn FnMut(SessionDescription) + Send>;
 
 pub struct PCTransport {
     peer_connection: PeerConnection,
@@ -43,37 +45,48 @@ impl PCTransport {
         self.on_offer_handler = Some(handler);
     }
 
-    pub fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) {
+    pub async fn add_ice_candidate(&mut self, ice_candidate: IceCandidate) -> Result<(), RTCError> {
         if self.peer_connection.remote_description().is_none() {
             self.pending_candidates.push(ice_candidate);
-            return;
+            return Ok(());
         }
 
-        self.peer_connection.add_ice_candidate(ice_candidate);
+        self.peer_connection.add_ice_candidate(ice_candidate).await?;
+        Ok(())
     }
 
-    pub async fn set_remote_description(&mut self, remote_description: SessionDescription) -> Result<(), SdpError> {
+    pub async fn set_remote_description(
-        self.peer_connection.set_remote_description(remote_description).await?;
+        &mut self,
+        remote_description: SessionDescription,
+    ) -> Result<(), RTCError> {
+        self.peer_connection
+            .set_remote_description(remote_description)
+            .await?;
 
         for ic in self.pending_candidates.drain(..) {
-            self.peer_connection.add_ice_candidate(ic);
+            self.peer_connection.add_ice_candidate(ic).await?;
         }
         self.restarting_ice = false;
 
         if self.renegotiate {
             self.renegotiate = false;
-            self.create_and_send_offer(RTCOfferAnswerOptions::default()).await?;
+            self.create_and_send_offer(RTCOfferAnswerOptions::default())
+                .await?;
         }
 
         Ok(())
     }
 
-    pub async fn negotiate(&mut self) -> Result<(), SdpError> {
+    pub async fn negotiate(&mut self) -> Result<(), RTCError> {
         // TODO(theomonnom) Debounce here with NEGOTIATION_FREQUENCY
-        self.create_and_send_offer(RTCOfferAnswerOptions::default()).await
+        self.create_and_send_offer(RTCOfferAnswerOptions::default())
+            .await
     }
 
-    async fn create_and_send_offer(&mut self, options: RTCOfferAnswerOptions) -> Result<(), SdpError> {
+    async fn create_and_send_offer(
+        &mut self,
+        options: RTCOfferAnswerOptions,
+    ) -> Result<(), RTCError> {
         if self.on_offer_handler.is_none() {
             return Ok(());
         }
@@ -86,7 +99,9 @@ impl PCTransport {
         if self.peer_connection.signaling_state() == SignalingState::HaveLocalOffer {
             if options.ice_restart {
                 if let Some(remote_description) = self.peer_connection.remote_description() {
-                    self.peer_connection.set_remote_description(remote_description).await?;
+                    self.peer_connection
+                        .set_remote_description(remote_description)
+                        .await?;
                 } else {
                     error!("trying to ice restart when the pc doesn't have remote description");
                 }
@@ -98,8 +113,10 @@ impl PCTransport {
 
         let offer = self.peer_connection.create_offer(options).await?;
         trace!("created offer {:?}", offer);
-        self.peer_connection.set_local_description(offer.clone()).await?;
+        self.peer_connection
+            .set_local_description(offer.clone())
+            .await?;
         self.on_offer_handler.as_mut().unwrap()(offer);
         Ok(())
     }
-}
+}

crates/livekit-core/src/rtc_engine.rs

@@ -1,24 +1,27 @@
 use std::sync::{Arc, Mutex, Weak};
+use std::time::Duration;
 
 use lazy_static::lazy_static;
 use log::{error, trace};
 use prost::Message as ProstMessage;
 use thiserror::Error;
 use tokio::sync::mpsc;
+use tokio::time::sleep;
 
 use livekit_webrtc::data_channel::{DataChannel, DataChannelInit};
-use livekit_webrtc::jsep::{IceCandidate, SessionDescription};
+use livekit_webrtc::jsep::{IceCandidate, SdpParseError, SessionDescription};
-use livekit_webrtc::peer_connection::PeerConnectionState;
+use livekit_webrtc::peer_connection::{PeerConnectionState, RTCOfferAnswerOptions};
 use livekit_webrtc::peer_connection_factory::{
     ContinualGatheringPolicy, ICEServer, IceTransportsType, RTCConfiguration,
 };
 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, SignalTarget, TrickleRequest};
+use crate::proto::{
-use crate::proto::signal_response::Message;
+    DataPacket, JoinResponse, signal_request, signal_response, SignalTarget, TrickleRequest,
-use crate::signal_client;
+};
 use crate::signal_client::{SignalClient, SignalError};
 
 const LOSSY_DC_LABEL: &str = "_lossy";
@@ -27,9 +30,22 @@ const RELIABLE_DC_LABEL: &str = "_reliable";
 #[derive(Error, Debug)]
 pub enum EngineError {
     #[error("signal failure")]
-    SignalError(#[from] SignalError),
+    Signal(#[from] SignalError),
     #[error("internal webrtc failure")]
-    RTCError(#[from] RTCError),
+    Rtc(#[from] RTCError),
+    #[error("failed to parse sdp")]
+    Parse(#[from] SdpParseError),
+    #[error("serde error")]
+    Serde(#[from] serde_json::Error),
+}
+
+#[derive(PartialEq, Debug, Copy, Clone)]
+enum PCState {
+    New,
+    Connected,
+    Disconnected,
+    Reconnecting,
+    Closed,
 }
 
 lazy_static! {
@@ -56,19 +72,19 @@ struct PeerInternal {
 
     lossy_data_rx: mpsc::Receiver<DataPacket>,
     reliable_data_rx: mpsc::Receiver<DataPacket>,
+
+    pc_state: PCState,
 }
 
 struct RTCInternal {
+    #[allow(unused)]
     lk_runtime: Arc<LKRuntime>,
     signal_client: Arc<SignalClient>,
     pc_internal: PeerInternal,
 }
 
 impl RTCInternal {
-    async fn connect(
+    async fn connect(url: &str, token: &str) -> Result<Self, EngineError> {
-        url: &str,
-        token: &str,
-    ) -> Result<Self, EngineError> {
         let mut lk_runtime = None;
         {
             // Acquire an existing/a new LKRuntime
@@ -84,8 +100,14 @@ impl RTCInternal {
         let lk_runtime = lk_runtime.unwrap();
         let signal_client = Arc::new(signal_client::connect(url, token).await?);
 
-        if let Message::Join(join) = signal_client.recv().await? {
+        trace!("waiting JoinReponse..");
-            let pc_internal = Self::configure(lk_runtime.clone(), join)?;
+        if let signal_response::Message::Join(join) = signal_client.recv().await? {
+            trace!("configuring peerconnections: {:?}", join);
+            let mut pc_internal = Self::configure(lk_runtime.clone(), join.clone())?;
+
+            if !join.subscriber_primary {
+                pc_internal.publisher_pc.negotiate().await?;
+            }
 
             Ok(Self {
                 lk_runtime,
@@ -97,44 +119,100 @@ impl RTCInternal {
         }
     }
 
+    fn request_signal(&mut self, msg: signal_request::Message) {
+        tokio::spawn({
+            let sc = self.signal_client.clone();
+
+            async move {
+                if let Err(err) = sc.send(msg).await {
+                    error!("failed to send signal: {:?}", err);
+                }
+            }
+        });
+    }
+
+    async fn handle_signal(&mut self, signal: signal_response::Message) -> Result<(), EngineError> {
+        match signal {
+            signal_response::Message::Answer(answer) => {
+                let sdp = SessionDescription::from(answer.r#type.parse().unwrap(), &answer.sdp)?;
+                self.pc_internal.publisher_pc.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 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 {
+                    r#type: "answer".to_string(),
+                    sdp: answer.to_string(),
+                }));
+            },
+            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()
+                )?;
+
+                if trickle.target == SignalTarget::Publisher as i32 {
+                    self.pc_internal.publisher_pc.add_ice_candidate(ice).await?;
+                } else {
+                    self.pc_internal.subscriber_pc.add_ice_candidate(ice).await?;
+                }
+            }
+            _ => {},
+        }
+
+        Ok(())
+    }
+
     async fn run(&mut self) {
         loop {
             tokio::select! {
                 Ok(signal) = self.signal_client.recv() => {
-
+                    if let Err(err) = self.handle_signal(signal).await {
+                        error!("failed to handle signal: {:?}", err);
+                    }
                 },
                 Some(ice_candidate) = self.pc_internal.pub_ice_rx.recv() => {
-                    tokio::spawn({
+                    self.request_signal(signal_request::Message::Trickle(TrickleRequest {
-                        let sc = self.signal_client.clone();
+                        candidate_init: ice_candidate.to_string(),
-
+                        target: SignalTarget::Publisher as i32
-                        async move {
+                    }));
-                            let _ = sc.send(signal_request::Message::Trickle(TrickleRequest {
-                                candidate_init: ice_candidate.to_string(),
-                                target: SignalTarget::Publisher as i32
-                            })).await;
-                        }
-                    });
                 },
                 Some(ice_candidate) = self.pc_internal.sub_ice_rx.recv() => {
-                    tokio::spawn({
+                    self.request_signal(signal_request::Message::Trickle(TrickleRequest {
-                        let sc = self.signal_client.clone();
+                        candidate_init: ice_candidate.to_string(),
-
+                        target: SignalTarget::Subscriber as i32
-                        async move {
+                    }));
-                            let _ = sc.send(signal_request::Message::Trickle(TrickleRequest {
-                                candidate_init: ice_candidate.to_string(),
-                                target: SignalTarget::Subscriber as i32
-                            })).await;
-                        }
-                    });
                 },
                 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() => {
 
@@ -146,7 +224,10 @@ impl RTCInternal {
         }
     }
 
-    fn configure(lk_runtime: Arc<LKRuntime>, join: JoinResponse) -> Result<PeerInternal, EngineError> {
+    fn configure(
+        lk_runtime: Arc<LKRuntime>,
+        join: JoinResponse,
+    ) -> Result<PeerInternal, EngineError> {
         let cfg = RTCConfiguration {
             ice_servers: {
                 let mut servers = vec![];
@@ -175,50 +256,62 @@ impl RTCInternal {
         let (lossy_data_tx, lossy_data_rx) = mpsc::channel(8);
         let (reliable_data_tx, reliable_data_rx) = mpsc::channel(8);
 
-        publisher_pc.peer_connection().on_ice_candidate(Box::new(move |ice_candidate| {
+        publisher_pc
-            trace!("publisher - on_ice_candidate: {:?}", ice_candidate);
+            .peer_connection()
-            let _ = pub_ice_tx.blocking_send(ice_candidate);
+            .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| {
+        subscriber_pc
-            trace!("subscriber - on_ice_candidate: {:?}", ice_candidate);
+            .peer_connection()
-            let _ = sub_ice_tx.blocking_send(ice_candidate);
+            .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
         }));
 
-        let mut primary_pc = &publisher_pc;
+        let mut primary_pc = &mut publisher_pc;
-        let mut secondary_pc = &subscriber_pc;
+        let mut secondary_pc = &mut subscriber_pc;
         if join.subscriber_primary {
-            primary_pc = &subscriber_pc;
+            primary_pc = &mut subscriber_pc;
-            secondary_pc = &publisher_pc;
+            secondary_pc = &mut publisher_pc;
         }
 
-        primary_pc.peer_connection().on_connection_change(Box::new(move |state| {
+        primary_pc
-            let _ = primary_connection_state_tx.blocking_send(state);
+            .peer_connection()
-        }));
+            .on_connection_change(Box::new(move |state| {
+                let _ = primary_connection_state_tx.blocking_send(state);
+            }));
 
-        secondary_pc.peer_connection().on_connection_change(Box::new(move |state| {
+        secondary_pc
-            let _ = secondary_connection_state_tx.blocking_send(state);
+            .peer_connection()
-        }));
+            .on_connection_change(Box::new(move |state| {
+                let _ = secondary_connection_state_tx.blocking_send(state);
+            }));
 
-        let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(LOSSY_DC_LABEL, {
+        let mut lossy_dc = publisher_pc.peer_connection().create_data_channel(
-            let mut dc_init = DataChannelInit::default();
+            LOSSY_DC_LABEL,
-            dc_init.ordered = true;
+            DataChannelInit {
-            dc_init.max_retransmits = Some(0);
+                ordered: true,
-            dc_init
+                max_retransmits: Some(0),
-        })?;
+                ..DataChannelInit::default()
+            },
+        )?;
 
-        let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(RELIABLE_DC_LABEL, {
+        let mut reliable_dc = publisher_pc.peer_connection().create_data_channel(
-            let mut dc_init = DataChannelInit::default();
+            RELIABLE_DC_LABEL,
-            dc_init.ordered = true;
+            DataChannelInit {
-            dc_init
+                ordered: true,
-        })?;
+                ..DataChannelInit::default()
+            },
+        )?;
 
-        lossy_dc.on_message(Box::new(|data, binary| {
+        lossy_dc.on_message(Box::new(move |data, _| {
             if let Ok(data) = DataPacket::decode(data) {
                 let _ = lossy_data_tx.blocking_send(data);
             } else {
@@ -226,7 +319,7 @@ impl RTCInternal {
             }
         }));
 
-        reliable_dc.on_message(Box::new(|data, binary| {
+        reliable_dc.on_message(Box::new(move |data, _| {
             if let Ok(data) = DataPacket::decode(data) {
                 let _ = reliable_data_tx.blocking_send(data);
             } else {
@@ -246,6 +339,7 @@ impl RTCInternal {
             secondary_connection_state_rx,
             lossy_data_rx,
             reliable_data_rx,
+            pc_state: PCState::New,
         })
     }
 }
@@ -253,8 +347,14 @@ impl RTCInternal {
 pub struct RTCEngine {}
 
 /// Initialize the SignalClient & the PeerConnections
-//pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
+pub async fn connect(url: &str, token: &str) -> Result<RTCEngine, EngineError> {
-//}
+    let mut rtc_internal = RTCInternal::connect(url, token).await?;
+    tokio::spawn(async move {
+        rtc_internal.run().await
+    });
+
+    Ok(RTCEngine{})
+}
 
 impl RTCEngine {
     async fn rtc_handle() {
@@ -266,7 +366,11 @@ impl RTCEngine {
 async fn test_test() {
     env_logger::init();
 
-    //engine.connect("ws://localhost:7880", "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjE2NjQ1OTY4MDYsImlzcyI6IkFQSUNrSG04M01oZ2hQeCIsIm5hbWUiOiJ1c2VyMSIsIm5iZiI6MTY2MDk5NjgwNiwic3ViIjoidXNlcjEiLCJ2aWRlbyI6eyJyb29tIjoibXktZmlyc3Qtcm9vbSIsInJvb21Kb2luIjp0cnVlfX0.SWU_LETMK6ZmFOf38pYjVhpur0o7jJc6u61h8BH7g20").await.unwrap();
+    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>) {

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

@@ -26,6 +26,8 @@ class IceCandidate {
   std::unique_ptr<webrtc::IceCandidateInterface> ice_candidate_;
 };
 
+std::unique_ptr<IceCandidate> create_ice_candidate(rust::String sdp_mid, int sdp_mline_index, rust::String sdp);
+
 static std::unique_ptr<IceCandidate> _unique_ice_candidate() {
   return nullptr;  // Ignore
 }
@@ -43,6 +45,8 @@ class SessionDescription {
   std::unique_ptr<webrtc::SessionDescriptionInterface> session_description_;
 };
 
+std::unique_ptr<SessionDescription> create_session_description(SdpType type, rust::String sdp);
+
 static std::unique_ptr<SessionDescription> _unique_session_description() {
   return nullptr;  // Ignore
 }

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

@@ -21,6 +21,8 @@ enum class PeerConnectionState;
 enum class SignalingState;
 enum class IceConnectionState;
 enum class IceGatheringState;
+enum class SdpType;
+struct SdpParseError;
 struct RTCOfferAnswerOptions;
 struct RTCError;
 struct DataChannelInit;

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

@@ -4,14 +4,23 @@
 
 #include "livekit/jsep.h"
 
+#include <iomanip>
 #include <memory>
 
 #include "libwebrtc-sys/src/jsep.rs.h"
 #include "livekit/rtc_error.h"
-#include "rtc_base/ref_counted_object.h"
 
 namespace livekit {
 
+const 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();
+  ss << std::dec << std::setw(1) << error.line;
+  ss << std::dec << std::setw(1) << error.description;
+  return ss.str();
+}
+
 IceCandidate::IceCandidate(
     std::unique_ptr<webrtc::IceCandidateInterface> ice_candidate)
     : ice_candidate_(std::move(ice_candidate)) {}
@@ -26,6 +35,20 @@ std::unique_ptr<webrtc::IceCandidateInterface> IceCandidate::release() {
   return std::move(ice_candidate_);
 }
 
+std::unique_ptr<IceCandidate> create_ice_candidate(rust::String sdp_mid,
+                                                   int sdp_mline_index,
+                                                   rust::String sdp) {
+  webrtc::SdpParseError error;
+  auto ice_rtc = webrtc::CreateIceCandidate(sdp_mid.c_str(), sdp_mline_index,
+                                            sdp.c_str(), &error);
+  if (!ice_rtc) {
+    throw std::runtime_error(serialize_sdp_error(error));
+  }
+
+  return std::make_unique<IceCandidate>(
+      std::unique_ptr<webrtc::IceCandidateInterface>(ice_rtc));
+}
+
 SessionDescription::SessionDescription(
     std::unique_ptr<webrtc::SessionDescriptionInterface> session_description)
     : session_description_(std::move(session_description)) {}
@@ -45,6 +68,19 @@ SessionDescription::release() {
   return std::move(session_description_);
 }
 
+std::unique_ptr<SessionDescription> create_session_description(
+    SdpType type,
+    rust::String sdp) {
+  webrtc::SdpParseError error;
+  auto rtc_sdp = webrtc::CreateSessionDescription(
+      static_cast<webrtc::SdpType>(type), sdp.c_str(), &error);
+  if (!rtc_sdp) {
+    throw std::runtime_error(serialize_sdp_error(error));
+  }
+
+  return std::make_unique<SessionDescription>(std::move(rtc_sdp));
+}
+
 // CreateSdpObserver
 
 NativeCreateSdpObserver::NativeCreateSdpObserver(

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

@@ -1,4 +1,6 @@
-use std::fmt::{Debug, Formatter};
+use std::error::Error;
+use std::fmt::{Debug, Display, Formatter};
+use std::str::FromStr;
 
 use cxx::UniquePtr;
 
@@ -6,6 +8,21 @@ use crate::rtc_error::ffi::RTCError;
 
 #[cxx::bridge(namespace = "livekit")]
 pub mod ffi {
+    #[derive(Debug)]
+    #[repr(i32)]
+    pub enum SdpType {
+        Offer,
+        PrAnswer,
+        Answer,
+        Rollback,
+    }
+
+    #[derive(Debug)]
+    pub struct SdpParseError {
+        pub line: String,
+        pub description: String,
+    }
+
     extern "Rust" {
         type CreateSdpObserverWrapper;
         fn on_success(
@@ -47,11 +64,22 @@ pub mod ffi {
             observer: Box<SetRemoteSdpObserverWrapper>,
         ) -> UniquePtr<NativeSetRemoteSdpObserverHandle>;
 
+        fn create_ice_candidate(sdp_mid: String, sdp_mline_index: i32, sdp: String) -> Result<UniquePtr<IceCandidate>>;
+        fn create_session_description(sdp_type: SdpType, sdp: String) -> Result<UniquePtr<SessionDescription>>;
+
         fn _unique_ice_candidate() -> UniquePtr<IceCandidate>; // Ignore
         fn _unique_session_description() -> UniquePtr<SessionDescription>; // Ignore
     }
 }
 
+impl Error for ffi::SdpParseError {}
+
+impl Display for ffi::SdpParseError {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "SdpParseError occurred {}: {}", self.line, self.description)
+    }
+}
+
 impl Debug for ffi::SessionDescription {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         write!(f, "{}", self.stringify())
@@ -59,6 +87,7 @@ impl Debug for ffi::SessionDescription {
 }
 
 unsafe impl Send for ffi::SessionDescription {}
+unsafe impl Sync for ffi::SessionDescription {}
 
 impl Debug for ffi::IceCandidate {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
@@ -67,9 +96,38 @@ impl Debug for ffi::IceCandidate {
 }
 
 unsafe impl Send for ffi::IceCandidate {}
-
 unsafe impl Sync for ffi::IceCandidate {}
 
+impl ffi::SdpParseError {
+    /// # Safety
+    /// The value must be correctly encoded
+    pub unsafe fn from(value: &str) -> Self {
+        // Parse the hex encoded error from c++
+        let line_length = u32::from_str_radix(&value[0..8], 16).unwrap() as usize + 8;
+        let line = String::from(&value[8..line_length]);
+        let description = String::from(&value[line_length..]);
+
+        Self {
+            line,
+            description,
+        }
+    }
+}
+
+impl FromStr for ffi::SdpType {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "offer" => Ok(ffi::SdpType::Offer),
+            "pranswer" => Ok(ffi::SdpType::PrAnswer),
+            "answer" => Ok(ffi::SdpType::Answer),
+            "rollback" => Ok(ffi::SdpType::Rollback),
+            _ => Err(()),
+        }
+    }
+}
+
 // CreateSdpObserver
 
 pub trait CreateSdpObserver: Send {
@@ -134,3 +192,37 @@ impl SetRemoteSdpObserverWrapper {
         self.observer.on_set_remote_description_complete(error);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use log::info;
+
+    use crate::jsep::ffi;
+
+    #[test]
+    fn throw_error() {
+        let sdp_string = "v=0
+o=- 6549709950142776241 2 IN IP4 127.0.0.1
+s=-
+t=0 0
+======================== ERROR HERE
+a=group:BUNDLE 0
+a=extmap-allow-mixed
+a=msid-semantic: WMS
+m=application 9 UDP/DTLS/SCTP webrtc-datachannel
+c=IN IP4 0.0.0.0
+a=ice-ufrag:Tw7h
+a=ice-pwd:6XOVUD6HpcB4c1M8EB8jXJE9
+a=ice-options:trickle
+a=fingerprint:sha-256 4F:EC:23:59:5D:A5:E6:3E:3E:5D:8A:09:B6:FA:04:AA:19:99:49:67:BD:65:93:06:BB:EE:AC:D5:21:0F:57:D6
+a=setup:actpass
+a=mid:0
+a=sctp-port:5000
+a=max-message-size:262144
+";
+
+        let sdp = ffi::create_session_description(ffi::SdpType::Offer, sdp_string.to_string());
+        let err = unsafe { ffi::SdpParseError::from(sdp.err().unwrap().what()) };
+        info!("parse err: {:?}", err)
+    }
+}

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

@@ -253,11 +253,23 @@ pub mod ffi {
 }
 
 // https://webrtc.github.io/webrtc-org/native-code/native-apis/
+unsafe impl Send for ffi::PeerConnection {}
 unsafe impl Sync for ffi::PeerConnection {}
 
-unsafe impl Send for ffi::PeerConnection {}
-
 unsafe impl Send for ffi::NativePeerConnectionObserver {}
+unsafe impl Sync for ffi::NativePeerConnectionObserver {}
+
+unsafe impl Sync for ffi::NativeAddIceCandidateObserver {}
+unsafe impl Send for ffi::NativeAddIceCandidateObserver {}
+
+unsafe impl Sync for ffi::NativeSetRemoteSdpObserverHandle {}
+unsafe impl Send for ffi::NativeSetRemoteSdpObserverHandle {}
+
+unsafe impl Sync for ffi::NativeSetLocalSdpObserverHandle {}
+unsafe impl Send for ffi::NativeSetLocalSdpObserverHandle {}
+
+unsafe impl Sync for ffi::NativeCreateSdpObserverHandle {}
+unsafe impl Send for ffi::NativeCreateSdpObserverHandle {}
 
 impl Default for ffi::RTCOfferAnswerOptions {
     /*
@@ -280,17 +292,21 @@ impl Default for ffi::RTCOfferAnswerOptions {
     }
 }
 
+pub trait AddIceCandidateObserver: Send {
+    fn on_complete(&self, error: RTCError);
+}
+
 pub struct AddIceCandidateObserverWrapper {
-    observer: Box<dyn Fn(RTCError) + Send>,
+    observer: Box<dyn AddIceCandidateObserver>,
 }
 
 impl AddIceCandidateObserverWrapper {
-    pub fn new(observer: Box<dyn Fn(RTCError) + Send>) -> Self {
+    pub fn new(observer: Box<dyn AddIceCandidateObserver>) -> Self {
         Self { observer }
     }
 
     fn on_complete(&self, error: RTCError) {
-        (self.observer)(error);
+        self.observer.on_complete(error);
     }
 }
 

crates/livekit-webrtc/src/jsep.rs

@@ -1,7 +1,8 @@
 use cxx::UniquePtr;
-
 use libwebrtc_sys::jsep as sys_jsep;
 
+pub use sys_jsep::ffi::{SdpType, SdpParseError};
+
 // TODO Maybe we can replace that by a serialized IceCandidateInit
 #[derive(Debug)]
 pub struct IceCandidate {
@@ -9,6 +10,15 @@ pub struct IceCandidate {
 }
 
 impl IceCandidate {
+    pub fn from(sdp_mid: &str, sdp_mline_index: i32, sdp: &str) -> Result<IceCandidate, SdpParseError> {
+        let res = sys_jsep::ffi::create_ice_candidate(sdp_mid.to_string(), sdp_mline_index, sdp.to_string());
+
+        match res {
+            Ok(cxx_handle) => Ok(IceCandidate::new(cxx_handle)),
+            Err(e) => Err(unsafe { SdpParseError::from(e.what()) }),
+        }
+    }
+
     pub(crate) fn new(cxx_handle: UniquePtr<sys_jsep::ffi::IceCandidate>) -> Self {
         Self { cxx_handle }
     }
@@ -30,6 +40,15 @@ pub struct SessionDescription {
 }
 
 impl SessionDescription {
+    pub fn from(sdp_type: SdpType, description: &str) -> Result<SessionDescription, SdpParseError> {
+        let res = sys_jsep::ffi::create_session_description(sdp_type, description.to_string());
+
+        match res {
+            Ok(cxx_handle) => Ok(SessionDescription::new(cxx_handle)),
+            Err(e) => Err(unsafe { SdpParseError::from(e.what()) }),
+        }
+    }
+
     pub(crate) fn new(cxx_handle: UniquePtr<sys_jsep::ffi::SessionDescription>) -> Self {
         Self { cxx_handle }
     }

crates/livekit-webrtc/src/peer_connection.rs

@@ -1,9 +1,7 @@
-use std::fmt::Debug;
 use std::sync::{Arc, Mutex};
 
 use cxx::UniquePtr;
 use log::trace;
-use thiserror::Error;
 use tokio::sync::mpsc;
 
 use libwebrtc_sys::data_channel as sys_dc;
@@ -22,14 +20,6 @@ use crate::rtc_error::RTCError;
 use crate::rtp_receiver::RtpReceiver;
 use crate::rtp_transceiver::RtpTransceiver;
 
-#[derive(Error, Debug)]
-pub enum SdpError {
-    #[error("recv failure: {0}")]
-    RecvError(String),
-    #[error("internal libwebrtc error")]
-    RTCError(#[from] RTCError),
-}
-
 pub struct PeerConnection {
     cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>,
     observer: Box<InternalObserver>,
@@ -51,7 +41,7 @@ impl PeerConnection {
         }
     }
 
-    pub async fn create_offer(&mut self, options: RTCOfferAnswerOptions) -> Result<SessionDescription, SdpError> {
+    pub async fn create_offer(&mut self, options: RTCOfferAnswerOptions) -> Result<SessionDescription, RTCError> {
         let (tx, mut rx) = mpsc::channel(1);
 
         let wrapper =
@@ -65,13 +55,10 @@ impl PeerConnection {
                 .create_offer(native_wrapper.pin_mut(), options);
         }
 
-        match rx.recv().await {
+        rx.recv().await.unwrap()
-            Some(value) => value.map_err(Into::into),
-            None => Err(SdpError::RecvError("channel closed".to_string())),
-        }
     }
 
-    pub async fn create_answer(&mut self, options: RTCOfferAnswerOptions) -> Result<SessionDescription, SdpError> {
+    pub async fn create_answer(&mut self, options: RTCOfferAnswerOptions) -> Result<SessionDescription, RTCError> {
         let (tx, mut rx) = mpsc::channel(1);
 
         let wrapper =
@@ -85,16 +72,13 @@ impl PeerConnection {
                 .create_answer(native_wrapper.pin_mut(), options);
         }
 
-        match rx.recv().await {
+        rx.recv().await.unwrap()
-            Some(value) => value.map_err(Into::into),
-            None => Err(SdpError::RecvError("channel closed".to_string())),
-        }
     }
 
     pub async fn set_local_description(
         &mut self,
         desc: SessionDescription,
-    ) -> Result<(), SdpError> {
+    ) -> Result<(), RTCError> {
         let (tx, mut rx) = mpsc::channel(1);
         let wrapper =
             sys_jsep::SetLocalSdpObserverWrapper::new(Box::new(InternalSetLocalSdpObserver { tx }));
@@ -107,16 +91,13 @@ impl PeerConnection {
                 .set_local_description(desc.release(), native_wrapper.pin_mut());
         }
 
-        match rx.recv().await {
+        rx.recv().await.unwrap()
-            Some(value) => value.map_err(Into::into),
-            None => Err(SdpError::RecvError("channel closed".to_string())),
-        }
     }
 
     pub async fn set_remote_description(
         &mut self,
         desc: SessionDescription,
-    ) -> Result<(), SdpError> {
+    ) -> Result<(), RTCError> {
         let (tx, mut rx) = mpsc::channel(1);
         let wrapper =
             sys_jsep::SetRemoteSdpObserverWrapper::new(Box::new(InternalSetRemoteSdpObserver {
@@ -131,10 +112,7 @@ impl PeerConnection {
                 .set_remote_description(desc.release(), native_wrapper.pin_mut());
         }
 
-        match rx.recv().await {
+        rx.recv().await.unwrap()
-            Some(value) => value.map_err(Into::into),
-            None => Err(SdpError::RecvError("channel closed".to_string())),
-        }
     }
 
     pub fn create_data_channel(
@@ -155,10 +133,10 @@ impl PeerConnection {
     }
 
     // TODO(theomonnom) Use IceCandidateInit instead of IceCandidate
-    pub async fn add_ice_candidate(&mut self, candidate: IceCandidate) -> Result<(), SdpError> {
+    pub async fn add_ice_candidate(&mut self, candidate: IceCandidate) -> Result<(), RTCError> {
         let (tx, mut rx) = mpsc::channel(1);
-        let observer = sys_pc::AddIceCandidateObserverWrapper::new(Box::new(move |error| {
+        let observer = sys_pc::AddIceCandidateObserverWrapper::new(Box::new(InternalAddIceCandidateObserver {
-            tx.blocking_send(error).unwrap();
+            tx,
         }));
 
         let mut native_observer =
@@ -167,10 +145,7 @@ impl PeerConnection {
             .pin_mut()
             .add_ice_candidate(candidate.release(), native_observer.pin_mut());
 
-        match rx.recv().await {
+        rx.recv().await.unwrap()
-            Some(value) => Ok(()),
-            None => Err(SdpError::RecvError("channel closed".to_string())),
-        }
     }
 
     pub fn local_description(&self) -> Option<SessionDescription> {
@@ -313,6 +288,20 @@ impl PeerConnection {
     }
 }
 
+
+// SetLocalSdpObserver
+
+struct InternalAddIceCandidateObserver {
+    tx: mpsc::Sender<Result<(), RTCError>>,
+}
+
+impl sys_pc::AddIceCandidateObserver for InternalAddIceCandidateObserver {
+    fn on_complete(&self, error: RTCError) {
+        let res = if error.ok() { Ok(()) } else { Err(error) };
+        let _ = self.tx.blocking_send(res);
+    }
+}
+
 // CreateSdpObserver
 
 struct InternalCreateSdpObserver {
@@ -324,13 +313,11 @@ impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver {
         &self,
         session_description: UniquePtr<libwebrtc_sys::jsep::ffi::SessionDescription>,
     ) {
-        self.tx
+        let _ = self.tx.blocking_send(Ok(SessionDescription::new(session_description)));
-            .blocking_send(Ok(SessionDescription::new(session_description)))
-            .unwrap();
     }
 
     fn on_failure(&self, error: RTCError) {
-        self.tx.blocking_send(Err(error)).unwrap();
+        let _ = self.tx.blocking_send(Err(error));
     }
 }
 
@@ -343,7 +330,7 @@ struct InternalSetLocalSdpObserver {
 impl sys_jsep::SetLocalSdpObserver for InternalSetLocalSdpObserver {
     fn on_set_local_description_complete(&self, error: RTCError) {
         let res = if error.ok() { Ok(()) } else { Err(error) };
-        self.tx.blocking_send(res).unwrap();
+        let _ = self.tx.blocking_send(res);
     }
 }
 
@@ -356,7 +343,7 @@ struct InternalSetRemoteSdpObserver {
 impl sys_jsep::SetRemoteSdpObserver for InternalSetRemoteSdpObserver {
     fn on_set_remote_description_complete(&self, error: RTCError) {
         let res = if error.ok() { Ok(()) } else { Err(error) };
-        self.tx.blocking_send(res).unwrap();
+        let _ = self.tx.blocking_send(res);
     }
 }
 
@@ -634,6 +621,8 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
 mod tests {
     use log::trace;
     use tokio::sync::mpsc;
+    use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions;
+    use libwebrtc_sys::peer_connection_factory::ffi::{ContinualGatheringPolicy, IceTransportsType};
 
     use crate::data_channel::{DataChannel, DataChannelInit};
     use crate::jsep::IceCandidate;
@@ -657,6 +646,8 @@ mod tests {
                 username: "".into(),
                 password: "".into(),
             }],
+            continual_gathering_policy: ContinualGatheringPolicy::GatherOnce,
+            ice_transport_type: IceTransportsType::All
         };
 
         let mut bob = factory.create_peer_connection(config.clone()).unwrap();
@@ -682,12 +673,12 @@ mod tests {
             .create_data_channel("test_dc", DataChannelInit::default())
             .unwrap();
 
-        let offer = bob.create_offer().await.unwrap();
+        let offer = bob.create_offer(RTCOfferAnswerOptions::default()).await.unwrap();
         trace!("Bob offer: {:?}", offer);
         bob.set_local_description(offer.clone()).await.unwrap();
         alice.set_remote_description(offer).await.unwrap();
 
-        let answer = alice.create_answer().await.unwrap();
+        let answer = alice.create_answer(RTCOfferAnswerOptions::default()).await.unwrap();
         trace!("Alice answer: {:?}", answer);
         alice.set_local_description(answer.clone()).await.unwrap();
         bob.set_remote_description(answer).await.unwrap();