core/muxing: Redesign StreamMuxer trait

core/CHANGELOG.md

@@ -4,12 +4,13 @@
 - Remove `StreamMuxer::flush_all`. See [PR 2669].
 - Rename `StreamMuxer::close` to `StreamMuxer::poll_close`. See [PR 2666].
 - Remove deprecated function `StreamMuxer::is_remote_acknowledged`. See [PR 2665].
+- Re-design `StreamMuxer` trait to take `&mut self` and remove all associated types but `Substream`. See [PR 2648].
 
 [PR 2529]: https://github.com/libp2p/rust-libp2p/pull/2529
 [PR 2666]: https://github.com/libp2p/rust-libp2p/pull/2666
 [PR 2665]: https://github.com/libp2p/rust-libp2p/pull/2665
 [PR 2669]: https://github.com/libp2p/rust-libp2p/pull/2669
-
+[PR 2648]: https://github.com/libp2p/rust-libp2p/pull/2648
 
 # 0.32.1
 

core/src/either.rs

@@ -19,7 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{
-    muxing::{StreamMuxer, StreamMuxerEvent},
+    muxing::{OutboundSubstreamId, StreamMuxer, StreamMuxerEvent},
     transport::{ListenerEvent, Transport, TransportError},
     Multiaddr, ProtocolName,
 };
@@ -202,165 +202,36 @@ where
     B: StreamMuxer,
 {
     type Substream = EitherOutput<A::Substream, B::Substream>;
-    type OutboundSubstream = EitherOutbound<A, B>;
-    type Error = io::Error;
 
-    fn poll_event(
-        &self,
+    fn poll(
+        &mut self,
         cx: &mut Context<'_>,
-    ) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>> {
-        match self {
-            EitherOutput::First(inner) => inner.poll_event(cx).map(|result| {
-                result.map_err(|e| e.into()).map(|event| match event {
-                    StreamMuxerEvent::AddressChange(addr) => StreamMuxerEvent::AddressChange(addr),
-                    StreamMuxerEvent::InboundSubstream(substream) => {
-                        StreamMuxerEvent::InboundSubstream(EitherOutput::First(substream))
-                    }
-                })
-            }),
-            EitherOutput::Second(inner) => inner.poll_event(cx).map(|result| {
-                result.map_err(|e| e.into()).map(|event| match event {
-                    StreamMuxerEvent::AddressChange(addr) => StreamMuxerEvent::AddressChange(addr),
-                    StreamMuxerEvent::InboundSubstream(substream) => {
-                        StreamMuxerEvent::InboundSubstream(EitherOutput::Second(substream))
-                    }
-                })
-            }),
-        }
-    }
-
-    fn open_outbound(&self) -> Self::OutboundSubstream {
+    ) -> Poll<io::Result<StreamMuxerEvent<Self::Substream>>> {
         match self {
-            EitherOutput::First(inner) => EitherOutbound::A(inner.open_outbound()),
-            EitherOutput::Second(inner) => EitherOutbound::B(inner.open_outbound()),
-        }
-    }
-
-    fn poll_outbound(
-        &self,
-        cx: &mut Context<'_>,
-        substream: &mut Self::OutboundSubstream,
-    ) -> Poll<Result<Self::Substream, Self::Error>> {
-        match (self, substream) {
-            (EitherOutput::First(ref inner), EitherOutbound::A(ref mut substream)) => inner
-                .poll_outbound(cx, substream)
-                .map(|p| p.map(EitherOutput::First))
-                .map_err(|e| e.into()),
-            (EitherOutput::Second(ref inner), EitherOutbound::B(ref mut substream)) => inner
-                .poll_outbound(cx, substream)
-                .map(|p| p.map(EitherOutput::Second))
-                .map_err(|e| e.into()),
-            _ => panic!("Wrong API usage"),
-        }
-    }
-
-    fn destroy_outbound(&self, substream: Self::OutboundSubstream) {
-        match self {
-            EitherOutput::First(inner) => match substream {
-                EitherOutbound::A(substream) => inner.destroy_outbound(substream),
-                _ => panic!("Wrong API usage"),
-            },
-            EitherOutput::Second(inner) => match substream {
-                EitherOutbound::B(substream) => inner.destroy_outbound(substream),
-                _ => panic!("Wrong API usage"),
-            },
-        }
-    }
-
-    fn read_substream(
-        &self,
-        cx: &mut Context<'_>,
-        sub: &mut Self::Substream,
-        buf: &mut [u8],
-    ) -> Poll<Result<usize, Self::Error>> {
-        match (self, sub) {
-            (EitherOutput::First(ref inner), EitherOutput::First(ref mut sub)) => {
-                inner.read_substream(cx, sub, buf).map_err(|e| e.into())
-            }
-            (EitherOutput::Second(ref inner), EitherOutput::Second(ref mut sub)) => {
-                inner.read_substream(cx, sub, buf).map_err(|e| e.into())
-            }
-            _ => panic!("Wrong API usage"),
-        }
-    }
-
-    fn write_substream(
-        &self,
-        cx: &mut Context<'_>,
-        sub: &mut Self::Substream,
-        buf: &[u8],
-    ) -> Poll<Result<usize, Self::Error>> {
-        match (self, sub) {
-            (EitherOutput::First(ref inner), EitherOutput::First(ref mut sub)) => {
-                inner.write_substream(cx, sub, buf).map_err(|e| e.into())
-            }
-            (EitherOutput::Second(ref inner), EitherOutput::Second(ref mut sub)) => {
-                inner.write_substream(cx, sub, buf).map_err(|e| e.into())
-            }
-            _ => panic!("Wrong API usage"),
-        }
-    }
-
-    fn flush_substream(
-        &self,
-        cx: &mut Context<'_>,
-        sub: &mut Self::Substream,
-    ) -> Poll<Result<(), Self::Error>> {
-        match (self, sub) {
-            (EitherOutput::First(ref inner), EitherOutput::First(ref mut sub)) => {
-                inner.flush_substream(cx, sub).map_err(|e| e.into())
-            }
-            (EitherOutput::Second(ref inner), EitherOutput::Second(ref mut sub)) => {
-                inner.flush_substream(cx, sub).map_err(|e| e.into())
+            EitherOutput::First(inner) => {
+                inner.poll(cx).map_ok(|e| e.map_stream(EitherOutput::First))
             }
-            _ => panic!("Wrong API usage"),
+            EitherOutput::Second(inner) => inner
+                .poll(cx)
+                .map_ok(|e| e.map_stream(EitherOutput::Second)),
         }
     }
 
-    fn shutdown_substream(
-        &self,
-        cx: &mut Context<'_>,
-        sub: &mut Self::Substream,
-    ) -> Poll<Result<(), Self::Error>> {
-        match (self, sub) {
-            (EitherOutput::First(ref inner), EitherOutput::First(ref mut sub)) => {
-                inner.shutdown_substream(cx, sub).map_err(|e| e.into())
-            }
-            (EitherOutput::Second(ref inner), EitherOutput::Second(ref mut sub)) => {
-                inner.shutdown_substream(cx, sub).map_err(|e| e.into())
-            }
-            _ => panic!("Wrong API usage"),
-        }
-    }
-
-    fn destroy_substream(&self, substream: Self::Substream) {
+    fn open_outbound(&mut self) -> OutboundSubstreamId {
         match self {
-            EitherOutput::First(inner) => match substream {
-                EitherOutput::First(substream) => inner.destroy_substream(substream),
-                _ => panic!("Wrong API usage"),
-            },
-            EitherOutput::Second(inner) => match substream {
-                EitherOutput::Second(substream) => inner.destroy_substream(substream),
-                _ => panic!("Wrong API usage"),
-            },
+            EitherOutput::First(inner) => inner.open_outbound(),
+            EitherOutput::Second(inner) => inner.open_outbound(),
         }
     }
 
-    fn poll_close(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    fn poll_close(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         match self {
-            EitherOutput::First(inner) => inner.poll_close(cx).map_err(|e| e.into()),
-            EitherOutput::Second(inner) => inner.poll_close(cx).map_err(|e| e.into()),
+            EitherOutput::First(inner) => inner.poll_close(cx),
+            EitherOutput::Second(inner) => inner.poll_close(cx),
         }
     }
 }
 
-#[derive(Debug, Copy, Clone)]
-#[must_use = "futures do nothing unless polled"]
-pub enum EitherOutbound<A: StreamMuxer, B: StreamMuxer> {
-    A(A::OutboundSubstream),
-    B(B::OutboundSubstream),
-}
-
 /// Implements `Stream` and dispatches all method calls to either `First` or `Second`.
 #[pin_project(project = EitherListenStreamProj)]
 #[derive(Debug, Copy, Clone)]