update to support new Transport interface

Refs libp2p#2652

Author: melekes

transports/webrtc/Cargo.toml

@@ -18,8 +18,8 @@ futures-lite = "1"
 futures-timer = "3"
 hex = "0.4"
 if-watch = "0.2"
-libp2p-core = { version = "0.33.0", path = "../../core", default-features = false  }
-libp2p-noise = { version = "0.36.0", path = "../../transports/noise" }
+libp2p-core = { version = "0.34.0", path = "../../core", default-features = false  }
+libp2p-noise = { version = "0.37.0", path = "../../transports/noise" }
 log = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 stun = "0.4"
@@ -35,8 +35,8 @@ webrtc-util = { version = "0.5.3", default-features = false, features = ["conn",
 [dev-dependencies]
 anyhow = "1.0"
 env_logger = "0.9"
-libp2p-request-response = { version = "0.18.0", path = "../../protocols/request-response" }
-libp2p-swarm = { version = "0.36.0", path = "../../swarm" }
+libp2p-request-response = { version = "0.20.0", path = "../../protocols/request-response" }
+libp2p-swarm = { version = "0.38.0", path = "../../swarm" }
 rand = "0.8"
 rand_core = "0.5"
 rcgen = "0.9"

transports/webrtc/src/connection.rs

@@ -33,7 +33,6 @@ use webrtc::peer_connection::RTCPeerConnection;
 use webrtc_data::data_channel::DataChannel as DetachedDataChannel;
 
 use std::io;
-use std::pin::Pin;
 use std::sync::{Arc, Mutex as StdMutex};
 use std::task::{Context, Poll};
 
@@ -226,68 +225,68 @@ impl<'a> StreamMuxer for Connection {
     /// abruptly interrupt the execution.
     fn destroy_outbound(&self, _s: Self::OutboundSubstream) {}
 
-    fn read_substream(
-        &self,
-        cx: &mut Context<'_>,
-        s: &mut Self::Substream,
-        buf: &mut [u8],
-    ) -> Poll<Result<usize, Self::Error>> {
-        Pin::new(s).poll_read(cx, buf)
-    }
-
-    fn write_substream(
-        &self,
-        cx: &mut Context<'_>,
-        s: &mut Self::Substream,
-        buf: &[u8],
-    ) -> Poll<Result<usize, Self::Error>> {
-        Pin::new(s).poll_write(cx, buf)
-    }
-
-    fn flush_substream(
-        &self,
-        cx: &mut Context<'_>,
-        s: &mut Self::Substream,
-    ) -> Poll<Result<(), Self::Error>> {
-        trace!("Flushing substream {}", s.stream_identifier());
-        Pin::new(s).poll_flush(cx)
-    }
-
-    fn shutdown_substream(
-        &self,
-        cx: &mut Context<'_>,
-        s: &mut Self::Substream,
-    ) -> Poll<Result<(), Self::Error>> {
-        trace!("Closing substream {}", s.stream_identifier());
-        Pin::new(s).poll_close(cx)
-    }
-
-    fn destroy_substream(&self, s: Self::Substream) {
-        trace!("Destroying substream {}", s.stream_identifier());
-        let mut data_channels_inner = self.data_channels_inner.lock().unwrap();
-        data_channels_inner.map.remove(&s.stream_identifier());
-    }
+    // fn read_substream(
+    //     &self,
+    //     cx: &mut Context<'_>,
+    //     s: &mut Self::Substream,
+    //     buf: &mut [u8],
+    // ) -> Poll<Result<usize, Self::Error>> {
+    //     Pin::new(s).poll_read(cx, buf)
+    // }
+
+    // fn write_substream(
+    //     &self,
+    //     cx: &mut Context<'_>,
+    //     s: &mut Self::Substream,
+    //     buf: &[u8],
+    // ) -> Poll<Result<usize, Self::Error>> {
+    //     Pin::new(s).poll_write(cx, buf)
+    // }
+
+    // fn flush_substream(
+    //     &self,
+    //     cx: &mut Context<'_>,
+    //     s: &mut Self::Substream,
+    // ) -> Poll<Result<(), Self::Error>> {
+    //     trace!("Flushing substream {}", s.stream_identifier());
+    //     Pin::new(s).poll_flush(cx)
+    // }
+
+    // fn shutdown_substream(
+    //     &self,
+    //     cx: &mut Context<'_>,
+    //     s: &mut Self::Substream,
+    // ) -> Poll<Result<(), Self::Error>> {
+    //     trace!("Closing substream {}", s.stream_identifier());
+    //     Pin::new(s).poll_close(cx)
+    // }
+
+    // fn destroy_substream(&self, s: Self::Substream) {
+    //     trace!("Destroying substream {}", s.stream_identifier());
+    //     let mut data_channels_inner = self.data_channels_inner.lock().unwrap();
+    //     data_channels_inner.map.remove(&s.stream_identifier());
+    // }
 
     fn poll_close(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         debug!("Closing connection");
 
         let mut data_channels_inner = self.data_channels_inner.lock().unwrap();
 
         // First, flush all the buffered data.
-        for (_, ch) in &mut data_channels_inner.map {
-            match ready!(self.flush_substream(cx, ch)) {
-                Ok(_) => continue,
-                Err(e) => return Poll::Ready(Err(e)),
-            }
-        }
+        // for (_, ch) in &mut data_channels_inner.map {
+        //     match ready!(self.flush_substream(cx, ch)) {
+        //         Ok(_) => continue,
+        //         Err(e) => return Poll::Ready(Err(e)),
+        //     }
+        // }
 
         // Second, shutdown all the substreams.
-        for (_, ch) in &mut data_channels_inner.map {
-            match ready!(self.shutdown_substream(cx, ch)) {
-                Ok(_) => continue,
-                Err(e) => return Poll::Ready(Err(e)),
-            }
-        }
+        // for (_, ch) in &mut data_channels_inner.map {
+        //     match ready!(self.shutdown_substream(cx, ch)) {
+        //         Ok(_) => continue,
+        //         Err(e) => return Poll::Ready(Err(e)),
+        //     }
+        // }
 
         // Third, close `incoming_data_channels_rx`
         data_channels_inner.incoming_data_channels_rx.close();

transports/webrtc/src/error.rs

@@ -42,6 +42,9 @@ pub enum Error {
         got: PeerId,
     },
 
+    #[error("no active listeners")]
+    NoListeners,
+
     #[error("internal error: {0} (see debug logs)")]
     InternalError(String),
 }