feat: Add Transport::box_multiplexed

core/src/either.rs

@@ -255,6 +255,24 @@ where
     }
 }
 
+impl<T, A, B> From<EitherOutput<(T, A), (T, B)>> for (T, EitherOutput<A, B>) {
+    fn from(either_output: EitherOutput<(T, A), (T, B)>) -> Self {
+        match either_output {
+            EitherOutput::First((t, a)) => (t, EitherOutput::First(a)),
+            EitherOutput::Second((t, b)) => (t, EitherOutput::Second(b)),
+        }
+    }
+}
+
+impl<T, A, B> From<EitherOutput<(A, T), (B, T)>> for (EitherOutput<A, B>, T) {
+    fn from(either_output: EitherOutput<(A, T), (B, T)>) -> Self {
+        match either_output {
+            EitherOutput::First((a, t)) => (EitherOutput::First(a), t),
+            EitherOutput::Second((b, t)) => (EitherOutput::Second(b), t),
+        }
+    }
+}
+
 /// Implements `Future` and dispatches all method calls to either `First` or `Second`.
 #[pin_project(project = EitherFutureProj)]
 #[derive(Debug, Copy, Clone)]

core/src/transport.rs

@@ -46,8 +46,9 @@ pub mod upgrade;
 mod boxed;
 mod optional;
 
-use crate::ConnectedPoint;
+use crate::{muxing::StreamMuxerBox, ConnectedPoint, PeerId, StreamMuxer};
 
+use self::boxed::boxed;
 pub use self::boxed::Boxed;
 pub use self::choice::OrTransport;
 pub use self::memory::MemoryTransport;
@@ -236,6 +237,25 @@ pub trait Transport {
     {
         upgrade::Builder::new(self, version)
     }
+
+    /// Box a multiplexed transport, including the inner muxer
+    /// and all errors.
+    fn box_multiplexed<M>(self) -> Boxed<(PeerId, StreamMuxerBox)>
+    where
+        Self: Sized + Send + Unpin + 'static,
+        Self::Dial: Send + 'static,
+        Self::ListenerUpgrade: Send + 'static,
+        Self::Error: Send + Sync,
+        Self::Output: Into<(PeerId, M)>,
+        M: StreamMuxer + Send + 'static,
+        M::Substream: Send + 'static,
+        M::Error: Send + Sync + 'static,
+    {
+        boxed(self.map(|o, _| {
+            let (i, m) = o.into();
+            (i, StreamMuxerBox::new(m))
+        }))
+    }
 }
 
 /// The ID of a single listener.

core/src/transport/upgrade.rs

@@ -26,8 +26,8 @@ use crate::{
     connection::ConnectedPoint,
     muxing::{StreamMuxer, StreamMuxerBox},
     transport::{
-        and_then::AndThen, boxed::boxed, timeout::TransportTimeout, ListenerId, Transport,
-        TransportError, TransportEvent,
+        and_then::AndThen, timeout::TransportTimeout, ListenerId, Transport, TransportError,
+        TransportEvent,
     },
     upgrade::{
         self, apply_inbound, apply_outbound, InboundUpgrade, InboundUpgradeApply, OutboundUpgrade,
@@ -303,7 +303,7 @@ impl<T> Multiplexed<T> {
         M::Substream: Send + 'static,
         M::Error: Send + Sync + 'static,
     {
-        boxed(self.map(|(i, m), _| (i, StreamMuxerBox::new(m))))
+        self.box_multiplexed()
     }
 
     /// Adds a timeout to the setup and protocol upgrade process for all

transports/quic/tests/smoke.rs

@@ -4,7 +4,6 @@ use futures::channel::{mpsc, oneshot};
 use futures::future::{poll_fn, Either};
 use futures::stream::StreamExt;
 use futures::{future, AsyncReadExt, AsyncWriteExt, FutureExt, SinkExt};
-use libp2p_core::either::EitherOutput;
 use libp2p_core::muxing::{StreamMuxerBox, StreamMuxerExt, SubstreamBox};
 use libp2p_core::transport::{Boxed, OrTransport, TransportEvent};
 use libp2p_core::{multiaddr::Protocol, upgrade, Multiaddr, PeerId, Transport};
@@ -129,12 +128,7 @@ fn new_tcp_quic_transport() -> (PeerId, Boxed<(PeerId, StreamMuxerBox)>) {
         )
         .multiplex(yamux::YamuxConfig::default());
 
-    let transport = OrTransport::new(quic_transport, tcp_transport)
-        .map(|either_output, _| match either_output {
-            EitherOutput::First((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
-            EitherOutput::Second((peer_id, muxer)) => (peer_id, StreamMuxerBox::new(muxer)),
-        })
-        .boxed();
+    let transport = OrTransport::new(quic_transport, tcp_transport).box_multiplexed();
 
     (peer_id, transport)
 }
@@ -394,9 +388,7 @@ fn create_transport<P: Provider>(
     let peer_id = keypair.public().to_peer_id();
     let mut config = quic::Config::new(&keypair);
     with_config(&mut config);
-    let transport = quic::GenTransport::<P>::new(config)
-        .map(|(p, c), _| (p, StreamMuxerBox::new(c)))
-        .boxed();
+    let transport = quic::GenTransport::<P>::new(config).box_multiplexed();
 
     (peer_id, transport)
 }

transports/webrtc/examples/listen_ping.rs

@@ -1,7 +1,6 @@
 use anyhow::Result;
 use futures::StreamExt;
 use libp2p_core::identity;
-use libp2p_core::muxing::StreamMuxerBox;
 use libp2p_core::Transport;
 use libp2p_ping as ping;
 use libp2p_swarm::{keep_alive, NetworkBehaviour, Swarm};
@@ -29,12 +28,8 @@ fn create_swarm() -> Result<Swarm<Behaviour>> {
         libp2p_webrtc::tokio::Certificate::generate(&mut thread_rng())?,
     );
 
-    let transport = transport
-        .map(|(peer_id, conn), _| (peer_id, StreamMuxerBox::new(conn)))
-        .boxed();
-
     Ok(Swarm::with_tokio_executor(
-        transport,
+        transport.box_multiplexed(),
         Behaviour::default(),
         peer_id,
     ))

transports/webrtc/tests/smoke.rs

@@ -73,8 +73,7 @@ fn create_transport() -> (PeerId, Boxed<(PeerId, StreamMuxerBox)>) {
         keypair,
         webrtc::tokio::Certificate::generate(&mut thread_rng()).unwrap(),
     )
-    .map(|(p, c), _| (p, StreamMuxerBox::new(c)))
-    .boxed();
+    .box_multiplexed();
 
     (peer_id, transport)
 }