don't do framing during noise handshake

transports/webrtc/src/tokio/mod.rs

@@ -21,6 +21,7 @@
 mod connection;
 mod error;
 mod fingerprint;
+mod noise_substream;
 mod req_res_chan;
 mod sdp;
 mod substream;

transports/webrtc/src/tokio/noise_substream.rs

@@ -0,0 +1,82 @@
+// Copyright 2022 Parity Technologies (UK) Ltd.
+//
+// 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.
+
+use futures::prelude::*;
+use webrtc::data::data_channel::{DataChannel, PollDataChannel};
+
+use std::{
+    io,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+
+/// A substream on top of a WebRTC data channel without framing (see [`Substream`]).
+#[derive(Debug)]
+pub struct NoiseSubstream(PollDataChannel);
+
+impl NoiseSubstream {
+    /// Constructs a new [`NoiseSubstream`].
+    pub fn new(data_channel: Arc<DataChannel>) -> Self {
+        Self(PollDataChannel::new(data_channel))
+    }
+}
+
+impl AsyncRead for NoiseSubstream {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        let mut read_buf = tokio::io::ReadBuf::new(buf);
+        futures::ready!(tokio::io::AsyncRead::poll_read(
+            Pin::new(&mut self.0),
+            cx,
+            &mut read_buf
+        ))?;
+        Poll::Ready(Ok(read_buf.filled().len()))
+    }
+}
+
+impl AsyncWrite for NoiseSubstream {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        tokio::io::AsyncWrite::poll_write(Pin::new(&mut self.0), cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        tokio::io::AsyncWrite::poll_flush(Pin::new(&mut self.0), cx)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        tokio::io::AsyncWrite::poll_shutdown(Pin::new(&mut self.0), cx)
+    }
+
+    fn poll_write_vectored(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[io::IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        tokio::io::AsyncWrite::poll_write_vectored(Pin::new(&mut self.0), cx, bufs)
+    }
+}

transports/webrtc/src/tokio/substream.rs

@@ -70,7 +70,7 @@ pub struct Substream {
 }
 
 impl Substream {
-    /// Returns a new `Substream` and a listener, which will notify the receiver when/if the substream
+    /// Returns a new [`Substream`] and a listener, which will notify the receiver when/if the substream
     /// is dropped.
     pub(crate) fn new(data_channel: Arc<DataChannel>) -> (Self, DropListener) {
         let (sender, receiver) = oneshot::channel();

transports/webrtc/src/tokio/upgrade.rs

@@ -39,7 +39,9 @@ use webrtc::peer_connection::RTCPeerConnection;
 
 use std::{net::SocketAddr, sync::Arc, time::Duration};
 
-use crate::tokio::{error::Error, fingerprint::Fingerprint, sdp, substream::Substream, Connection};
+use crate::tokio::{
+    error::Error, fingerprint::Fingerprint, noise_substream::NoiseSubstream, sdp, Connection,
+};
 
 /// Creates a new outbound WebRTC connection.
 pub async fn outbound(
@@ -201,7 +203,7 @@ async fn get_remote_fingerprint(conn: &RTCPeerConnection) -> Fingerprint {
 
 async fn create_substream_for_noise_handshake(
     conn: &RTCPeerConnection,
-) -> Result<Substream, Error> {
+) -> Result<NoiseSubstream, Error> {
     // Open a data channel to do Noise on top and verify the remote.
     let data_channel = conn
         .create_data_channel(
@@ -230,8 +232,5 @@ async fn create_substream_for_noise_handshake(
         }
     };
 
-    let (substream, drop_listener) = Substream::new(channel);
-    drop(drop_listener); // Don't care about cancelled substreams during initial handshake.
-
-    Ok(substream)
+    Ok(NoiseSubstream::new(channel))
 }