Transport: Poll Transport directly, remove ListenersStream

core/src/either.rs

@@ -344,8 +344,8 @@ impl<A: ProtocolName, B: ProtocolName> ProtocolName for EitherName<A, B> {
     }
 }
 #[pin_project(project = EitherTransportProj)]
-#[derive(Debug, Copy, Clone)]
-#[must_use = "futures do nothing unless polled"]
+#[derive(Debug)]
+#[must_use = "transports do nothing unless polled"]
 pub enum EitherTransport<A, B> {
     Left(#[pin] A),
     Right(#[pin] B),

core/src/transport.rs

@@ -108,15 +108,8 @@ pub trait Transport {
     /// obtained from [dialing](Transport::dial).
     type Dial: Future<Output = Result<Self::Output, Self::Error>>;
 
-    // TODO: fix docs
-    /// Listens on the given [`Multiaddr`], producing a stream of pending, inbound connections
-    /// and addresses this transport is listening on (cf. [`TransportEvent`]).
-    ///
-    /// Returning an error from the stream is considered fatal. The listener can also report
-    /// non-fatal errors by producing a [`TransportEvent::Error`].
-    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>>
-    where
-        Self: Sized;
+    /// Listens on the given [`Multiaddr`] for inbound connections.
+    fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>>;
 
     /// Remove a listener.
     ///
@@ -128,9 +121,7 @@ pub trait Transport {
     ///
     /// If [`TransportError::MultiaddrNotSupported`] is returned, it may be desirable to
     /// try an alternative [`Transport`], if available.
-    fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>>
-    where
-        Self: Sized;
+    fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>>;
 
     /// As [`Transport::dial`] but has the local node act as a listener on the outgoing connection.
     ///
@@ -140,17 +131,23 @@ pub trait Transport {
     fn dial_as_listener(
         &mut self,
         addr: Multiaddr,
-    ) -> Result<Self::Dial, TransportError<Self::Error>>
-    where
-        Self: Sized;
+    ) -> Result<Self::Dial, TransportError<Self::Error>>;
 
-    // TODO: Add docs
+    /// Poll for [`TransportEvent`]s.
+    ///
+    /// A [`TransportEvent::Incoming`] should be produced whenever a connection is received at the lowest
+    /// level of the transport stack. The item must be a [`ListenerUpgrade`](Transport::ListenerUpgrade)
+    /// future that resolves to an [`Output`](Transport::Output) value once all protocol upgrades have
+    /// been applied.
+    ///
+    /// Transports are expected to produce [`TransportEvent::Incoming`] events only for
+    /// listen addresses which have previously been announced via
+    /// a [`TransportEvent::NewAddress`] event and which have not been invalidated by
+    /// an [`TransportEvent::AddressExpired`] event yet.
     fn poll(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>>
-    where
-        Self: Sized;
+    ) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>>;
 
     /// Performs a transport-specific mapping of an address `observed` by
     /// a remote onto a local `listen` address to yield an address for
@@ -160,7 +157,7 @@ pub trait Transport {
     /// Boxes the transport, including custom transport errors.
     fn boxed(self) -> boxed::Boxed<Self::Output>
     where
-        Self: Transport + Sized + Send + Unpin + 'static,
+        Self: Sized + Send + Unpin + 'static,
         Self::Dial: Send + 'static,
         Self::ListenerUpgrade: Send + 'static,
         Self::Error: Send + Sync,
@@ -246,11 +243,6 @@ impl Default for ListenerId {
 }
 
 /// Event produced by [`Transport`]s.
-///
-/// Transports are expected to produce `Incoming` events only for
-/// listen addresses which have previously been announced via
-/// a `NewAddress` event and which have not been invalidated by
-/// an `AddressExpired` event yet.
 pub enum TransportEvent<TUpgr, TErr> {
     /// A new address is being listened on.
     NewAddress {
@@ -289,7 +281,7 @@ pub enum TransportEvent<TUpgr, TErr> {
     ///
     /// The listener will continue to be polled for new events and the event
     /// is for informational purposes only.
-    Error {
+    ListenerError {
         /// The ID of the listener that errored.
         listener_id: ListenerId,
         /// The error value.
@@ -298,6 +290,8 @@ pub enum TransportEvent<TUpgr, TErr> {
 }
 
 impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
+    /// In case this [`TransportEvent`] is an upgrade, apply the given function
+    /// to the upgrade and produce another transport event based the the function's result.
     pub fn map_upgrade<U>(self, map: impl FnOnce(TUpgr) -> U) -> TransportEvent<U, TErr> {
         match self {
             TransportEvent::Incoming {
@@ -325,8 +319,8 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
                 listen_addr,
                 listener_id,
             },
-            TransportEvent::Error { listener_id, error } => {
-                TransportEvent::Error { listener_id, error }
+            TransportEvent::ListenerError { listener_id, error } => {
+                TransportEvent::ListenerError { listener_id, error }
             }
             TransportEvent::ListenerClosed {
                 listener_id,
@@ -338,6 +332,9 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
         }
     }
 
+    /// In case this [`TransportEvent`] is an [`ListenerError`](TransportEvent::ListenerError),
+    /// or [`ListenerClosed`](TransportEvent::ListenerClosed) apply the given function to the
+    /// error and produce another transport event based on the function's result.
     pub fn map_err<E>(self, map_err: impl FnOnce(TErr) -> E) -> TransportEvent<TUpgr, E> {
         match self {
             TransportEvent::Incoming {
@@ -365,7 +362,7 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
                 listen_addr,
                 listener_id,
             },
-            TransportEvent::Error { listener_id, error } => TransportEvent::Error {
+            TransportEvent::ListenerError { listener_id, error } => TransportEvent::ListenerError {
                 listener_id,
                 error: map_err(error),
             },
@@ -379,16 +376,17 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
         }
     }
 
-    /// Returns `true` if this is an `Upgrade` listener event.
+    /// Returns `true` if this is an [`Incoming`](TransportEvent::Incoming) transport event.
     pub fn is_upgrade(&self) -> bool {
         matches!(self, TransportEvent::Incoming { .. })
     }
 
-    /// Try to turn this listener event into upgrade parts.
+    /// Try to turn this transport event into the upgrade parts of the
+    /// incoming connection.
     ///
-    /// Returns `None` if the event is not actually an upgrade,
+    /// Returns `None` if the event is not actually an incoming connection,
     /// otherwise the upgrade and the remote address.
-    pub fn into_upgrade(self) -> Option<(TUpgr, Multiaddr)> {
+    pub fn into_incoming(self) -> Option<(TUpgr, Multiaddr)> {
         if let TransportEvent::Incoming {
             upgrade,
             send_back_addr,
@@ -401,14 +399,14 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
         }
     }
 
-    /// Returns `true` if this is a `NewAddress` listener event.
+    /// Returns `true` if this is a [`TransportEvent::NewAddress`].
     pub fn is_new_address(&self) -> bool {
         matches!(self, TransportEvent::NewAddress { .. })
     }
 
-    /// Try to turn this listener event into the `NewAddress` part.
+    /// Try to turn this transport event into the new `Multiaddr`.
     ///
-    /// Returns `None` if the event is not actually a `NewAddress`,
+    /// Returns `None` if the event is not actually a [`TransportEvent::NewAddress`],
     /// otherwise the address.
     pub fn into_new_address(self) -> Option<Multiaddr> {
         if let TransportEvent::NewAddress { listen_addr, .. } = self {
@@ -418,14 +416,14 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
         }
     }
 
-    /// Returns `true` if this is an `AddressExpired` listener event.
+    /// Returns `true` if this is an [`TransportEvent::AddressExpired`].
     pub fn is_address_expired(&self) -> bool {
         matches!(self, TransportEvent::AddressExpired { .. })
     }
 
-    /// Try to turn this listener event into the `AddressExpired` part.
+    /// Try to turn this transport event into the expire `Multiaddr`.
     ///
-    /// Returns `None` if the event is not actually a `AddressExpired`,
+    /// Returns `None` if the event is not actually a [`TransportEvent::AddressExpired`],
     /// otherwise the address.
     pub fn into_address_expired(self) -> Option<Multiaddr> {
         if let TransportEvent::AddressExpired { listen_addr, .. } = self {
@@ -435,17 +433,17 @@ impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
         }
     }
 
-    /// Returns `true` if this is an `Error` listener event.
-    pub fn is_error(&self) -> bool {
-        matches!(self, TransportEvent::Error { .. })
+    /// Returns `true` if this is an [`TransportEvent::ListenerError`] transport event.
+    pub fn is_listener_error(&self) -> bool {
+        matches!(self, TransportEvent::ListenerError { .. })
     }
 
-    /// Try to turn this listener event into the `Error` part.
+    /// Try to turn this transport event into the listener error.
     ///
-    /// Returns `None` if the event is not actually a `Error`,
+    /// Returns `None` if the event is not actually a [`TransportEvent::ListenerError`]`,
     /// otherwise the error.
-    pub fn into_error(self) -> Option<TErr> {
-        if let TransportEvent::Error { error, .. } = self {
+    pub fn into_listener_error(self) -> Option<TErr> {
+        if let TransportEvent::ListenerError { error, .. } = self {
             Some(error)
         } else {
             None
@@ -489,8 +487,8 @@ impl<TUpgr, TErr: fmt::Debug> fmt::Debug for TransportEvent<TUpgr, TErr> {
                 .field("listener_id", listener_id)
                 .field("reason", reason)
                 .finish(),
-            TransportEvent::Error { listener_id, error } => f
-                .debug_struct("TransportEvent::Error")
+            TransportEvent::ListenerError { listener_id, error } => f
+                .debug_struct("TransportEvent::ListenerError")
                 .field("listener_id", listener_id)
                 .field("error", error)
                 .finish(),

core/src/transport/and_then.rs

@@ -27,7 +27,7 @@ use futures::{future::Either, prelude::*};
 use multiaddr::Multiaddr;
 use std::{error, marker::PhantomPinned, pin::Pin, task::Context, task::Poll};
 
-/// See the `Transport::and_then` method.
+/// See the [`Transport::and_then`] method.
 #[pin_project::pin_project]
 #[derive(Debug, Clone)]
 pub struct AndThen<T, C> {

core/src/transport/memory.rs

@@ -323,7 +323,7 @@ pub struct Listener {
     addr: Multiaddr,
     /// Receives incoming connections.
     receiver: ChannelReceiver,
-    /// Generate `TransportEvent::NewAddress` to inform about our listen address.
+    /// Generate [`TransportEvent::NewAddress`] to inform about our listen address.
     tell_listen_addr: bool,
 }
 
@@ -544,7 +544,7 @@ mod tests {
             t1.listen_on(t1_addr.clone()).unwrap();
             let upgrade = loop {
                 let event = t1.select_next_some().await;
-                if let Some(upgrade) = event.into_upgrade() {
+                if let Some(upgrade) = event.into_incoming() {
                     break upgrade;
                 }
             };

core/src/transport/upgrade.rs

@@ -498,7 +498,7 @@ where
     C: AsyncRead + AsyncWrite + Unpin,
 {
     future: Pin<Box<F>>,
-    upgrade: future::Either<Option<U>, (Option<PeerId>, OutboundUpgradeApply<C, U>)>,
+    upgrade: future::Either<Option<U>, (PeerId, OutboundUpgradeApply<C, U>)>,
 }
 
 impl<F, U, C, D> Future for DialUpgradeFuture<F, U, C>
@@ -527,18 +527,15 @@ where
                     let u = up
                         .take()
                         .expect("DialUpgradeFuture is constructed with Either::Left(Some).");
-                    future::Either::Right((Some(i), apply_outbound(c, u, upgrade::Version::V1)))
+                    future::Either::Right((i, apply_outbound(c, u, upgrade::Version::V1)))
                 }
-                future::Either::Right((ref mut i, ref mut up)) => {
+                future::Either::Right((i, ref mut up)) => {
                     let d = match ready!(
                         Future::poll(Pin::new(up), cx).map_err(TransportUpgradeError::Upgrade)
                     ) {
                         Ok(d) => d,
                         Err(err) => return Poll::Ready(Err(err)),
                     };
-                    let i = i
-                        .take()
-                        .expect("DialUpgradeFuture polled after completion.");
                     return Poll::Ready(Ok((i, d)));
                 }
             }
@@ -560,7 +557,7 @@ where
     U: InboundUpgrade<Negotiated<C>>,
 {
     future: Pin<Box<F>>,
-    upgrade: future::Either<Option<U>, (Option<PeerId>, InboundUpgradeApply<C, U>)>,
+    upgrade: future::Either<Option<U>, (PeerId, InboundUpgradeApply<C, U>)>,
 }
 
 impl<F, U, C, D> Future for ListenerUpgradeFuture<F, U, C>
@@ -589,18 +586,15 @@ where
                     let u = up
                         .take()
                         .expect("ListenerUpgradeFuture is constructed with Either::Left(Some).");
-                    future::Either::Right((Some(i), apply_inbound(c, u)))
+                    future::Either::Right((i, apply_inbound(c, u)))
                 }
-                future::Either::Right((ref mut i, ref mut up)) => {
+                future::Either::Right((i, ref mut up)) => {
                     let d = match ready!(TryFuture::try_poll(Pin::new(up), cx)
                         .map_err(TransportUpgradeError::Upgrade))
                     {
                         Ok(v) => v,
                         Err(err) => return Poll::Ready(Err(err)),
                     };
-                    let i = i
-                        .take()
-                        .expect("ListenerUpgradeFuture polled after completion.");
                     return Poll::Ready(Ok((i, d)));
                 }
             }

core/tests/transport_upgrade.rs

@@ -125,7 +125,7 @@ fn upgrade_pipeline() {
     let server = async move {
         loop {
             let (upgrade, _send_back_addr) =
-                match listener_transport.select_next_some().await.into_upgrade() {
+                match listener_transport.select_next_some().await.into_incoming() {
                     Some(u) => u,
                     None => continue,
                 };

muxers/mplex/benches/split_send_size.rs

@@ -59,13 +59,13 @@ fn prepare(c: &mut Criterion) {
     let tcp_addr = multiaddr![Ip4(std::net::Ipv4Addr::new(127, 0, 0, 1)), Tcp(0u16)];
     for &size in BENCH_SIZES.iter() {
         tcp.throughput(Throughput::Bytes(payload.len() as u64));
-        let mut receiver_trans = tcp_transport(size);
-        let mut sender_trans = tcp_transport(size);
+        let mut receiver_transport = tcp_transport(size);
+        let mut sender_transport = tcp_transport(size);
         tcp.bench_function(format!("{}", size), |b| {
             b.iter(|| {
                 run(
-                    black_box(&mut receiver_trans),
-                    black_box(&mut sender_trans),
+                    black_box(&mut receiver_transport),
+                    black_box(&mut sender_transport),
                     black_box(&payload),
                     black_box(&tcp_addr),
                 )
@@ -78,13 +78,13 @@ fn prepare(c: &mut Criterion) {
     let mem_addr = multiaddr![Memory(0u64)];
     for &size in BENCH_SIZES.iter() {
         mem.throughput(Throughput::Bytes(payload.len() as u64));
-        let mut receiver_trans = mem_transport(size);
-        let mut sender_trans = mem_transport(size);
+        let mut receiver_transport = mem_transport(size);
+        let mut sender_transport = mem_transport(size);
         mem.bench_function(format!("{}", size), |b| {
             b.iter(|| {
                 run(
-                    black_box(&mut receiver_trans),
-                    black_box(&mut sender_trans),
+                    black_box(&mut receiver_transport),
+                    black_box(&mut sender_transport),
                     black_box(&payload),
                     black_box(&mem_addr),
                 )
@@ -134,7 +134,7 @@ fn run(
                         }
                     }
                 }
-                _ => panic!("Unexpected listener event"),
+                _ => panic!("Unexpected transport event"),
             }
         }
     };

muxers/mplex/tests/async_write.rs

@@ -54,7 +54,7 @@ fn async_write() {
             .next()
             .await
             .expect("some event")
-            .into_upgrade()
+            .into_incoming()
             .unwrap()
             .0
             .await