talksik/client-sdk-rust
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

test #1

Browse Source
This commit is contained in:
Théo Monnom
2022-09-14 21:33:34 +02:00
parent 12a6d232b5
commit f643f458f8
23 changed files with 519 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/livekit-webrtc/src/peer_connection.rs
+224 -79
View File
@@ -1,23 +1,22 @@
use std::sync::{Arc, Mutex};
use cxx::UniquePtr;
use tokio::sync::{mpsc, oneshot};
use libwebrtc_sys::peer_connection as sys_pc;
use libwebrtc_sys::jsep as sys_jsep;
use libwebrtc_sys::peer_connection as sys_pc;
use std::sync::{Arc, Mutex};
use thiserror::Error;
use tokio::sync::{mpsc, oneshot};
use crate::data_channel::DataChannel;
use crate::media_stream::MediaStream;
use crate::ice_candidate::IceCandidate;
use crate::rtc_error::RTCError;
use crate::rtp_receiver::RtpReceiver;
use crate::rtp_transceiver::RtpTransceiver;
use crate::session_description::SessionDescription;
use crate::rtc_error::RTCError;
use crate::jsep::{SessionDescription, IceCandidate};
pub use libwebrtc_sys::peer_connection::ffi::PeerConnectionState;
pub use libwebrtc_sys::peer_connection::ffi::SignalingState;
pub use libwebrtc_sys::peer_connection::ffi::IceConnectionState;
pub use libwebrtc_sys::peer_connection::ffi::IceGatheringState;
pub use libwebrtc_sys::peer_connection::ffi::PeerConnectionState;
pub use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions;
pub use libwebrtc_sys::peer_connection::ffi::SignalingState;
#[derive(Error, Debug)]
pub enum SdpError {
@@ -29,44 +28,27 @@ pub enum SdpError {
pub struct PeerConnection {
cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>,
observer: InternalObserver
observer: InternalObserver,
}
impl PeerConnection {
pub fn new(cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>) -> Self {
pub(crate) fn new(cxx_handle: UniquePtr<sys_pc::ffi::PeerConnection>) -> Self {
Self {
cxx_handle,
observer: InternalObserver {
on_signaling_change_handler: Arc::new(Default::default()),
on_add_stream_handler: Arc::new(Default::default()),
on_remove_stream_handler: Arc::new(Default::default()),
on_data_channel_handler: Arc::new(Default::default()),
on_renegotiation_needed_handler: Arc::new(Default::default()),
on_negotiation_needed_event_handler: Arc::new(Default::default()),
on_ice_connection_change_handler: Arc::new(Default::default()),
on_standardized_ice_connection_change_handler: Arc::new(Default::default()),
on_connection_change_handler: Arc::new(Default::default()),
on_ice_gathering_change_handler: Arc::new(Default::default()),
on_ice_candidate_handler: Arc::new(Default::default()),
on_ice_candidate_error_handler: Arc::new(Default::default()),
on_ice_candidates_removed_handler: Arc::new(Default::default()),
on_ice_connection_receiving_change_handler: Arc::new(Default::default()),
on_ice_selected_candidate_pair_changed_handler: Arc::new(Default::default()),
on_add_track_handler: Arc::new(Default::default()),
on_track_handler: Arc::new(Default::default()),
on_remove_track_handler: Arc::new(Default::default()),
on_interesting_usage_handler: Arc::new(Default::default())
}
observer: InternalObserver::default()
}
}
pub async fn create_offer(&mut self) -> Result<SessionDescription, SdpError> {
let (tx, mut rx) = mpsc::channel(1);
let wrapper = sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx }));
let wrapper =
sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx }));
let native_wrapper = sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper));
self.cxx_handle.pin_mut().create_offer(native_wrapper, RTCOfferAnswerOptions::default());
self.cxx_handle
.pin_mut()
.create_offer(native_wrapper, RTCOfferAnswerOptions::default());
match rx.recv().await {
Some(value) => value.map_err(Into::into),
@@ -77,10 +59,13 @@ impl PeerConnection {
pub async fn create_answer(&mut self) -> Result<SessionDescription, SdpError> {
let (tx, mut rx) = mpsc::channel(1);
let wrapper = sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx }));
let wrapper =
sys_jsep::CreateSdpObserverWrapper::new(Box::new(InternalCreateSdpObserver { tx }));
let native_wrapper = sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper));
self.cxx_handle.pin_mut().create_answer(native_wrapper, RTCOfferAnswerOptions::default());
self.cxx_handle
.pin_mut()
.create_answer(native_wrapper, RTCOfferAnswerOptions::default());
match rx.recv().await {
Some(value) => value.map_err(Into::into),
@@ -88,15 +73,42 @@ impl PeerConnection {
}
}
pub async fn set_local_description(&mut self, desc: SessionDescription) -> Result<(), SdpError> {
pub async fn set_local_description(
&mut self,
desc: SessionDescription,
) -> Result<(), SdpError> {
let (tx, mut rx) = mpsc::channel(1);
let wrapper =
sys_jsep::SetLocalSdpObserverWrapper::new(Box::new(InternalSetLocalSdpObserver { tx }));
let native_wrapper = sys_jsep::ffi::create_native_set_local_sdp_observer(Box::new(wrapper));
self.cxx_handle
.pin_mut()
.set_local_description(desc.release(), native_wrapper);
Ok(())
match rx.recv().await {
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> {
pub async fn set_remote_description(
&mut self,
desc: SessionDescription,
) -> Result<(), SdpError> {
let (tx, mut rx) = mpsc::channel(1);
let wrapper =
sys_jsep::SetRemoteSdpObserverWrapper::new(Box::new(InternalSetRemoteSdpObserver { tx }));
let native_wrapper = sys_jsep::ffi::create_native_set_remote_sdp_observer(Box::new(wrapper));
Ok(())
self.cxx_handle
.pin_mut()
.set_remote_description(desc.release(), native_wrapper);
match rx.recv().await {
Some(value) => value.map_err(Into::into),
None => Err(SdpError::RecvError("channel closed".to_string())),
}
}
pub fn on_signaling_change(&mut self, handler: OnSignalingChangeHandler) {
@@ -116,15 +128,30 @@ impl PeerConnection {
}
pub fn on_renegotiation_needed(&mut self, handler: OnRenegotiationNeededHandler) {
*self.observer.on_renegotiation_needed_handler.lock().unwrap() = Some(handler);
*self
.observer
.on_renegotiation_needed_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_ice_connection_change(&mut self, handler: OnIceConnectionChangeHandler) {
*self.observer.on_ice_connection_change_handler.lock().unwrap() = Some(handler);
*self
.observer
.on_ice_connection_change_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_standardized_ice_connection_change(&mut self, handler: OnStandardizedIceConnectionChangeHandler) {
*self.observer.on_standardized_ice_connection_change_handler.lock().unwrap() = Some(handler);
pub fn on_standardized_ice_connection_change(
&mut self,
handler: OnStandardizedIceConnectionChangeHandler,
) {
*self
.observer
.on_standardized_ice_connection_change_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_connection_change(&mut self, handler: OnConnectionChangeHandler) {
@@ -132,7 +159,11 @@ impl PeerConnection {
}
pub fn on_ice_gathering_change(&mut self, handler: OnIceGatheringChangeHandler) {
*self.observer.on_ice_gathering_change_handler.lock().unwrap() = Some(handler);
*self
.observer
.on_ice_gathering_change_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_ice_candidate(&mut self, handler: OnIceCandidateHandler) {
@@ -144,15 +175,33 @@ impl PeerConnection {
}
pub fn on_ice_candidates_removed(&mut self, handler: OnIceCandidatesRemovedHandler) {
*self.observer.on_ice_candidates_removed_handler.lock().unwrap() = Some(handler);
*self
.observer
.on_ice_candidates_removed_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_ice_connection_receiving_change(&mut self, handler: OnIceConnectionReceivingChangeHandler) {
*self.observer.on_ice_connection_receiving_change_handler.lock().unwrap() = Some(handler);
pub fn on_ice_connection_receiving_change(
&mut self,
handler: OnIceConnectionReceivingChangeHandler,
) {
*self
.observer
.on_ice_connection_receiving_change_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_ice_selected_candidate_pair_changed(&mut self, handler: OnIceSelectedCandidatePairChangedHandler) {
*self.observer.on_ice_selected_candidate_pair_changed_handler.lock().unwrap() = Some(handler);
pub fn on_ice_selected_candidate_pair_changed(
&mut self,
handler: OnIceSelectedCandidatePairChangedHandler,
) {
*self
.observer
.on_ice_selected_candidate_pair_changed_handler
.lock()
.unwrap() = Some(handler);
}
pub fn on_add_track(&mut self, handler: OnAddTrackHandler) {
@@ -175,44 +224,48 @@ impl PeerConnection {
// CreateSdpObserver
struct InternalCreateSdpObserver {
tx: mpsc::Sender<Result<SessionDescription, RTCError>>
tx: mpsc::Sender<Result<SessionDescription, RTCError>>,
}
impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver {
fn on_success(&self, session_description: UniquePtr<libwebrtc_sys::jsep::ffi::SessionDescription>) {
self.tx.blocking_send(Ok(SessionDescription{})).unwrap(); // TODO
fn on_success(
&self,
session_description: UniquePtr<libwebrtc_sys::jsep::ffi::SessionDescription>,
) {
self.tx.blocking_send(Ok(SessionDescription::new(session_description))).unwrap();
}
fn on_failure(&self, error: RTCError) {
self.tx.blocking_send(Err(error)).unwrap(); // TODO
self.tx.blocking_send(Err(error)).unwrap();
}
}
// SetLocalSdpObserver
struct InternalSetLocalSdpObserver {
tx: mpsc::Sender<Result<(), RTCError>>
tx: mpsc::Sender<Result<(), RTCError>>,
}
impl sys_jsep::SetLocalSdpObserver for InternalSetLocalSdpObserver {
fn on_set_local_description_complete(&self, error: RTCError) {
self.tx.blocking_send(Ok(())).unwrap();
let res = if error.ok() { Ok(()) } else { Err(error) };
self.tx.blocking_send(res).unwrap();
}
}
// SetRemoteSdpObserver
struct InternalSetRemoteSdpObserver {
tx: mpsc::Sender<Result<(), RTCError>>,
}
impl sys_jsep::SetRemoteSdpObserver for InternalSetRemoteSdpObserver {
fn on_set_remote_description_complete(&self, error: RTCError) {
todo!()
let res = if error.ok() { Ok(()) } else { Err(error) };
self.tx.blocking_send(res).unwrap();
}
}
// PeerConnectionObserver
// TODO(theomonnom) Should we return futures?
@@ -223,20 +276,23 @@ pub type OnDataChannelHandler = Box<dyn FnMut(DataChannel) + Send + Sync>;
pub type OnRenegotiationNeededHandler = Box<dyn FnMut() + Send + Sync>;
pub type OnNegotiationNeededEventHandler = Box<dyn FnMut(u32) + Send + Sync>;
pub type OnIceConnectionChangeHandler = Box<dyn FnMut(IceConnectionState) + Send + Sync>;
pub type OnStandardizedIceConnectionChangeHandler = Box<dyn FnMut(IceConnectionState) + Send + Sync>;
pub type OnStandardizedIceConnectionChangeHandler =
Box<dyn FnMut(IceConnectionState) + Send + Sync>;
pub type OnConnectionChangeHandler = Box<dyn FnMut(PeerConnectionState) + Send + Sync>;
pub type OnIceGatheringChangeHandler = Box<dyn FnMut(IceGatheringState) + Send + Sync>;
pub type OnIceCandidateHandler = Box<dyn FnMut(IceCandidate) + Send + Sync>;
pub type OnIceCandidateErrorHandler = Box<dyn FnMut(String, i32, String, i32, String) + Send + Sync>;
pub type OnIceCandidateErrorHandler =
Box<dyn FnMut(String, i32, String, i32, String) + Send + Sync>;
pub type OnIceCandidatesRemovedHandler = Box<dyn FnMut(Vec<IceCandidate>) + Send + Sync>;
pub type OnIceConnectionReceivingChangeHandler = Box<dyn FnMut(bool) + Send + Sync>;
pub type OnIceSelectedCandidatePairChangedHandler = Box<dyn FnMut(libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent) + Send + Sync>;
pub type OnIceSelectedCandidatePairChangedHandler =
Box<dyn FnMut(libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent) + Send + Sync>;
pub type OnAddTrackHandler = Box<dyn FnMut(RtpReceiver, Vec<MediaStream>) + Send + Sync>;
pub type OnTrackHandler = Box<dyn FnMut(RtpTransceiver) + Send + Sync>;
pub type OnRemoveTrackHandler = Box<dyn FnMut(RtpReceiver) + Send + Sync>;
pub type OnInterestingUsageHandler = Box<dyn FnMut(i32) + Send + Sync>;
struct InternalObserver {
pub(crate) struct InternalObserver {
on_signaling_change_handler: Arc<Mutex<Option<OnSignalingChangeHandler>>>,
on_add_stream_handler: Arc<Mutex<Option<OnAddStreamHandler>>>,
on_remove_stream_handler: Arc<Mutex<Option<OnRemoveStreamHandler>>>,
@@ -244,18 +300,47 @@ struct InternalObserver {
on_renegotiation_needed_handler: Arc<Mutex<Option<OnRenegotiationNeededHandler>>>,
on_negotiation_needed_event_handler: Arc<Mutex<Option<OnNegotiationNeededEventHandler>>>,
on_ice_connection_change_handler: Arc<Mutex<Option<OnIceConnectionChangeHandler>>>,
on_standardized_ice_connection_change_handler: Arc<Mutex<Option<OnStandardizedIceConnectionChangeHandler>>>,
on_standardized_ice_connection_change_handler:
Arc<Mutex<Option<OnStandardizedIceConnectionChangeHandler>>>,
on_connection_change_handler: Arc<Mutex<Option<OnConnectionChangeHandler>>>,
on_ice_gathering_change_handler: Arc<Mutex<Option<OnIceGatheringChangeHandler>>>,
on_ice_candidate_handler: Arc<Mutex<Option<OnIceCandidateHandler>>>,
on_ice_candidate_error_handler: Arc<Mutex<Option<OnIceCandidateErrorHandler>>>,
on_ice_candidates_removed_handler: Arc<Mutex<Option<OnIceCandidatesRemovedHandler>>>,
on_ice_connection_receiving_change_handler: Arc<Mutex<Option<OnIceConnectionReceivingChangeHandler>>>,
on_ice_selected_candidate_pair_changed_handler: Arc<Mutex<Option<OnIceSelectedCandidatePairChangedHandler>>>,
on_ice_connection_receiving_change_handler:
Arc<Mutex<Option<OnIceConnectionReceivingChangeHandler>>>,
on_ice_selected_candidate_pair_changed_handler:
Arc<Mutex<Option<OnIceSelectedCandidatePairChangedHandler>>>,
on_add_track_handler: Arc<Mutex<Option<OnAddTrackHandler>>>,
on_track_handler: Arc<Mutex<Option<OnTrackHandler>>>,
on_remove_track_handler: Arc<Mutex<Option<OnRemoveTrackHandler>>>,
on_interesting_usage_handler: Arc<Mutex<Option<OnInterestingUsageHandler>>>
on_interesting_usage_handler: Arc<Mutex<Option<OnInterestingUsageHandler>>>,
}
impl Default for InternalObserver {
fn default() -> Self {
Self {
on_signaling_change_handler: Arc::new(Default::default()),
on_add_stream_handler: Arc::new(Default::default()),
on_remove_stream_handler: Arc::new(Default::default()),
on_data_channel_handler: Arc::new(Default::default()),
on_renegotiation_needed_handler: Arc::new(Default::default()),
on_negotiation_needed_event_handler: Arc::new(Default::default()),
on_ice_connection_change_handler: Arc::new(Default::default()),
on_standardized_ice_connection_change_handler: Arc::new(Default::default()),
on_connection_change_handler: Arc::new(Default::default()),
on_ice_gathering_change_handler: Arc::new(Default::default()),
on_ice_candidate_handler: Arc::new(Default::default()),
on_ice_candidate_error_handler: Arc::new(Default::default()),
on_ice_candidates_removed_handler: Arc::new(Default::default()),
on_ice_connection_receiving_change_handler: Arc::new(Default::default()),
on_ice_selected_candidate_pair_changed_handler: Arc::new(Default::default()),
on_add_track_handler: Arc::new(Default::default()),
on_track_handler: Arc::new(Default::default()),
on_remove_track_handler: Arc::new(Default::default()),
on_interesting_usage_handler: Arc::new(Default::default()),
}
}
}
// Observers are being called on the Signaling Thread
@@ -267,21 +352,30 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
}
}
fn on_add_stream(&mut self, stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>) {
fn on_add_stream(
&mut self,
stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
) {
let mut handler = self.on_add_stream_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
}
}
fn on_remove_stream(&mut self, stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>) {
fn on_remove_stream(
&mut self,
stream: UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>,
) {
let mut handler = self.on_remove_stream_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
}
}
fn on_data_channel(&mut self, data_channel: UniquePtr<libwebrtc_sys::data_channel::ffi::DataChannel>) {
fn on_data_channel(
&mut self,
data_channel: UniquePtr<libwebrtc_sys::data_channel::ffi::DataChannel>,
) {
let mut handler = self.on_data_channel_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
@@ -310,7 +404,10 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
}
fn on_standardized_ice_connection_change(&mut self, new_state: IceConnectionState) {
let mut handler = self.on_standardized_ice_connection_change_handler.lock().unwrap();
let mut handler = self
.on_standardized_ice_connection_change_handler
.lock()
.unwrap();
if let Some(f) = handler.as_mut() {
f(new_state);
}
@@ -337,14 +434,24 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
}
}
fn on_ice_candidate_error(&mut self, address: String, port: i32, url: String, error_code: i32, error_text: String) {
fn on_ice_candidate_error(
&mut self,
address: String,
port: i32,
url: String,
error_code: i32,
error_text: String,
) {
let mut handler = self.on_ice_candidate_error_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
f(address, port, url, error_code, error_text);
}
}
fn on_ice_candidates_removed(&mut self, removed: Vec<UniquePtr<libwebrtc_sys::candidate::ffi::Candidate>>) {
fn on_ice_candidates_removed(
&mut self,
removed: Vec<UniquePtr<libwebrtc_sys::candidate::ffi::Candidate>>,
) {
let mut handler = self.on_ice_candidates_removed_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
@@ -352,34 +459,53 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
}
fn on_ice_connection_receiving_change(&mut self, receiving: bool) {
let mut handler = self.on_ice_connection_receiving_change_handler.lock().unwrap();
let mut handler = self
.on_ice_connection_receiving_change_handler
.lock()
.unwrap();
if let Some(f) = handler.as_mut() {
f(receiving);
}
}
fn on_ice_selected_candidate_pair_changed(&mut self, event: libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent) {
let mut handler = self.on_ice_selected_candidate_pair_changed_handler.lock().unwrap();
fn on_ice_selected_candidate_pair_changed(
&mut self,
event: libwebrtc_sys::peer_connection::ffi::CandidatePairChangeEvent,
) {
let mut handler = self
.on_ice_selected_candidate_pair_changed_handler
.lock()
.unwrap();
if let Some(f) = handler.as_mut() {
f(event);
}
}
fn on_add_track(&mut self, receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>, streams: Vec<UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>>) {
fn on_add_track(
&mut self,
receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
streams: Vec<UniquePtr<libwebrtc_sys::media_stream_interface::ffi::MediaStreamInterface>>,
) {
let mut handler = self.on_add_track_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
}
}
fn on_track(&mut self, transceiver: UniquePtr<libwebrtc_sys::rtp_transceiver::ffi::RtpTransceiver>) {
fn on_track(
&mut self,
transceiver: UniquePtr<libwebrtc_sys::rtp_transceiver::ffi::RtpTransceiver>,
) {
let mut handler = self.on_track_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
}
}
fn on_remove_track(&mut self, receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>) {
fn on_remove_track(
&mut self,
receiver: UniquePtr<libwebrtc_sys::rtp_receiver::ffi::RtpReceiver>,
) {
let mut handler = self.on_remove_track_handler.lock().unwrap();
if let Some(f) = handler.as_mut() {
// TODO(theomonnom)
@@ -392,4 +518,23 @@ impl sys_pc::PeerConnectionObserver for InternalObserver {
f(usage_pattern);
}
}
}
}
#[cfg(test)]
mod tests {
use libwebrtc_sys::peer_connection_factory::ffi::RTCConfiguration;
use crate::peer_connection_factory::PeerConnectionFactory;
#[tokio::test]
async fn create_pc() {
let factory = PeerConnectionFactory::new();
let mut pc = factory.create_peer_connection(
RTCConfiguration {
ice_servers: vec![],
},
Box::new(()),
).unwrap();
let offer = pc.create_offer().await.unwrap();
}
}