core/muxing: Flatten StreamMuxer interface to poll_{inbound,outbound,address_change,close}

core/CHANGELOG.md

@@ -1,13 +1,14 @@
 # 0.34.0 - unreleased
 
-- Introduce `StreamMuxerEvent::map_inbound_stream`. See [PR 2691].
 - Remove `{read,write,flush,shutdown,destroy}_substream` functions from `StreamMuxer` trait
   in favor of forcing `StreamMuxer::Substream` to implement `AsyncRead + AsyncWrite`. See [PR 2707].
 - Replace `Into<std::io::Error>` bound on `StreamMuxer::Error` with `std::error::Error`. See [PR 2710].
+- Remove `StreamMuxer::poll_event` in favor of individual functions: `poll_inbound`, `poll_outbound`
+  and `poll_address_change`. Consequently, `StreamMuxerEvent` is also removed. See [PR 2724].
 
-[PR 2691]: https://github.com/libp2p/rust-libp2p/pull/2691
 [PR 2707]: https://github.com/libp2p/rust-libp2p/pull/2707
 [PR 2710]: https://github.com/libp2p/rust-libp2p/pull/2710
+[PR 2724]: https://github.com/libp2p/rust-libp2p/pull/2724
 
 # 0.33.0
 

core/Cargo.toml

@@ -12,6 +12,7 @@ categories = ["network-programming", "asynchronous"]
 
 [dependencies]
 asn1_der = "0.7.4"
+bitflags = "1.3.2"
 bs58 = "0.4.0"
 ed25519-dalek = "1.0.1"
 either = "1.5"

core/src/either.rs

@@ -19,7 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{
-    muxing::{StreamMuxer, StreamMuxerEvent},
+    muxing::StreamMuxer,
     transport::{ListenerEvent, Transport, TransportError},
     Multiaddr, ProtocolName,
 };
@@ -202,60 +202,38 @@ where
     B: StreamMuxer,
 {
     type Substream = EitherOutput<A::Substream, B::Substream>;
-    type OutboundSubstream = EitherOutbound<A, B>;
     type Error = EitherError<A::Error, B::Error>;
 
-    fn poll_event(
-        &self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>> {
+    fn poll_inbound(&self, cx: &mut Context<'_>) -> Poll<Result<Self::Substream, Self::Error>> {
         match self {
             EitherOutput::First(inner) => inner
-                .poll_event(cx)
-                .map_err(EitherError::A)
-                .map_ok(|event| event.map_inbound_stream(EitherOutput::First)),
+                .poll_inbound(cx)
+                .map_ok(EitherOutput::First)
+                .map_err(EitherError::A),
             EitherOutput::Second(inner) => inner
-                .poll_event(cx)
-                .map_err(EitherError::B)
-                .map_ok(|event| event.map_inbound_stream(EitherOutput::Second)),
+                .poll_inbound(cx)
+                .map_ok(EitherOutput::Second)
+                .map_err(EitherError::B),
         }
     }
 
-    fn open_outbound(&self) -> Self::OutboundSubstream {
+    fn poll_outbound(&self, cx: &mut Context<'_>) -> Poll<Result<Self::Substream, Self::Error>> {
         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))
+            EitherOutput::First(inner) => inner
+                .poll_outbound(cx)
+                .map_ok(EitherOutput::First)
                 .map_err(EitherError::A),
-            (EitherOutput::Second(ref inner), EitherOutbound::B(ref mut substream)) => inner
-                .poll_outbound(cx, substream)
-                .map(|p| p.map(EitherOutput::Second))
+            EitherOutput::Second(inner) => inner
+                .poll_outbound(cx)
+                .map_ok(EitherOutput::Second)
                 .map_err(EitherError::B),
-            _ => panic!("Wrong API usage"),
         }
     }
 
-    fn destroy_outbound(&self, substream: Self::OutboundSubstream) {
+    fn poll_address_change(&self, cx: &mut Context<'_>) -> Poll<Result<Multiaddr, Self::Error>> {
         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"),
-            },
+            EitherOutput::First(inner) => inner.poll_address_change(cx).map_err(EitherError::A),
+            EitherOutput::Second(inner) => inner.poll_address_change(cx).map_err(EitherError::B),
         }
     }
 
@@ -267,13 +245,6 @@ where
     }
 }
 
-#[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)]

core/src/muxing.rs

@@ -64,61 +64,22 @@ mod singleton;
 ///
 /// A substream created by a [`StreamMuxer`] is a type that implements [`AsyncRead`] and [`AsyncWrite`].
 ///
-/// Inbound substreams are reported via [`StreamMuxer::poll_event`].
-/// Outbound substreams can be opened via [`StreamMuxer::open_outbound`] and subsequent polling via
-/// [`StreamMuxer::poll_outbound`].
+/// TODO(docs)
 pub trait StreamMuxer {
     /// Type of the object that represents the raw substream where data can be read and written.
     type Substream: AsyncRead + AsyncWrite;
 
-    /// Future that will be resolved when the outgoing substream is open.
-    type OutboundSubstream;
-
     /// Error type of the muxer
     type Error: std::error::Error;
 
-    /// Polls for a connection-wide event.
-    ///
-    /// This function behaves the same as a `Stream`.
-    ///
-    /// If `Pending` is returned, then the current task will be notified once the muxer
-    /// is ready to be polled, similar to the API of `Stream::poll()`.
-    /// Only the latest task that was used to call this method may be notified.
-    ///
-    /// It is permissible and common to use this method to perform background
-    /// work, such as processing incoming packets and polling timers.
-    ///
-    /// An error can be generated if the connection has been closed.
-    fn poll_event(
-        &self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>>;
-
-    /// Opens a new outgoing substream, and produces the equivalent to a future that will be
-    /// resolved when it becomes available.
-    ///
-    /// The API of `OutboundSubstream` is totally opaque, and the object can only be interfaced
-    /// through the methods on the `StreamMuxer` trait.
-    fn open_outbound(&self) -> Self::OutboundSubstream;
+    /// TODO(docs)
+    fn poll_inbound(&self, cx: &mut Context<'_>) -> Poll<Result<Self::Substream, Self::Error>>;
 
-    /// Polls the outbound substream.
-    ///
-    /// If `Pending` is returned, then the current task will be notified once the substream
-    /// is ready to be polled, similar to the API of `Future::poll()`.
-    /// However, for each individual outbound substream, only the latest task that was used to
-    /// call this method may be notified.
-    ///
-    /// May panic or produce an undefined result if an earlier polling of the same substream
-    /// returned `Ready` or `Err`.
-    fn poll_outbound(
-        &self,
-        cx: &mut Context<'_>,
-        s: &mut Self::OutboundSubstream,
-    ) -> Poll<Result<Self::Substream, Self::Error>>;
+    /// TODO(docs)
+    fn poll_outbound(&self, cx: &mut Context<'_>) -> Poll<Result<Self::Substream, Self::Error>>;
 
-    /// Destroys an outbound substream future. Use this after the outbound substream has finished,
-    /// or if you want to interrupt it.
-    fn destroy_outbound(&self, s: Self::OutboundSubstream);
+    /// TODO(docs)
+    fn poll_address_change(&self, cx: &mut Context<'_>) -> Poll<Result<Multiaddr, Self::Error>>;
 
     /// Closes this `StreamMuxer`.
     ///
@@ -132,38 +93,3 @@ pub trait StreamMuxer {
     /// >           immediately dropping the muxer.
     fn poll_close(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
 }
-
-/// Event about a connection, reported by an implementation of [`StreamMuxer`].
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum StreamMuxerEvent<T> {
-    /// Remote has opened a new substream. Contains the substream in question.
-    InboundSubstream(T),
-
-    /// Address to the remote has changed. The previous one is now obsolete.
-    ///
-    /// > **Note**: This can for example happen when using the QUIC protocol, where the two nodes
-    /// >           can change their IP address while retaining the same QUIC connection.
-    AddressChange(Multiaddr),
-}
-
-impl<T> StreamMuxerEvent<T> {
-    /// If `self` is a [`StreamMuxerEvent::InboundSubstream`], returns the content. Otherwise
-    /// returns `None`.
-    pub fn into_inbound_substream(self) -> Option<T> {
-        if let StreamMuxerEvent::InboundSubstream(s) = self {
-            Some(s)
-        } else {
-            None
-        }
-    }
-
-    /// Map the stream within [`StreamMuxerEvent::InboundSubstream`] to a new type.
-    pub fn map_inbound_stream<O>(self, map: impl FnOnce(T) -> O) -> StreamMuxerEvent<O> {
-        match self {
-            StreamMuxerEvent::InboundSubstream(stream) => {
-                StreamMuxerEvent::InboundSubstream(map(stream))
-            }
-            StreamMuxerEvent::AddressChange(addr) => StreamMuxerEvent::AddressChange(addr),
-        }
-    }
-}