transports/webrtc: Implement message framing

misc/prost-codec/src/lib.rs

@@ -79,3 +79,12 @@ pub enum Error {
         std::io::Error,
     ),
 }
+
+impl From<Error> for std::io::Error {
+    fn from(e: Error) -> Self {
+        match e {
+            Error::Decode(e) => e.into(),
+            Error::Io(e) => e,
+        }
+    }
+}

transports/webrtc/Cargo.toml

@@ -10,6 +10,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
+asynchronous-codec = "0.6"
 async-trait = "0.1"
 bytes = "1"
 fnv = "1.0"
@@ -23,15 +24,21 @@ libp2p-noise = { version = "0.40.0", path = "../../transports/noise" }
 log = "0.4"
 multihash = { version = "0.16", default-features = false, features = ["sha2"] }
 multibase = "0.9"
+prost = "0.11"
+prost-codec = { version = "0.2", path = "../../misc/prost-codec" }
 rand = "0.8"
 rcgen = "0.9.3"
 serde = { version = "1.0", features = ["derive"] }
 stun = "0.4"
 thiserror = "1"
 tinytemplate = "1.2"
 tokio-crate = { package = "tokio", version = "1.18", features = ["net"]}
+tokio-util = { version = "0.7", features = ["compat"] }
 webrtc = "0.5.0"
 
+[build-dependencies]
+prost-build = "0.11"
+
 [dev-dependencies]
 anyhow = "1.0"
 env_logger = "0.9"
@@ -40,3 +47,5 @@ rand_core = "0.5"
 quickcheck = "1"
 hex-literal = "0.3"
 multihash = { version = "0.16", default-features = false, features = ["sha3"] }
+unsigned-varint = { version = "0.7", features = ["asynchronous_codec"] }
+asynchronous-codec = { version = "0.6" }

transports/webrtc/build.rs

@@ -0,0 +1,23 @@
+// Copyright 2022 Protocol Labs.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+fn main() {
+    prost_build::compile_protos(&["src/message.proto"], &["src"]).unwrap();
+}

transports/webrtc/src/connection.rs

@@ -18,8 +18,6 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-mod poll_data_channel;
-
 use futures::{
     channel::{
         mpsc,
@@ -42,7 +40,7 @@ use std::{
 };
 
 use crate::error::Error;
-pub(crate) use poll_data_channel::PollDataChannel;
+use crate::substream::Substream;
 
 const MAX_DATA_CHANNELS_IN_FLIGHT: usize = 10;
 
@@ -138,7 +136,7 @@ impl Connection {
 }
 
 impl StreamMuxer for Connection {
-    type Substream = PollDataChannel;
+    type Substream = Substream;
     type Error = Error;
 
     fn poll_inbound(
@@ -149,7 +147,7 @@ impl StreamMuxer for Connection {
             Some(detached) => {
                 trace!("Incoming substream {}", detached.stream_identifier());
 
-                Poll::Ready(Ok(PollDataChannel::new(detached)))
+                Poll::Ready(Ok(Substream::new(detached)))
             }
             None => Poll::Ready(Err(Error::Internal(
                 "incoming_data_channels_rx is closed (no messages left)".to_string(),
@@ -198,7 +196,7 @@ impl StreamMuxer for Connection {
         match ready!(fut.as_mut().poll(cx)) {
             Ok(detached) => {
                 self.outbound_fut = None;
-                Poll::Ready(Ok(PollDataChannel::new(detached)))
+                Poll::Ready(Ok(Substream::new(detached)))
             }
             Err(e) => {
                 self.outbound_fut = None;

transports/webrtc/src/lib.rs

@@ -85,9 +85,15 @@ mod error;
 mod fingerprint;
 mod req_res_chan;
 mod sdp;
+mod substream;
 mod transport;
 mod udp_mux;
 mod upgrade;
+mod message_proto {
+    #![allow(clippy::derive_partial_eq_without_eq)]
+
+    include!(concat!(env!("OUT_DIR"), "/webrtc.pb.rs"));
+}
 
 pub use connection::Connection;
 pub use error::Error;

transports/webrtc/src/message.proto

@@ -0,0 +1,20 @@
+syntax = "proto2";
+
+package webrtc.pb;
+
+message Message {
+  enum Flag {
+    // The sender will no longer send messages on the stream.
+    FIN = 0;
+    // The sender will no longer read messages on the stream. Incoming data is
+    // being discarded on receipt.
+    STOP_SENDING = 1;
+    // The sender abruptly terminates the sending part of the stream. The
+    // receiver can discard any data that it already received on that stream.
+    RESET = 2;
+  }
+
+  optional Flag flag=1;
+
+  optional bytes message = 2;
+}