Rust version with thread-based worker (#540)

.github/workflows/mediasoup-rust.yaml

@@ -12,7 +12,7 @@ jobs:
       matrix:
         rust:
           - stable
-          - nightly
+          # - nightly
 
     steps:
       - name: Checkout
@@ -51,10 +51,10 @@ jobs:
           command: clippy
           args: -- -D warnings
 
-      - run: npm install
-
       - name: cargo test
         uses: actions-rs/cargo@v1
+        env:
+          KEEP_BUILD_ARTIFACTS: '1'
         with:
           command: test
           args: --verbose

Cargo.toml

@@ -1,4 +1,5 @@
 [workspace]
 members = [
-    "rust"
+    "rust",
+    "worker"
 ]

rust/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mediasoup"
-version = "0.6.0"
+version = "0.7.1"
 description = "Cutting Edge WebRTC Video Conferencing in Rust"
 categories = ["api-bindings", "multimedia", "network-programming"]
 authors = ["Nazar Mokrynskyi <[email protected]>"]
@@ -11,51 +11,57 @@ documentation = "https://docs.rs/mediasoup"
 repository = "https://github.com/versatica/mediasoup/tree/v3/rust"
 readme = "readme.md"
 
+[package.metadata.docs.rs]
+default-target = "x86_64-unknown-linux-gnu"
+targets = []
+
 [dependencies]
-async-channel = "1.5.1"
+async-channel = "1.6.1"
 async-executor = "1.4.0"
 async-fs = "1.5.0"
 async-io = "1.3.1"
 async-lock = "2.3.0"
-async-oneshot = "0.4.2"
-async-process = "1.0.1"
-async-trait = "0.1.42"
-bytes = "1.0.0"
+async-oneshot = "0.5.0"
+async-trait = "0.1.48"
+bytes = "1.0.1"
 event-listener-primitives = "0.2.2"
 fastrand = "1.4.0"
 futures-lite = "1.11.3"
 h264-profile-level-id = "0.1.1"
-log = "0.4.11"
-libc = "0.2.82"
-nix = "0.19.1"
-once_cell = "1.5.2"
-serde_json = "1.0.61"
+log = "0.4.14"
+libc = "0.2.90"
+once_cell = "1.7.2"
+serde_json = "1.0.64"
 serde_repr = "0.1.6"
-thiserror = "1.0.23"
+thiserror = "1.0.24"
 
 [dependencies.lru]
 default-features = false
 version = "0.6.5"
 
+[dependencies.mediasoup-sys]
+path = "../worker"
+version = "0.1.1"
+
 [dependencies.parking_lot]
 version = "0.11.1"
 features = ["serde"]
 
 [dependencies.regex]
 default-features = false
 features = ["std", "perf"]
-version = "1.4.2"
+version = "1.4.5"
 
 [dependencies.serde]
 features = ["derive"]
-version = "1.0.118"
+version = "1.0.124"
 
 [dependencies.uuid]
 features = ["serde", "v4"]
-version = "0.8.1"
+version = "0.8.2"
 
 [dev-dependencies]
 actix = "0.10.0"
 actix-web = "3.3.2"
 actix-web-actors = "3.0.0"
-env_logger = "0.8.2"
+env_logger = "0.8.3"

rust/changelog.md

@@ -0,0 +1,15 @@
+# Changelog
+
+### 0.7.0
+
+* Switch from running C++ worker processes to worker threads using mediasoup-sys that wraps mediasoup-worker into library
+* Simplify `WorkerManager::new()` and `WorkerManager::with_executor()` API as the result of above
+* Support `rtxPacketsDiscarded` in `Producer` stats
+* Enable Rust 2018 idioms warnings
+* Make sure all public types have `Debug` implementation on them
+* Enforce docs on public types and add missing documentation
+* Remove `RtpCodecParametersParameters::new()` (`RtpCodecParametersParameters::default()` does the same thing)
+
+### 0.6.0
+
+Initial upstreamed release.

rust/examples/echo.rs

@@ -18,7 +18,6 @@ use mediasoup::worker::WorkerSettings;
 use mediasoup::worker_manager::WorkerManager;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::env;
 use std::num::{NonZeroU32, NonZeroU8};
 
 fn media_codecs() -> Vec<RtpCodecCapability> {
@@ -35,7 +34,7 @@ fn media_codecs() -> Vec<RtpCodecCapability> {
             mime_type: MimeTypeVideo::Vp8,
             preferred_payload_type: None,
             clock_rate: NonZeroU32::new(90000).unwrap(),
-            parameters: RtpCodecParametersParameters::new(),
+            parameters: RtpCodecParametersParameters::default(),
             rtcp_feedback: vec![
                 RtcpFeedback::Nack,
                 RtcpFeedback::NackPli,
@@ -403,11 +402,7 @@ async fn ws_index(
 async fn main() -> std::io::Result<()> {
     env_logger::init();
 
-    let worker_manager = WorkerManager::new(
-        env::var("MEDIASOUP_WORKER_BIN")
-            .map(|path| path.into())
-            .unwrap_or_else(|_| "../worker/out/Release/mediasoup-worker".into()),
-    );
+    let worker_manager = WorkerManager::new();
     HttpServer::new(move || {
         App::new()
             .data(worker_manager.clone())

rust/readme.md

@@ -35,7 +35,7 @@ enable different use cases and scenarios, without any constraint or assumption.
 * Congestion control.
 * Sender and receiver bandwidth estimation with spatial/temporal layers distribution algorithm.
 * Data message exchange (via WebRTC DataChannels, SCTP over plain UDP, and direct termination in Rust/Node.js).
-* Extremely powerful (media worker subprocess coded in C++ on top of [libuv](https://libuv.org)).
+* Extremely powerful (media worker subprocess/thead coded in C++ on top of [libuv](https://libuv.org)).
 
 ## Authors
 * Iñaki Baz Castillo [[website](https://inakibaz.me)|[github](https://github.com/ibc/)]

rust/src/data_structures.rs

@@ -39,6 +39,7 @@ impl DerefMut for AppData {
 }
 
 impl AppData {
+    /// Construct app data from almost anything
     pub fn new<T: Any + Send + Sync>(app_data: T) -> Self {
         Self(Arc::new(app_data))
     }
@@ -80,7 +81,7 @@ pub struct IceParameters {
     pub ice_lite: Option<bool>,
 }
 
-/// ICE candidate type
+/// ICE candidate type.
 #[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Deserialize, Serialize)]
 #[serde(rename_all = "lowercase")]
 pub enum IceCandidateType {
@@ -104,14 +105,17 @@ pub enum IceCandidateType {
 #[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Deserialize, Serialize)]
 #[serde(rename_all = "lowercase")]
 pub enum IceCandidateTcpType {
+    /// Passive.
     Passive,
 }
 
 /// Transport protocol.
 #[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Deserialize, Serialize)]
 #[serde(rename_all = "lowercase")]
 pub enum TransportProtocol {
+    /// TCP.
     Tcp,
+    /// UDP.
     Udp,
 }
 
@@ -120,7 +124,7 @@ pub enum TransportProtocol {
 #[serde(rename_all = "camelCase")]
 pub struct IceCandidate {
     /// Unique identifier that allows ICE to correlate candidates that appear on multiple
-    /// `transports`.
+    /// transports.
     pub foundation: String,
     /// The assigned priority of the candidate.
     pub priority: u32,
@@ -151,7 +155,7 @@ pub enum IceState {
     /// ICE was `Connected` or `Completed` but it has suddenly failed (this can just happen if the
     /// selected tuple has `Tcp` protocol).
     Disconnected,
-    /// ICE state when the `transport` has been closed.
+    /// ICE state when the transport has been closed.
     Closed,
 }
 
@@ -166,6 +170,7 @@ pub enum IceState {
 #[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Deserialize, Serialize)]
 #[serde(untagged)]
 pub enum TransportTuple {
+    /// Transport tuple with remote endpoint info.
     #[serde(rename_all = "camelCase")]
     WithRemote {
         /// Local IP address.
@@ -179,6 +184,7 @@ pub enum TransportTuple {
         /// Protocol
         protocol: TransportProtocol,
     },
+    /// Transport tuple without remote endpoint info.
     #[serde(rename_all = "camelCase")]
     LocalOnly {
         /// Local IP address.
@@ -247,22 +253,27 @@ impl Default for DtlsRole {
 /// corresponding certificate fingerprint value.
 #[derive(Debug, Copy, Clone, PartialOrd, PartialEq)]
 pub enum DtlsFingerprint {
+    /// sha-1
     Sha1 {
         /// Certificate fingerprint value.
         value: [u8; 20],
     },
+    /// sha-224
     Sha224 {
         /// Certificate fingerprint value.
         value: [u8; 28],
     },
+    /// sha-256
     Sha256 {
         /// Certificate fingerprint value.
         value: [u8; 32],
     },
+    /// sha-384
     Sha384 {
         /// Certificate fingerprint value.
         value: [u8; 48],
     },
+    /// sha-512
     Sha512 {
         /// Certificate fingerprint value.
         value: [u8; 64],
@@ -541,7 +552,7 @@ impl<'de> Deserialize<'de> for DtlsFingerprint {
         impl<'de> Visitor<'de> for DtlsFingerprintVisitor {
             type Value = DtlsFingerprint;
 
-            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                 formatter.write_str(
                     r#"DTLS fingerprint algorithm and value like {"algorithm": "sha-256", "value": "1B:EA:BF:33:B8:11:26:6D:91:AD:1B:A0:16:FD:5D:60:59:33:F7:46:A3:BA:99:2A:1D:04:99:A6:F2:C6:2D:43"}"#,
                 )
@@ -551,8 +562,8 @@ impl<'de> Deserialize<'de> for DtlsFingerprint {
             where
                 V: MapAccess<'de>,
             {
-                let mut algorithm = None::<Cow<str>>;
-                let mut value = None::<Cow<str>>;
+                let mut algorithm = None::<Cow<'_, str>>;
+                let mut value = None::<Cow<'_, str>>;
                 while let Some(key) = map.next_key()? {
                     match key {
                         Field::Algorithm => {
@@ -685,20 +696,27 @@ pub struct DtlsParameters {
     pub fingerprints: Vec<DtlsFingerprint>,
 }
 
+/// Trace event direction
 #[derive(Debug, Copy, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "lowercase")]
-pub enum EventDirection {
+pub enum TraceEventDirection {
+    /// In
     In,
+    /// Out
     Out,
 }
 
 /// Container used for sending/receiving messages using `DirectTransport` data producers and data
 /// consumers.
-#[derive(Clone)]
+#[derive(Debug, Clone)]
 pub enum WebRtcMessage {
+    /// String
     String(String),
+    /// Binary
     Binary(Bytes),
+    /// EmptyString
     EmptyString,
+    /// EmptyBinary
     EmptyBinary,
 }
 
@@ -714,16 +732,15 @@ impl WebRtcMessage {
     // | WebRTC Binary Empty                | 57        |
     // +------------------------------------+-----------+
 
-    pub(crate) fn new(ppid: u32, payload: Bytes) -> Self {
-        // TODO: Make this fallible instead
+    pub(crate) fn new(ppid: u32, payload: Bytes) -> Result<Self, u32> {
         match ppid {
-            51 => WebRtcMessage::String(String::from_utf8(payload.to_vec()).unwrap()),
-            53 => WebRtcMessage::Binary(payload),
-            56 => WebRtcMessage::EmptyString,
-            57 => WebRtcMessage::EmptyBinary,
-            _ => {
-                panic!("Bad ppid {}", ppid);
-            }
+            51 => Ok(WebRtcMessage::String(
+                String::from_utf8(payload.to_vec()).unwrap(),
+            )),
+            53 => Ok(WebRtcMessage::Binary(payload)),
+            56 => Ok(WebRtcMessage::EmptyString),
+            57 => Ok(WebRtcMessage::EmptyBinary),
+            ppid => Err(ppid),
         }
     }
 

rust/src/lib.rs

@@ -1,3 +1,4 @@
+#![warn(rust_2018_idioms, missing_debug_implementations, missing_docs)]
 //! Rust port of [mediasoup](https://github.com/versatica/mediasoup) TypeScript library!
 //!
 //! For general information go to readme in repository.
@@ -19,13 +20,8 @@
 //! It is recommended to visit mediasoup website and read
 //! [design overview](https://mediasoup.org/documentation/v3/mediasoup/design/) first.
 //!
-//! Pre-requisite for using this library is to have mediasoup worker. The simples way to get it is
-//! to run `npm i mediasoup` (see
-//! [installation instructions](https://mediasoup.org/documentation/v3/mediasoup/installation/)) and
-//! get worker binary from `node_modules/mediasoup/worker/out/Release/mediasoup-worker`.
-//!
-//! If you are hacking on the library just run `npm i` in the root of the repo and tests will pick
-//! up the worker binary automatically.
+//! There are some requirements for building underlying C++ `mediasoup-worker`, please find them in
+//! [installation instructions](https://mediasoup.org/documentation/v3/mediasoup/installation/)
 //!
 //! # Examples
 //! There are some examples in `examples` and `examples-frontend` directories (for server- and
@@ -36,7 +32,7 @@
 //! # How to start
 //! With that in mind, you want start with creating [`WorkerManager`](worker_manager::WorkerManager)
 //! instance and then 1 or more workers. Workers a responsible for low-level job of sending media
-//! and data back and forth. Each worker is backed by single-core C++ worker process. On each worker
+//! and data back and forth. Each worker is backed by single-core C++ worker thread. On each worker
 //! you create one or more routers that enable injection, selection and forwarding of media and data
 //! through [`transport`] instances. There are a few different transports available, but most likely
 //! you'll want to use [`WebRtcTransport`](webrtc_transport::WebRtcTransport) most often. With

rust/src/messages.rs

@@ -23,7 +23,7 @@ use crate::sctp_parameters::{NumSctpStreams, SctpParameters, SctpStreamParameter
 use crate::srtp_parameters::{SrtpCryptoSuite, SrtpParameters};
 use crate::transport::{TransportId, TransportTraceEventType};
 use crate::webrtc_transport::{TransportListenIps, WebRtcTransportOptions};
-use crate::worker::{WorkerDump, WorkerResourceUsage, WorkerUpdateSettings};
+use crate::worker::{WorkerDump, WorkerUpdateSettings};
 use parking_lot::Mutex;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
@@ -174,13 +174,9 @@ macro_rules! request_response_generic {
     };
 }
 
-request_response!("worker.dump", WorkerDumpRequest {}, WorkerDump);
+request_response!("worker.close", WorkerCloseRequest {});
 
-request_response!(
-    "worker.getResourceUsage",
-    WorkerGetResourceRequest {},
-    WorkerResourceUsage,
-);
+request_response!("worker.dump", WorkerDumpRequest {}, WorkerDump);
 
 request_response!(
     "worker.updateSettings",