refactor(swarm)!: don't be generic over Transport

misc/metrics/src/swarm.rs

@@ -370,10 +370,8 @@ enum PendingInboundConnectionError {
     ConnectionLimit,
 }
 
-impl<TTransErr> From<&libp2p_swarm::PendingInboundConnectionError<TTransErr>>
-    for PendingInboundConnectionError
-{
-    fn from(error: &libp2p_swarm::PendingInboundConnectionError<TTransErr>) -> Self {
+impl From<&libp2p_swarm::PendingInboundConnectionError> for PendingInboundConnectionError {
+    fn from(error: &libp2p_swarm::PendingInboundConnectionError) -> Self {
         match error {
             libp2p_swarm::PendingInboundConnectionError::WrongPeerId { .. } => {
                 PendingInboundConnectionError::WrongPeerId

swarm/CHANGELOG.md

@@ -7,9 +7,13 @@
 
 - Add `estblished_in` to `SwarmEvent::ConnectionEstablished`. See [PR 3134].
 
+- Remove type parameter from `PendingOutboundConnectionError` and `PendingInboundConnectionError`.
+  These two types are always used with `std::io::Error`. See [PR XXXX.
+
 [PR 3170]: https://github.com/libp2p/rust-libp2p/pull/3170
 [PR 3134]: https://github.com/libp2p/rust-libp2p/pull/3134
 [PR 3153]: https://github.com/libp2p/rust-libp2p/pull/3153
+[PR XXXX]: https://github.com/libp2p/rust-libp2p/pull/XXXX
 
 # 0.41.1
 

swarm/src/connection/error.rs

@@ -76,12 +76,11 @@ impl<THandlerErr> From<io::Error> for ConnectionError<THandlerErr> {
 /// Note: Addresses for an outbound connection are dialed in parallel. Thus, compared to
 /// [`PendingInboundConnectionError`], one or more [`TransportError`]s can occur for a single
 /// connection.
-pub type PendingOutboundConnectionError<TTransErr> =
-    PendingConnectionError<Vec<(Multiaddr, TransportError<TTransErr>)>>;
+pub type PendingOutboundConnectionError =
+    PendingConnectionError<Vec<(Multiaddr, TransportError<io::Error>)>>;
 
 /// Errors that can occur in the context of a pending incoming `Connection`.
-pub type PendingInboundConnectionError<TTransErr> =
-    PendingConnectionError<TransportError<TTransErr>>;
+pub type PendingInboundConnectionError = PendingConnectionError<TransportError<io::Error>>;
 
 /// Errors that can occur in the context of a pending `Connection`.
 #[derive(Debug)]

swarm/src/connection/pool.rs

@@ -26,7 +26,7 @@ use crate::{
         Connected, ConnectionError, ConnectionLimit, IncomingInfo, PendingConnectionError,
         PendingInboundConnectionError, PendingOutboundConnectionError,
     },
-    transport::{Transport, TransportError},
+    transport::TransportError,
     ConnectedPoint, ConnectionHandler, Executor, IntoConnectionHandler, Multiaddr, PeerId,
 };
 use concurrent_dial::ConcurrentDial;
@@ -79,9 +79,8 @@ impl ExecSwitch {
 }
 
 /// A connection `Pool` manages a set of connections for each peer.
-pub struct Pool<THandler, TTrans>
+pub struct Pool<THandler>
 where
-    TTrans: Transport,
     THandler: IntoConnectionHandler,
 {
     local_id: PeerId,
@@ -124,10 +123,10 @@ where
 
     /// Sender distributed to pending tasks for reporting events back
     /// to the pool.
-    pending_connection_events_tx: mpsc::Sender<task::PendingConnectionEvent<TTrans>>,
+    pending_connection_events_tx: mpsc::Sender<task::PendingConnectionEvent>,
 
     /// Receiver for events reported from pending tasks.
-    pending_connection_events_rx: mpsc::Receiver<task::PendingConnectionEvent<TTrans>>,
+    pending_connection_events_rx: mpsc::Receiver<task::PendingConnectionEvent>,
 
     /// Sender distributed to established tasks for reporting events back
     /// to the pool.
@@ -213,7 +212,7 @@ impl<THandler> PendingConnection<THandler> {
     }
 }
 
-impl<THandler: IntoConnectionHandler, TTrans: Transport> fmt::Debug for Pool<THandler, TTrans> {
+impl<THandler: IntoConnectionHandler> fmt::Debug for Pool<THandler> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
         f.debug_struct("Pool")
             .field("counters", &self.counters)
@@ -223,10 +222,7 @@ impl<THandler: IntoConnectionHandler, TTrans: Transport> fmt::Debug for Pool<THa
 
 /// Event that can happen on the `Pool`.
 #[derive(Debug)]
-pub enum PoolEvent<THandler: IntoConnectionHandler, TTrans>
-where
-    TTrans: Transport,
-{
+pub enum PoolEvent<THandler: IntoConnectionHandler> {
     /// A new connection has been established.
     ConnectionEstablished {
         id: ConnectionId,
@@ -239,7 +235,7 @@ where
         /// [`Some`] when the new connection is an outgoing connection.
         /// Addresses are dialed in parallel. Contains the addresses and errors
         /// of dial attempts that failed before the one successful dial.
-        concurrent_dial_errors: Option<Vec<(Multiaddr, TransportError<TTrans::Error>)>>,
+        concurrent_dial_errors: Option<Vec<(Multiaddr, TransportError<std::io::Error>)>>,
         /// How long it took to establish this connection.
         established_in: std::time::Duration,
     },
@@ -272,7 +268,7 @@ where
         /// The ID of the failed connection.
         id: ConnectionId,
         /// The error that occurred.
-        error: PendingOutboundConnectionError<TTrans::Error>,
+        error: PendingOutboundConnectionError,
         /// The handler that was supposed to handle the connection.
         handler: THandler,
         /// The (expected) peer of the failed connection.
@@ -288,7 +284,7 @@ where
         /// Local connection address.
         local_addr: Multiaddr,
         /// The error that occurred.
-        error: PendingInboundConnectionError<TTrans::Error>,
+        error: PendingInboundConnectionError,
         /// The handler that was supposed to handle the connection.
         handler: THandler,
     },
@@ -312,10 +308,9 @@ where
     },
 }
 
-impl<THandler, TTrans> Pool<THandler, TTrans>
+impl<THandler> Pool<THandler>
 where
     THandler: IntoConnectionHandler,
-    TTrans: Transport,
 {
     /// Creates a new empty `Pool`.
     pub fn new(local_id: PeerId, config: PoolConfig, limits: ConnectionLimits) -> Self {
@@ -429,12 +424,9 @@ where
     }
 }
 
-impl<THandler, TTrans> Pool<THandler, TTrans>
+impl<THandler> Pool<THandler>
 where
     THandler: IntoConnectionHandler,
-    TTrans: Transport + 'static,
-    TTrans::Output: Send + 'static,
-    TTrans::Error: Send + 'static,
 {
     /// Adds a pending outgoing connection to the pool in the form of a `Future`
     /// that establishes and negotiates the connection.
@@ -448,22 +440,15 @@ where
                 'static,
                 (
                     Multiaddr,
-                    Result<
-                        <TTrans as Transport>::Output,
-                        TransportError<<TTrans as Transport>::Error>,
-                    >,
+                    Result<(PeerId, StreamMuxerBox), TransportError<std::io::Error>>,
                 ),
             >,
         >,
         peer: Option<PeerId>,
         handler: THandler,
         role_override: Endpoint,
         dial_concurrency_factor_override: Option<NonZeroU8>,
-    ) -> Result<ConnectionId, (ConnectionLimit, THandler)>
-    where
-        TTrans: Send,
-        TTrans::Dial: Send + 'static,
-    {
+    ) -> Result<ConnectionId, (ConnectionLimit, THandler)> {
         if let Err(limit) = self.counters.check_max_pending_outgoing() {
             return Err((limit, handler));
         };
@@ -515,7 +500,7 @@ where
         info: IncomingInfo<'_>,
     ) -> Result<ConnectionId, (ConnectionLimit, THandler)>
     where
-        TFut: Future<Output = Result<TTrans::Output, TTrans::Error>> + Send + 'static,
+        TFut: Future<Output = Result<(PeerId, StreamMuxerBox), std::io::Error>> + Send + 'static,
     {
         let endpoint = info.create_connected_point();
 
@@ -552,9 +537,8 @@ where
     }
 
     /// Polls the connection pool for events.
-    pub fn poll(&mut self, cx: &mut Context<'_>) -> Poll<PoolEvent<THandler, TTrans>>
+    pub fn poll(&mut self, cx: &mut Context<'_>) -> Poll<PoolEvent<THandler>>
     where
-        TTrans: Transport<Output = (PeerId, StreamMuxerBox)>,
         THandler: IntoConnectionHandler + 'static,
         THandler::Handler: ConnectionHandler + Send,
         <THandler::Handler as ConnectionHandler>::OutboundOpenInfo: Send,
@@ -677,7 +661,7 @@ where
                         ),
                     };
 
-                    let error: Result<(), PendingInboundConnectionError<_>> = self
+                    let error = self
                         .counters
                         // Check general established connection limit.
                         .check_max_established(&endpoint)

swarm/src/connection/pool/task.rs

@@ -26,7 +26,7 @@ use crate::{
     connection::{
         self, ConnectionError, PendingInboundConnectionError, PendingOutboundConnectionError,
     },
-    transport::{Transport, TransportError},
+    transport::TransportError,
     ConnectionHandler, Multiaddr, PeerId,
 };
 use futures::{
@@ -35,6 +35,8 @@ use futures::{
     SinkExt, StreamExt,
 };
 use libp2p_core::connection::ConnectionId;
+use libp2p_core::muxing::StreamMuxerBox;
+use libp2p_core::transport::Boxed;
 use std::pin::Pin;
 use void::Void;
 
@@ -48,26 +50,19 @@ pub enum Command<T> {
     Close,
 }
 
-#[derive(Debug)]
-pub enum PendingConnectionEvent<TTrans>
-where
-    TTrans: Transport,
-{
+pub enum PendingConnectionEvent {
     ConnectionEstablished {
         id: ConnectionId,
-        output: TTrans::Output,
+        output: (PeerId, StreamMuxerBox),
         /// [`Some`] when the new connection is an outgoing connection.
         /// Addresses are dialed in parallel. Contains the addresses and errors
         /// of dial attempts that failed before the one successful dial.
-        outgoing: Option<(Multiaddr, Vec<(Multiaddr, TransportError<TTrans::Error>)>)>,
+        outgoing: Option<(Multiaddr, Vec<(Multiaddr, TransportError<std::io::Error>)>)>,
     },
     /// A pending connection failed.
     PendingFailed {
         id: ConnectionId,
-        error: Either<
-            PendingOutboundConnectionError<TTrans::Error>,
-            PendingInboundConnectionError<TTrans::Error>,
-        >,
+        error: Either<PendingOutboundConnectionError, PendingInboundConnectionError>,
     },
 }
 
@@ -97,14 +92,12 @@ pub enum EstablishedConnectionEvent<THandler: ConnectionHandler> {
     },
 }
 
-pub async fn new_for_pending_outgoing_connection<TTrans>(
+pub async fn new_for_pending_outgoing_connection(
     connection_id: ConnectionId,
-    dial: ConcurrentDial<TTrans>,
+    dial: ConcurrentDial<Boxed<(PeerId, StreamMuxerBox)>>,
     abort_receiver: oneshot::Receiver<Void>,
-    mut events: mpsc::Sender<PendingConnectionEvent<TTrans>>,
-) where
-    TTrans: Transport,
-{
+    mut events: mpsc::Sender<PendingConnectionEvent>,
+) {
     match futures::future::select(abort_receiver, Box::pin(dial)).await {
         Either::Left((Err(oneshot::Canceled), _)) => {
             let _ = events
@@ -135,14 +128,13 @@ pub async fn new_for_pending_outgoing_connection<TTrans>(
     }
 }
 
-pub async fn new_for_pending_incoming_connection<TFut, TTrans>(
+pub async fn new_for_pending_incoming_connection<TFut>(
     connection_id: ConnectionId,
     future: TFut,
     abort_receiver: oneshot::Receiver<Void>,
-    mut events: mpsc::Sender<PendingConnectionEvent<TTrans>>,
+    mut events: mpsc::Sender<PendingConnectionEvent>,
 ) where
-    TTrans: Transport,
-    TFut: Future<Output = Result<TTrans::Output, TTrans::Error>> + Send + 'static,
+    TFut: Future<Output = Result<(PeerId, StreamMuxerBox), std::io::Error>> + Send + 'static,
 {
     match futures::future::select(abort_receiver, Box::pin(future)).await {
         Either::Left((Err(oneshot::Canceled), _)) => {

swarm/src/lib.rs

@@ -234,7 +234,7 @@ pub enum SwarmEvent<TBehaviourOutEvent, THandlerErr> {
         /// Address used to send back data to the remote.
         send_back_addr: Multiaddr,
         /// The error that happened.
-        error: PendingInboundConnectionError<io::Error>,
+        error: PendingInboundConnectionError,
     },
     /// Outgoing connection attempt failed.
     OutgoingConnectionError {
@@ -305,7 +305,7 @@ where
     transport: transport::Boxed<(PeerId, StreamMuxerBox)>,
 
     /// The nodes currently active.
-    pool: Pool<THandler<TBehaviour>, transport::Boxed<(PeerId, StreamMuxerBox)>>,
+    pool: Pool<THandler<TBehaviour>>,
 
     /// The local peer ID.
     local_peer_id: PeerId,
@@ -802,7 +802,7 @@ where
 
     fn handle_pool_event(
         &mut self,
-        event: PoolEvent<THandler<TBehaviour>, transport::Boxed<(PeerId, StreamMuxerBox)>>,
+        event: PoolEvent<THandler<TBehaviour>>,
     ) -> Option<SwarmEvent<TBehaviour::OutEvent, THandlerErr<TBehaviour>>> {
         match event {
             PoolEvent::ConnectionEstablished {
@@ -1199,7 +1199,7 @@ where
                         }
                     }
                     PendingNotifyHandler::Any(ids) => {
-                        match notify_any::<_, _, TBehaviour>(ids, &mut this.pool, event, cx) {
+                        match notify_any::<_, TBehaviour>(ids, &mut this.pool, event, cx) {
                             None => continue,
                             Some((event, ids)) => {
                                 let handler = PendingNotifyHandler::Any(ids);
@@ -1308,15 +1308,13 @@ fn notify_one<THandlerInEvent>(
 ///
 /// Returns `None` if either all connections are closing or the event
 /// was successfully sent to a handler, in either case the event is consumed.
-fn notify_any<TTrans, THandler, TBehaviour>(
+fn notify_any<THandler, TBehaviour>(
     ids: SmallVec<[ConnectionId; 10]>,
-    pool: &mut Pool<THandler, TTrans>,
+    pool: &mut Pool<THandler>,
     event: THandlerInEvent<TBehaviour>,
     cx: &mut Context<'_>,
 ) -> Option<(THandlerInEvent<TBehaviour>, SmallVec<[ConnectionId; 10]>)>
 where
-    TTrans: Transport,
-    TTrans::Error: Send + 'static,
     TBehaviour: NetworkBehaviour,
     THandler: IntoConnectionHandler,
     THandler::Handler: ConnectionHandler<
@@ -1641,8 +1639,8 @@ pub enum DialError {
     Transport(Vec<(Multiaddr, TransportError<io::Error>)>),
 }
 
-impl From<PendingOutboundConnectionError<io::Error>> for DialError {
-    fn from(error: PendingOutboundConnectionError<io::Error>) -> Self {
+impl From<PendingOutboundConnectionError> for DialError {
+    fn from(error: PendingOutboundConnectionError) -> Self {
         match error {
             PendingConnectionError::ConnectionLimit(limit) => DialError::ConnectionLimit(limit),
             PendingConnectionError::Aborted => DialError::Aborted,