core/transport: impl Stream for transport::Boxed

core/src/transport/boxed.rs

@@ -19,7 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
-use futures::prelude::*;
+use futures::{prelude::*, stream::FusedStream};
 use multiaddr::Multiaddr;
 use std::{
     error::Error,
@@ -157,6 +157,20 @@ impl<O> Transport for Boxed<O> {
     }
 }
 
+impl<O> Stream for Boxed<O> {
+    type Item = TransportEvent<ListenerUpgrade<O>, io::Error>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Transport::poll(self, cx).map(Some)
+    }
+}
+
+impl<O> FusedStream for Boxed<O> {
+    fn is_terminated(&self) -> bool {
+        false
+    }
+}
+
 fn box_err<E: Error + Send + Sync + 'static>(e: E) -> io::Error {
     io::Error::new(io::ErrorKind::Other, e)
 }

core/src/transport/memory.rs

@@ -401,8 +401,6 @@ impl<T> Drop for Chan<T> {
 
 #[cfg(test)]
 mod tests {
-    use futures::future::poll_fn;
-
     use super::*;
 
     #[test]
@@ -495,12 +493,12 @@ mod tests {
         let t1_addr: Multiaddr = format!("/memory/{}", rand_port).parse().unwrap();
         let cloned_t1_addr = t1_addr.clone();
 
-        let mut t1 = MemoryTransport::default();
+        let mut t1 = MemoryTransport::default().boxed();
 
         let listener = async move {
             t1.listen_on(ListenerId::new(1), t1_addr.clone()).unwrap();
             let upgrade = loop {
-                let event = poll_fn(|cx| Pin::new(&mut t1).poll(cx)).await;
+                let event = t1.select_next_some().await;
                 if let Some(upgrade) = event.into_upgrade() {
                     break upgrade;
                 }
@@ -533,16 +531,14 @@ mod tests {
             Protocol::Memory(rand::random::<u64>().saturating_add(1)).into();
         let listener_addr_cloned = listener_addr.clone();
 
-        let mut listener_transport = MemoryTransport::default();
+        let mut listener_transport = MemoryTransport::default().boxed();
 
         let listener = async move {
             listener_transport
                 .listen_on(ListenerId::new(1), listener_addr.clone())
                 .unwrap();
             loop {
-                if let TransportEvent::Incoming { send_back_addr, .. } =
-                    poll_fn(|cx| Pin::new(&mut listener_transport).poll(cx)).await
-                {
+                if let TransportEvent::Incoming { send_back_addr, .. } = listener_transport.select_next_some().await {
                     assert!(
                         send_back_addr != listener_addr,
                         "Expect dialer address not to equal listener address."
@@ -572,15 +568,15 @@ mod tests {
             Protocol::Memory(rand::random::<u64>().saturating_add(1)).into();
         let listener_addr_cloned = listener_addr.clone();
 
-        let mut listener_transport = MemoryTransport::default();
+        let mut listener_transport = MemoryTransport::default().boxed();
 
         let listener = async move {
             listener_transport
                 .listen_on(ListenerId::new(1), listener_addr.clone())
                 .unwrap();
             loop {
                 if let TransportEvent::Incoming { send_back_addr, .. } =
-                    poll_fn(|cx| Pin::new(&mut listener_transport).poll(cx)).await
+                    listener_transport.select_next_some().await
                 {
                     let dialer_port =
                         NonZeroU64::new(parse_memory_addr(&send_back_addr).unwrap()).unwrap();

core/tests/transport_upgrade.rs

@@ -20,7 +20,6 @@
 
 mod util;
 
-use futures::future::poll_fn;
 use futures::prelude::*;
 use libp2p_core::identity;
 use libp2p_core::transport::{ListenerId, MemoryTransport, Transport};
@@ -96,7 +95,7 @@ fn upgrade_pipeline() {
             // Gracefully close the connection to allow protocol
             // negotiation to complete.
             util::CloseMuxer::new(mplex).map_ok(move |mplex| (peer, mplex))
-        });
+        }).boxed();
 
     let dialer_keys = identity::Keypair::generate_ed25519();
     let dialer_id = dialer_keys.public().to_peer_id();
@@ -114,7 +113,8 @@ fn upgrade_pipeline() {
             // Gracefully close the connection to allow protocol
             // negotiation to complete.
             util::CloseMuxer::new(mplex).map_ok(move |mplex| (peer, mplex))
-        });
+        })
+        .boxed();
 
     let listen_addr1 = Multiaddr::from(Protocol::Memory(random::<u64>()));
     let listen_addr2 = listen_addr1.clone();
@@ -125,14 +125,10 @@ fn upgrade_pipeline() {
 
     let server = async move {
         loop {
-            let (upgrade, _send_back_addr) =
-                match poll_fn(|cx| Pin::new(&mut listener_transport).poll(cx))
-                    .await
-                    .into_upgrade()
-                {
-                    Some(u) => u,
-                    None => continue,
-                };
+            let (upgrade, _send_back_addr) = match listener_transport.select_next_some().await.into_upgrade() {
+                Some(u) => u,
+                None => continue,
+            };
             let (peer, _mplex) = upgrade.await.unwrap();
             assert_eq!(peer, dialer_id);
         }

src/bandwidth.rs

@@ -86,7 +86,7 @@ where
         let this = self.project();
         match this.inner.poll(cx) {
             Poll::Ready(event) => {
-                let event = event.map_upgrade( {
+                let event = event.map_upgrade({
                     let sinks = this.sinks.clone();
                     |inner| BandwidthFuture { inner, sinks }
                 });

swarm/src/lib.rs

@@ -2069,12 +2069,12 @@ mod tests {
                 let mut listen_addresses = Vec::new();
                 let mut transports = Vec::new();
                 for _ in 0..num_listen_addrs {
-                    let mut transport = transport::MemoryTransport::default();
+                    let mut transport = transport::MemoryTransport::default().boxed();
                     transport
                         .listen_on(ListenerId::new(1), "/memory/0".parse().unwrap())
                         .unwrap();
 
-                    match poll_fn(|cx| Pin::new(&mut transport).poll(cx)).await {
+                    match transport.select_next_some().await {
                         TransportEvent::NewAddress { listen_addr, .. } => {
                             listen_addresses.push(listen_addr);
                         }
@@ -2094,9 +2094,8 @@ mod tests {
                     )
                     .unwrap();
                 for mut transport in transports.into_iter() {
-                    let poll_transport = poll_fn(|cx| Pin::new(&mut transport).poll(cx));
                     loop {
-                        match futures::future::select(poll_transport, swarm.next()).await {
+                        match futures::future::select(transport.select_next_some(), swarm.next()).await {
                             Either::Left((TransportEvent::Incoming { .. }, _)) => {
                                 break;
                             }