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

Better RTCError handling + tests

Browse Source
This commit is contained in:
Théo Monnom
2022-09-13 14:19:44 +02:00
parent d9fd69b5c0
commit 12a6d232b5
16 changed files with 325 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+40 -6
View File
@@ -439,9 +439,9 @@ dependencies = [
"env_logger",
"futures-util",
"log",
"prost",
"prost 0.11.0",
"prost-build",
"prost-types",
"prost-types 0.11.1",
"thiserror",
"tokio",
"tokio-tungstenite",
@@ -456,6 +456,7 @@ dependencies = [
"env_logger",
"libwebrtc-sys",
"log",
"thiserror",
"tokio",
]
@@ -597,7 +598,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "71adf41db68aa0daaefc69bb30bcd68ded9b9abaad5d1fbb6304c4fb390e083e"
dependencies = [
"bytes",
"prost-derive",
"prost-derive 0.10.1",
]
[[package]]
name = "prost"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "399c3c31cdec40583bb68f0b18403400d01ec4289c383aa047560439952c4dd7"
dependencies = [
"bytes",
"prost-derive 0.11.0",
]
[[package]]
@@ -615,8 +626,8 @@ dependencies = [
"log",
"multimap",
"petgraph",
"prost",
"prost-types",
"prost 0.10.4",
"prost-types 0.10.1",
"regex",
"tempfile",
"which",
@@ -635,6 +646,19 @@ dependencies = [
"syn",
]
[[package]]
name = "prost-derive"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7345d5f0e08c0536d7ac7229952590239e77abf0a0100a1b1d890add6ea96364"
dependencies = [
"anyhow",
"itertools",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "prost-types"
version = "0.10.1"
@@ -642,7 +666,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2d0a014229361011dc8e69c8a1ec6c2e8d0f2af7c91e3ea3f5b2170298461e68"
dependencies = [
"bytes",
"prost",
"prost 0.10.4",
]
[[package]]
name = "prost-types"
version = "0.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4dfaa718ad76a44b3415e6c4d53b17c8f99160dcb3a99b10470fce8ad43f6e3e"
dependencies = [
"bytes",
"prost 0.11.0",
]
[[package]]
crates/livekit-core/Cargo.toml
+3 -4
View File
@@ -2,7 +2,6 @@
name = "livekit-core"
version = "0.1.0"
edition = "2021"
crate-type = ["lib"]
[dependencies]
log = "0.4"
@@ -10,9 +9,9 @@ tokio-tungstenite = "0.17.2"
tokio = { version = "1.20.1", features = ["full"] }
url = "2.2.2"
futures-util = "0.3.23"
thiserror = "1.0.32"
prost = "0.10"
prost-types = "0.10"
thiserror = "1.0"
prost = "0.11.0"
prost-types = "0.11.1"
anyhow = "1.0.63"
[build-dependencies]
crates/livekit-webrtc/Cargo.toml
+1
View File
@@ -9,6 +9,7 @@ libwebrtc-sys = { path = "./libwebrtc-sys" }
tokio = { version = "1.20.1", features = ["full"] }
cxx = "1.0"
log = "0.4"
thiserror = "1.0"
[dev-dependencies]
env_logger = "0.9"
crates/livekit-webrtc/libwebrtc-sys/build.rs
+5 -4
View File
@@ -235,21 +235,22 @@ fn main() {
}
}
// TODO(theomonnom) Only add this define when building tests
builder.define("LIVEKIT_TEST", None);
builder.warnings(false).compile("lkwebrtc");
for entry in glob::glob("./src/**/*.cpp").unwrap() {
println!(
"cargo:rerun-if-changed={}",
entry.unwrap().display().to_string()
entry.unwrap().display()
);
}
for entry in glob::glob("./include/**/*.h").unwrap() {
println!(
"cargo:rerun-if-changed={}",
entry.unwrap().display().to_string()
entry.unwrap().display()
);
}
println!("cargo:rerun-if-changed=src/main.rs");
}
crates/livekit-webrtc/libwebrtc-sys/include/livekit/rtc_error.h
+9 -9
View File
@@ -6,20 +6,20 @@
#define CLIENT_SDK_NATIVE_RTC_ERROR_H
#include "api/rtc_error.h"
#include "libwebrtc-sys/src/rtc_error.rs.h"
#include "rust_types.h"
#include "rust/cxx.h"
namespace livekit {
class RTCError {
public:
explicit RTCError(webrtc::RTCError error);
RTCError to_error(const webrtc::RTCError &error);
std::string serialize_error(const RTCError &error); // to be used inside cxx::Exception msg
private:
webrtc::RTCError rtc_error_;
};
#ifdef LIVEKIT_TEST
rust::String serialize_deserialize();
void throw_error();
#endif
static std::unique_ptr<RTCError> _unique_rtc_error(){
return nullptr; // Ignore
}
} // livekit
#endif //CLIENT_SDK_NATIVE_RTC_ERROR_H
crates/livekit-webrtc/libwebrtc-sys/include/livekit/rust_types.h
+1
View File
@@ -17,6 +17,7 @@ namespace livekit {
// Shared types
struct RTCOfferAnswerOptions;
struct RTCError;
}
#endif //RUST_TYPES_H
crates/livekit-webrtc/libwebrtc-sys/src/jsep.cpp
+4 -3
View File
@@ -4,6 +4,7 @@
#include <memory>
#include "livekit/rtc_error.h"
#include "livekit/jsep.h"
#include "libwebrtc-sys/src/jsep.rs.h"
#include "api/make_ref_counted.h"
@@ -39,7 +40,7 @@ namespace livekit {
}
void NativeCreateSdpObserver::OnFailure(webrtc::RTCError error) {
observer_->on_failure(std::make_unique<RTCError>(error));
observer_->on_failure(to_error(error));
}
std::unique_ptr<NativeCreateSdpObserverHandle> create_native_create_sdp_observer(rust::Box<CreateSdpObserverWrapper> observer){
@@ -55,7 +56,7 @@ namespace livekit {
}
void NativeSetLocalSdpObserver::OnSetLocalDescriptionComplete(webrtc::RTCError error) {
observer_->on_set_local_description_complete(std::make_unique<RTCError>(error));
observer_->on_set_local_description_complete(to_error(error));
}
std::unique_ptr<NativeSetLocalSdpObserverHandle> create_native_set_local_sdp_observer(rust::Box<SetLocalSdpObserverWrapper> observer){
@@ -71,7 +72,7 @@ namespace livekit {
}
void NativeSetRemoteSdpObserver::OnSetRemoteDescriptionComplete(webrtc::RTCError error) {
observer_->on_set_remote_description_complete(std::make_unique<RTCError>(error));
observer_->on_set_remote_description_complete(to_error(error));
}
std::unique_ptr<NativeSetRemoteSdpObserverHandle> create_native_set_remote_sdp_observer(rust::Box<SetRemoteSdpObserverWrapper> observer){
crates/livekit-webrtc/libwebrtc-sys/src/jsep.rs
+11 -10
View File
@@ -5,21 +5,22 @@ use crate::rtc_error::ffi::RTCError;
#[cxx::bridge(namespace = "livekit")]
pub mod ffi {
extern "Rust" {
type CreateSdpObserverWrapper;
fn on_success(self: &CreateSdpObserverWrapper, session_description: UniquePtr<SessionDescription>);
fn on_failure(self: &CreateSdpObserverWrapper, error: UniquePtr<RTCError>);
fn on_failure(self: &CreateSdpObserverWrapper, error: RTCError);
type SetLocalSdpObserverWrapper;
fn on_set_local_description_complete(self: &SetLocalSdpObserverWrapper, error: UniquePtr<RTCError>);
fn on_set_local_description_complete(self: &SetLocalSdpObserverWrapper, error: RTCError);
type SetRemoteSdpObserverWrapper;
fn on_set_remote_description_complete(self: &SetRemoteSdpObserverWrapper, error: UniquePtr<RTCError>);
fn on_set_remote_description_complete(self: &SetRemoteSdpObserverWrapper, error: RTCError);
}
unsafe extern "C++" {
include!("libwebrtc-sys/src/rtc_error.rs.h");
include!("livekit/jsep.h");
include!("livekit/rtc_error.h");
type RTCError = crate::rtc_error::ffi::RTCError;
type IceCandidate;
@@ -41,7 +42,7 @@ pub mod ffi {
pub trait CreateSdpObserver: Send + Sync {
fn on_success(&self, session_description: UniquePtr<ffi::SessionDescription>);
fn on_failure(&self, error: UniquePtr<RTCError>);
fn on_failure(&self, error: RTCError);
}
pub struct CreateSdpObserverWrapper {
@@ -59,7 +60,7 @@ impl CreateSdpObserverWrapper {
self.observer.on_success(session_description);
}
fn on_failure(&self, error: UniquePtr<RTCError>) {
fn on_failure(&self, error: RTCError) {
self.observer.on_failure(error);
}
}
@@ -67,7 +68,7 @@ impl CreateSdpObserverWrapper {
// SetLocalSdpObserver
pub trait SetLocalSdpObserver: Send + Sync {
fn on_set_local_description_complete(&self, error: UniquePtr<RTCError>);
fn on_set_local_description_complete(&self, error: RTCError);
}
pub struct SetLocalSdpObserverWrapper {
@@ -81,7 +82,7 @@ impl SetLocalSdpObserverWrapper {
}
}
fn on_set_local_description_complete(&self, error: UniquePtr<RTCError>) {
fn on_set_local_description_complete(&self, error: RTCError) {
self.observer.on_set_local_description_complete(error);
}
}
@@ -89,7 +90,7 @@ impl SetLocalSdpObserverWrapper {
// SetRemoteSdpObserver
pub trait SetRemoteSdpObserver: Send + Sync {
fn on_set_remote_description_complete(&self, error: UniquePtr<RTCError>);
fn on_set_remote_description_complete(&self, error: RTCError);
}
pub struct SetRemoteSdpObserverWrapper {
@@ -103,7 +104,7 @@ impl SetRemoteSdpObserverWrapper {
}
}
fn on_set_remote_description_complete(&self, error: UniquePtr<RTCError>) {
fn on_set_remote_description_complete(&self, error: RTCError) {
self.observer.on_set_remote_description_complete(error);
}
}
crates/livekit-webrtc/libwebrtc-sys/src/peer_connection.rs
+23 -23
View File
@@ -24,44 +24,44 @@ pub mod ffi {
#[derive(Debug)]
#[repr(u32)]
pub enum PeerConnectionState {
kNew,
kConnecting,
kConnected,
kDisconnected,
kFailed,
kClosed,
New,
Connecting,
Connected,
Disconnected,
Failed,
Closed,
}
#[derive(Debug)]
#[repr(u32)]
pub enum SignalingState {
kStable,
kHaveLocalOffer,
kHaveLocalPrAnswer,
kHaveRemoteOffer,
kHaveRemotePrAnswer,
kClosed,
Stable,
HaveLocalOffer,
HaveLocalPrAnswer,
HaveRemoteOffer,
HaveRemotePrAnswer,
Closed,
}
#[derive(Debug)]
#[repr(u32)]
pub enum IceConnectionState {
kIceConnectionNew,
kIceConnectionChecking,
kIceConnectionConnected,
kIceConnectionCompleted,
kIceConnectionFailed,
kIceConnectionDisconnected,
kIceConnectionClosed,
kIceConnectionMax,
IceConnectionNew,
IceConnectionChecking,
IceConnectionConnected,
IceConnectionCompleted,
IceConnectionFailed,
IceConnectionDisconnected,
IceConnectionClosed,
IceConnectionMax,
}
#[derive(Debug)]
#[repr(u32)]
pub enum IceGatheringState {
kIceGatheringNew,
kIceGatheringGathering,
kIceGatheringComplete
IceGatheringNew,
IceGatheringGathering,
IceGatheringComplete
}
#[derive(Debug)]
crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.cpp
+2 -1
View File
@@ -11,6 +11,7 @@
#include "api/task_queue/default_task_queue_factory.h"
#include "api/rtc_event_log/rtc_event_log_factory.h"
#include "libwebrtc-sys/src/peer_connection_factory.rs.h"
#include "livekit/rtc_error.h"
namespace livekit{
@@ -57,7 +58,7 @@ namespace livekit{
auto result = peer_factory_->CreatePeerConnectionOrError(*config, std::move(deps));
if(!result.ok()){
throw std::runtime_error(result.error().message()); // TODO(theomonnom) Bridge RTCError - Mb use ProtoBuf
throw std::runtime_error(serialize_error(to_error(result.error())));
}
return std::make_unique<PeerConnection>(std::move(result.value()), std::move(observer));
crates/livekit-webrtc/libwebrtc-sys/src/peer_connection_factory.rs
+3 -3
View File
@@ -42,8 +42,7 @@ pub mod ffi {
unsafe fn create_peer_connection(self: &PeerConnectionFactory, config: UniquePtr<NativeRTCConfiguration>, observer: UniquePtr<NativePeerConnectionObserver>) -> Result<UniquePtr<PeerConnection>>;
}
}
}
@@ -51,6 +50,7 @@ pub mod ffi {
/*
@@ -182,5 +182,5 @@ mod test {
}
}
*/
crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.cpp
+46 -1
View File
@@ -3,9 +3,54 @@
//
#include "livekit/rtc_error.h"
#include <iomanip>
#include <sstream>
#include <string>
namespace livekit {
RTCError::RTCError(webrtc::RTCError error) : rtc_error_(std::move(error)) {
RTCError to_error(const webrtc::RTCError &error) {
RTCError lk_error;
lk_error.error_detail = static_cast<RTCErrorDetailType>(error.error_detail());
lk_error.error_type = static_cast<RTCErrorType>(error.type());
lk_error.has_sctp_cause_code = error.sctp_cause_code().has_value();
lk_error.sctp_cause_code = error.sctp_cause_code().value();
lk_error.message = error.message();
return lk_error;
}
std::string serialize_error(const RTCError &error) {
std::stringstream ss;
ss << std::hex << std::setfill('0');
ss << std::setw(8) << (uint32_t) error.error_type;
ss << std::setw(8) << (uint32_t) error.error_detail;
ss << std::setw(2) << (uint16_t) error.has_sctp_cause_code;
ss << std::setw(4) << (uint16_t) error.sctp_cause_code;
ss << std::dec << std::setw(1) << std::string(error.message);
return ss.str();
}
#ifdef LIVEKIT_TEST
rust::String serialize_deserialize(){
RTCError lk_error;
lk_error.error_type = RTCErrorType::InternalError;
lk_error.error_detail = RTCErrorDetailType::DataChannelFailure;
lk_error.has_sctp_cause_code = true;
lk_error.sctp_cause_code = 24;
lk_error.message = "this is not a test, I repeat, this is not a test";
return serialize_error(lk_error);
}
void throw_error() {
RTCError lk_error;
lk_error.error_type = RTCErrorType::InvalidModification;
lk_error.error_detail = RTCErrorDetailType::None;
lk_error.has_sctp_cause_code = false;
lk_error.sctp_cause_code = 0;
lk_error.message = "exception is thrown!";
throw std::runtime_error(serialize_error(lk_error));
}
#endif
} // livekit
crates/livekit-webrtc/libwebrtc-sys/src/rtc_error.rs
+108 -6
View File
@@ -1,13 +1,115 @@
// TODO(theomonnom) Don't use RTCError as Opaque. I should use a Struct and serialize/deserialize when needed for Result<>
use std::fmt::{Display, Formatter};
use std::error::Error;
#[cxx::bridge(namespace = "livekit")]
pub mod ffi {
unsafe extern "C++" {
include!("livekit/rtc_error.h");
type RTCError;
#[derive(Debug)]
#[repr(u32)]
pub enum RTCErrorType {
None,
UnsupportedOperation,
UnsupportedParameter,
InvalidParameter,
InvalidRange,
SyntaxError,
InvalidState,
InvalidModification,
NetworkError,
ResourceExhausted,
InternalError,
OperationErrorWithData,
}
fn _unique_rtc_error() -> UniquePtr<RTCError>;
#[derive(Debug)]
#[repr(u32)]
pub enum RTCErrorDetailType {
None,
DataChannelFailure,
DtlsFailure,
FingerprintFailure,
SctpFailure,
SdpSyntaxError,
HardwareEncoderNotAvailable,
HardwareEncoderError,
}
#[derive(Debug)]
pub struct RTCError {
pub error_type: RTCErrorType,
pub message: String,
pub error_detail: RTCErrorDetailType,
pub has_sctp_cause_code: bool, // cxx doesn't support the Option trait
pub sctp_cause_code: u16
}
}
impl ffi::RTCError {
/// # Safety
/// The value must be correctly encoded
pub unsafe fn from(value: &str) -> Self {
// Parse the hex encoded error from c++
let error_type = u32::from_str_radix(&value[0..8], 16).unwrap();
let error_detail = u32::from_str_radix(&value[8..16], 16).unwrap();
let has_scp_cause_code = u8::from_str_radix(&value[16..18], 16).unwrap();
let sctp_cause_code = u16::from_str_radix(&value[18..22], 16).unwrap();
let message = String::from(&value[22..]); // msg isn't encoded
Self {
error_type: unsafe { std::mem::transmute(error_type) },
error_detail: unsafe { std::mem::transmute(error_detail) },
sctp_cause_code,
has_sctp_cause_code: has_scp_cause_code == 1,
message,
}
}
}
impl Error for ffi::RTCError {
}
impl Display for ffi::RTCError {
fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
write!(f, "RtcError occurred {:?}: {}", self.error_type, self.message)
}
}
#[cfg(test)]
mod tests {
use crate::rtc_error::ffi::{RTCError, RTCErrorDetailType, RTCErrorType};
#[cxx::bridge(namespace = "livekit")]
pub mod ffi {
unsafe extern "C++" {
include!("livekit/rtc_error.h");
fn serialize_deserialize() -> String;
fn throw_error() -> Result<()>;
}
}
#[test]
fn serialize_deserialize(){
let str = ffi::serialize_deserialize();
let error = unsafe { RTCError::from(&str) };
assert_eq!(error.error_type, RTCErrorType::InternalError);
assert_eq!(error.error_detail, RTCErrorDetailType::DataChannelFailure);
assert_eq!(error.has_sctp_cause_code, true);
assert_eq!(error.sctp_cause_code, 24);
assert_eq!(error.message, "this is not a test, I repeat, this is not a test");
}
#[test]
fn throw_error(){
let exc: cxx::Exception = ffi::throw_error().err().unwrap();
let error = unsafe { RTCError::from(exc.what()) };
assert_eq!(error.error_type, RTCErrorType::InvalidModification);
assert_eq!(error.error_detail, RTCErrorDetailType::None);
assert_eq!(error.has_sctp_cause_code, false);
assert_eq!(error.sctp_cause_code, 0);
assert_eq!(error.message, "exception is thrown!");
}
}
crates/livekit-webrtc/src/peer_connection.rs
+65 -6
View File
@@ -1,8 +1,9 @@
use std::sync::{Arc, Mutex};
use cxx::UniquePtr;
use tokio::sync::mpsc;
use tokio::sync::{mpsc, oneshot};
use libwebrtc_sys::peer_connection as sys_pc;
use libwebrtc_sys::jsep as sys_jsep;
use thiserror::Error;
use crate::data_channel::DataChannel;
use crate::media_stream::MediaStream;
@@ -18,6 +19,14 @@ pub use libwebrtc_sys::peer_connection::ffi::IceConnectionState;
pub use libwebrtc_sys::peer_connection::ffi::IceGatheringState;
pub use libwebrtc_sys::peer_connection::ffi::RTCOfferAnswerOptions;
#[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: InternalObserver
@@ -51,18 +60,43 @@ impl PeerConnection {
}
}
pub async fn create_offer(&mut self) -> Result<SessionDescription, RTCError> {
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 native_wrapper = sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper));
self.cxx_handle.pin_mut().create_offer(native_wrapper, RTCOfferAnswerOptions::default());
rx.recv().await.unwrap()
match rx.recv().await {
Some(value) => value.map_err(Into::into),
None => Err(SdpError::RecvError("channel closed".to_string())),
}
}
pub async fn create_answer(&mut self) -> Result<SessionDescription, RTCError> {
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 native_wrapper = sys_jsep::ffi::create_native_create_sdp_observer(Box::new(wrapper));
self.cxx_handle.pin_mut().create_answer(native_wrapper, RTCOfferAnswerOptions::default());
match rx.recv().await {
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> {
Ok(())
}
pub async fn set_remote_description(&mut self, desc: SessionDescription) -> Result<(), SdpError> {
Ok(())
}
pub fn on_signaling_change(&mut self, handler: OnSignalingChangeHandler) {
@@ -149,11 +183,36 @@ impl sys_jsep::CreateSdpObserver for InternalCreateSdpObserver {
self.tx.blocking_send(Ok(SessionDescription{})).unwrap(); // TODO
}
fn on_failure(&self, error: UniquePtr<libwebrtc_sys::rtc_error::ffi::RTCError>) {
self.tx.blocking_send(Err(RTCError{})).unwrap(); // TODO
fn on_failure(&self, error: RTCError) {
self.tx.blocking_send(Err(error)).unwrap(); // TODO
}
}
// SetLocalSdpObserver
struct InternalSetLocalSdpObserver {
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();
}
}
// SetRemoteSdpObserver
struct InternalSetRemoteSdpObserver {
}
impl sys_jsep::SetRemoteSdpObserver for InternalSetRemoteSdpObserver {
fn on_set_remote_description_complete(&self, error: RTCError) {
todo!()
}
}
// PeerConnectionObserver
// TODO(theomonnom) Should we return futures?
crates/livekit-webrtc/src/peer_connection_factory.rs
+2 -3
View File
@@ -28,11 +28,10 @@ impl PeerConnectionFactory {
match pc_result {
Ok(cxx_handle) => {
let pc = PeerConnection::new(cxx_handle);
Ok(pc)
Ok(PeerConnection::new(cxx_handle))
}
Err(e) => {
Err(RTCError{}) // TODO
Err(unsafe {RTCError::from(e.what()) }) // TODO
}
}
}
crates/livekit-webrtc/src/rtc_error.rs
+2 -3
View File
@@ -1,5 +1,4 @@
use std::fmt;
pub use libwebrtc_sys::rtc_error::ffi::RTCError;
#[derive(Debug)]
pub struct RTCError {
}