feat(webrtc): add WebRTC for WASM environments

Cargo.toml

@@ -55,6 +55,8 @@ members = [
     "transports/websocket",
     "transports/webtransport-websys",
     "wasm-tests/webtransport-tests",
+    "wasm-tests/webrtc",
+    "wasm-tests/webrtc/server",
 ]
 resolver = "2"
 
@@ -104,7 +106,7 @@ quickcheck = { package = "quickcheck-ext", path = "misc/quickcheck-ext" }
 rw-stream-sink = { version = "0.4.0", path = "misc/rw-stream-sink" }
 multiaddr = "0.18.0"
 multihash = "0.19.0"
-
+webrtc-websys-test-server = { version = "0.1.0", path = "wasm-tests/webrtc/server" }
 
 [patch.crates-io]
 

transports/webrtc/Cargo.toml

@@ -17,28 +17,74 @@ bytes = "1"
 futures = "0.3"
 futures-timer = "3"
 hex = "0.4"
-if-watch = "3.0"
 libp2p-core = { workspace = true }
 libp2p-noise = { workspace = true }
 libp2p-identity = { workspace = true }
 log = "0.4"
 sha2 = "0.10.7"
-multihash = { workspace = true  }
+multihash = { workspace = true }
 quick-protobuf = "0.8"
 quick-protobuf-codec = { workspace = true }
-rand = "0.8"
 rcgen = "0.10.0"
 serde = { version = "1.0", features = ["derive"] }
-stun = "0.4"
 thiserror = "1"
 tinytemplate = "1.2"
-tokio = { version = "1.29", features = ["net"], optional = true}
+
+# tokio optional deps
+tokio = { version = "1.29", features = ["net"], optional = true }
 tokio-util = { version = "0.7", features = ["compat"], optional = true }
 webrtc = { version = "0.8.0", optional = true }
+stun = { version = "0.4", optional = true }
+rand = { version = "0.8", optional = true }
+if-watch = { version = "3.0", optional = true }
+
+# wasm-bindgen optional deps
+js-sys = { version = "0.3", optional = true }
+getrandom = { version = "0.2.9", features = ["js"], optional = true }
+regex = { version = "1.9", optional = true }
+send_wrapper = { version = "0.6.0", features = ["futures"], optional = true }
+wasm-bindgen = { version = "0.2.87", optional = true }
+wasm-bindgen-futures = { version = "0.4.37", optional = true }
+web-sys = { version = "0.3.64", optional = true, features = [
+    "MessageEvent",
+    "RtcPeerConnection",
+    "RtcSignalingState",
+    "RtcSdpType",
+    "RtcSessionDescription",
+    "RtcSessionDescriptionInit",
+    "RtcPeerConnectionIceEvent",
+    "RtcIceCandidate",
+    "RtcDataChannel",
+    "RtcDataChannelEvent",
+    "RtcCertificate",
+    "RtcConfiguration",
+    "RtcDataChannelInit",
+    "RtcDataChannelType",
+    "RtcDataChannelState",
+] }
 
 [features]
-tokio = ["dep:tokio", "dep:tokio-util", "dep:webrtc", "if-watch/tokio"]
+tokio = [
+    "dep:tokio",
+    "dep:tokio-util",
+    "dep:webrtc",
+    "dep:if-watch",
+    "if-watch/tokio",
+    "dep:rand",
+    "dep:stun",
+]
 pem = ["webrtc?/pem"]
+wasm-bindgen = [
+    "dep:js-sys",
+    "dep:getrandom",
+    "dep:regex",
+    "dep:send_wrapper",
+    "dep:wasm-bindgen",
+    "dep:wasm-bindgen-futures",
+    "dep:web-sys",
+    "futures-timer/wasm-bindgen",
+]
+
 
 [dev-dependencies]
 anyhow = "1.0"

transports/webrtc/src/common/mod.rs

@@ -0,0 +1 @@
+pub mod substream;

transports/webrtc/src/common/substream/mod.rs

@@ -0,0 +1,14 @@
+pub(crate) mod state;
+
+/// Maximum length of a message.
+///
+/// "As long as message interleaving is not supported, the sender SHOULD limit the maximum message
+/// size to 16 KB to avoid monopolization."
+/// Source: <https://www.rfc-editor.org/rfc/rfc8831#name-transferring-user-data-on-a>
+pub(crate) const MAX_MSG_LEN: usize = 16384; // 16kiB
+/// Length of varint, in bytes.
+pub(crate) const VARINT_LEN: usize = 2;
+/// Overhead of the protobuf encoding, in bytes.
+pub(crate) const PROTO_OVERHEAD: usize = 5;
+/// Maximum length of data, in bytes.
+pub(crate) const MAX_DATA_LEN: usize = MAX_MSG_LEN - VARINT_LEN - PROTO_OVERHEAD;

transports/webrtc/src/lib.rs

@@ -88,3 +88,9 @@ mod proto {
 
 #[cfg(feature = "tokio")]
 pub mod tokio;
+
+#[cfg(feature = "wasm-bindgen")]
+pub mod websys;
+
+/// Crates common to all features
+mod common;

transports/webrtc/src/tokio/substream.rs

@@ -31,29 +31,15 @@ use std::{
     task::{Context, Poll},
 };
 
-use crate::proto::{Flag, Message};
-use crate::tokio::{
-    substream::drop_listener::GracefullyClosed,
-    substream::framed_dc::FramedDc,
-    substream::state::{Closing, State},
+use crate::common::substream::{
+    state::{Closing, State},
+    MAX_DATA_LEN,
 };
+use crate::proto::{Flag, Message};
+use crate::tokio::{substream::drop_listener::GracefullyClosed, substream::framed_dc::FramedDc};
 
 mod drop_listener;
 mod framed_dc;
-mod state;
-
-/// Maximum length of a message.
-///
-/// "As long as message interleaving is not supported, the sender SHOULD limit the maximum message
-/// size to 16 KB to avoid monopolization."
-/// Source: <https://www.rfc-editor.org/rfc/rfc8831#name-transferring-user-data-on-a>
-const MAX_MSG_LEN: usize = 16384; // 16kiB
-/// Length of varint, in bytes.
-const VARINT_LEN: usize = 2;
-/// Overhead of the protobuf encoding, in bytes.
-const PROTO_OVERHEAD: usize = 5;
-/// Maximum length of data, in bytes.
-const MAX_DATA_LEN: usize = MAX_MSG_LEN - VARINT_LEN - PROTO_OVERHEAD;
 
 pub(crate) use drop_listener::DropListener;
 /// A substream on top of a WebRTC data channel.
@@ -251,6 +237,8 @@ fn io_poll_next(
 
 #[cfg(test)]
 mod tests {
+    use crate::common::substream::{MAX_MSG_LEN, PROTO_OVERHEAD, VARINT_LEN};
+
     use super::*;
     use asynchronous_codec::Encoder;
     use bytes::BytesMut;

transports/webrtc/src/tokio/substream/framed_dc.rs

@@ -25,7 +25,10 @@ use webrtc::data::data_channel::{DataChannel, PollDataChannel};
 
 use std::sync::Arc;
 
-use super::{MAX_DATA_LEN, MAX_MSG_LEN, VARINT_LEN};
+use crate::common::substream::{
+    state::{Closing, State},
+    MAX_DATA_LEN, MAX_MSG_LEN, VARINT_LEN,
+};
 use crate::proto::Message;
 
 pub(crate) type FramedDc = Framed<Compat<PollDataChannel>, quick_protobuf_codec::Codec<Message>>;

transports/webrtc/src/websys/cbfutures.rs

@@ -0,0 +1,98 @@
+use futures::FutureExt;
+use std::cell::Cell;
+use std::future::Future;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::task::{ready, Context, Poll, Waker};
+
+pub struct FusedCbFuture<T>(Option<CbFuture<T>>);
+
+impl<T> FusedCbFuture<T> {
+    pub(crate) fn new() -> Self {
+        Self(None)
+    }
+
+    pub(crate) fn maybe_init<F>(&mut self, init: F) -> &mut Self
+    where
+        F: FnOnce() -> CbFuture<T>,
+    {
+        if self.0.is_none() {
+            self.0 = Some(init());
+        }
+
+        self
+    }
+
+    /// Checks if future is already running
+    pub(crate) fn is_active(&self) -> bool {
+        self.0.is_some()
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct CbFuture<T>(Rc<CallbackFutureInner<T>>);
+
+struct CallbackFutureInner<T> {
+    waker: Cell<Option<Waker>>,
+    result: Cell<Option<T>>,
+}
+
+impl<T> std::fmt::Debug for CallbackFutureInner<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("CallbackFutureInner").finish()
+    }
+}
+
+impl<T> Default for CallbackFutureInner<T> {
+    fn default() -> Self {
+        Self {
+            waker: Cell::new(None),
+            result: Cell::new(None),
+        }
+    }
+}
+
+impl<T> CbFuture<T> {
+    /// New Callback Future
+    pub(crate) fn new() -> Self {
+        Self(Rc::new(CallbackFutureInner::<T>::default()))
+    }
+
+    // call this from your callback
+    pub(crate) fn publish(&self, result: T) {
+        self.0.result.set(Some(result));
+        if let Some(w) = self.0.waker.take() {
+            w.wake()
+        };
+    }
+}
+
+impl<T> Future for CbFuture<T> {
+    type Output = T;
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match self.0.result.take() {
+            Some(x) => Poll::Ready(x),
+            None => {
+                self.0.waker.set(Some(cx.waker().clone()));
+                Poll::Pending
+            }
+        }
+    }
+}
+
+impl<T> Future for FusedCbFuture<T> {
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let val = ready!(self
+            .0
+            .as_mut()
+            .expect("FusedCbFuture not initialized")
+            .poll_unpin(cx));
+
+        // Future finished, drop it
+        self.0.take();
+
+        Poll::Ready(val)
+    }
+}