refactor(core)!: remove EitherOutput

core/CHANGELOG.md

@@ -27,6 +27,8 @@
 
 - Remove `EitherFuture2` in favor of `EitherFuture`. See [PR 3340].
 
+- Remove `EitherOutput` in favor of `future::Either`. See [PR 3341].
+
 [PR 3031]: https://github.com/libp2p/rust-libp2p/pull/3031
 [PR 3058]: https://github.com/libp2p/rust-libp2p/pull/3058
 [PR 3097]: https://github.com/libp2p/rust-libp2p/pull/3097
@@ -36,6 +38,7 @@
 [PR 3338]: https://github.com/libp2p/rust-libp2p/pull/3338
 [PR 3339]: https://github.com/libp2p/rust-libp2p/pull/3339
 [PR 3340]: https://github.com/libp2p/rust-libp2p/pull/3340
+[PR 3341]: https://github.com/libp2p/rust-libp2p/pull/3341
 
 # 0.37.0
 

core/src/either.rs

@@ -25,167 +25,30 @@ use crate::{
     Multiaddr, ProtocolName,
 };
 use either::Either;
-use futures::{
-    io::{IoSlice, IoSliceMut},
-    prelude::*,
-};
+use futures::prelude::*;
 use pin_project::pin_project;
-use std::{io, pin::Pin, task::Context, task::Poll};
-
-/// Implements `AsyncRead` and `AsyncWrite` and dispatches all method calls to
-/// either `First` or `Second`.
-#[pin_project(project = EitherOutputProj)]
-#[derive(Debug, Copy, Clone)]
-pub enum EitherOutput<A, B> {
-    First(#[pin] A),
-    Second(#[pin] B),
-}
-
-impl<A, B> AsyncRead for EitherOutput<A, B>
-where
-    A: AsyncRead,
-    B: AsyncRead,
-{
-    fn poll_read(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
-        match self.project() {
-            EitherOutputProj::First(a) => AsyncRead::poll_read(a, cx, buf),
-            EitherOutputProj::Second(b) => AsyncRead::poll_read(b, cx, buf),
-        }
-    }
+use std::{pin::Pin, task::Context, task::Poll};
 
-    fn poll_read_vectored(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        bufs: &mut [IoSliceMut<'_>],
-    ) -> Poll<io::Result<usize>> {
-        match self.project() {
-            EitherOutputProj::First(a) => AsyncRead::poll_read_vectored(a, cx, bufs),
-            EitherOutputProj::Second(b) => AsyncRead::poll_read_vectored(b, cx, bufs),
-        }
-    }
-}
-
-impl<A, B> AsyncWrite for EitherOutput<A, B>
-where
-    A: AsyncWrite,
-    B: AsyncWrite,
-{
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<io::Result<usize>> {
-        match self.project() {
-            EitherOutputProj::First(a) => AsyncWrite::poll_write(a, cx, buf),
-            EitherOutputProj::Second(b) => AsyncWrite::poll_write(b, cx, buf),
-        }
-    }
-
-    fn poll_write_vectored(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        bufs: &[IoSlice<'_>],
-    ) -> Poll<io::Result<usize>> {
-        match self.project() {
-            EitherOutputProj::First(a) => AsyncWrite::poll_write_vectored(a, cx, bufs),
-            EitherOutputProj::Second(b) => AsyncWrite::poll_write_vectored(b, cx, bufs),
-        }
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        match self.project() {
-            EitherOutputProj::First(a) => AsyncWrite::poll_flush(a, cx),
-            EitherOutputProj::Second(b) => AsyncWrite::poll_flush(b, cx),
-        }
-    }
-
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        match self.project() {
-            EitherOutputProj::First(a) => AsyncWrite::poll_close(a, cx),
-            EitherOutputProj::Second(b) => AsyncWrite::poll_close(b, cx),
-        }
-    }
-}
-
-impl<A, B, I> Stream for EitherOutput<A, B>
-where
-    A: TryStream<Ok = I>,
-    B: TryStream<Ok = I>,
-{
-    type Item = Result<I, Either<A::Error, B::Error>>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match self.project() {
-            EitherOutputProj::First(a) => {
-                TryStream::try_poll_next(a, cx).map(|v| v.map(|r| r.map_err(Either::Left)))
-            }
-            EitherOutputProj::Second(b) => {
-                TryStream::try_poll_next(b, cx).map(|v| v.map(|r| r.map_err(Either::Right)))
-            }
-        }
-    }
-}
-
-impl<A, B, I> Sink<I> for EitherOutput<A, B>
-where
-    A: Sink<I>,
-    B: Sink<I>,
-{
-    type Error = Either<A::Error, B::Error>;
-
-    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(a) => Sink::poll_ready(a, cx).map_err(Either::Left),
-            EitherOutputProj::Second(b) => Sink::poll_ready(b, cx).map_err(Either::Right),
-        }
-    }
-
-    fn start_send(self: Pin<&mut Self>, item: I) -> Result<(), Self::Error> {
-        match self.project() {
-            EitherOutputProj::First(a) => Sink::start_send(a, item).map_err(Either::Left),
-            EitherOutputProj::Second(b) => Sink::start_send(b, item).map_err(Either::Right),
-        }
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(a) => Sink::poll_flush(a, cx).map_err(Either::Left),
-            EitherOutputProj::Second(b) => Sink::poll_flush(b, cx).map_err(Either::Right),
-        }
-    }
-
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(a) => Sink::poll_close(a, cx).map_err(Either::Left),
-            EitherOutputProj::Second(b) => Sink::poll_close(b, cx).map_err(Either::Right),
-        }
-    }
-}
-
-impl<A, B> StreamMuxer for EitherOutput<A, B>
+impl<A, B> StreamMuxer for future::Either<A, B>
 where
     A: StreamMuxer,
     B: StreamMuxer,
 {
-    type Substream = EitherOutput<A::Substream, B::Substream>;
+    type Substream = future::Either<A::Substream, B::Substream>;
     type Error = Either<A::Error, B::Error>;
 
     fn poll_inbound(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Self::Substream, Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(inner) => inner
+        match as_pin_mut(self) {
+            future::Either::Left(inner) => inner
                 .poll_inbound(cx)
-                .map_ok(EitherOutput::First)
+                .map_ok(future::Either::Left)
                 .map_err(Either::Left),
-            EitherOutputProj::Second(inner) => inner
+            future::Either::Right(inner) => inner
                 .poll_inbound(cx)
-                .map_ok(EitherOutput::Second)
+                .map_ok(future::Either::Right)
                 .map_err(Either::Right),
         }
     }
@@ -194,32 +57,54 @@ where
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Self::Substream, Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(inner) => inner
+        match as_pin_mut(self) {
+            future::Either::Left(inner) => inner
                 .poll_outbound(cx)
-                .map_ok(EitherOutput::First)
+                .map_ok(future::Either::Left)
                 .map_err(Either::Left),
-            EitherOutputProj::Second(inner) => inner
+            future::Either::Right(inner) => inner
                 .poll_outbound(cx)
-                .map_ok(EitherOutput::Second)
+                .map_ok(future::Either::Right)
                 .map_err(Either::Right),
         }
     }
 
     fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(inner) => inner.poll_close(cx).map_err(Either::Left),
-            EitherOutputProj::Second(inner) => inner.poll_close(cx).map_err(Either::Right),
+        match as_pin_mut(self) {
+            future::Either::Left(inner) => inner.poll_close(cx).map_err(Either::Left),
+            future::Either::Right(inner) => inner.poll_close(cx).map_err(Either::Right),
         }
     }
 
     fn poll(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
-        match self.project() {
-            EitherOutputProj::First(inner) => inner.poll(cx).map_err(Either::Left),
-            EitherOutputProj::Second(inner) => inner.poll(cx).map_err(Either::Right),
+        match as_pin_mut(self) {
+            future::Either::Left(inner) => inner.poll(cx).map_err(Either::Left),
+            future::Either::Right(inner) => inner.poll(cx).map_err(Either::Right),
+        }
+    }
+}
+
+/// Convert `Pin<&mut Either<A, B>>` to `Either<Pin<&mut A>, Pin<&mut B>>`,
+/// pinned projections of the inner variants.
+///
+/// Local function until <https://github.com/rust-lang/futures-rs/pull/2691> is merged.
+fn as_pin_mut<A, B>(
+    either: Pin<&mut future::Either<A, B>>,
+) -> future::Either<Pin<&mut A>, Pin<&mut B>> {
+    // SAFETY: `get_unchecked_mut` is fine because we don't move anything.
+    // We can use `new_unchecked` because the `inner` parts are guaranteed
+    // to be pinned, as they come from `self` which is pinned, and we never
+    // offer an unpinned `&mut L` or `&mut R` through `Pin<&mut Self>`. We
+    // also don't have an implementation of `Drop`, nor manual `Unpin`.
+    unsafe {
+        match *Pin::get_unchecked_mut(either) {
+            future::Either::Left(ref mut inner) => future::Either::Left(Pin::new_unchecked(inner)),
+            future::Either::Right(ref mut inner) => {
+                future::Either::Right(Pin::new_unchecked(inner))
+            }
         }
     }
 }
@@ -238,15 +123,15 @@ where
     AFuture: TryFuture<Ok = AInner>,
     BFuture: TryFuture<Ok = BInner>,
 {
-    type Output = Result<EitherOutput<AInner, BInner>, Either<AFuture::Error, BFuture::Error>>;
+    type Output = Result<future::Either<AInner, BInner>, Either<AFuture::Error, BFuture::Error>>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         match self.project() {
             EitherFutureProj::First(a) => TryFuture::try_poll(a, cx)
-                .map_ok(EitherOutput::First)
+                .map_ok(future::Either::Left)
                 .map_err(Either::Left),
             EitherFutureProj::Second(a) => TryFuture::try_poll(a, cx)
-                .map_ok(EitherOutput::Second)
+                .map_ok(future::Either::Right)
                 .map_err(Either::Right),
         }
     }
@@ -272,7 +157,7 @@ where
     B: Transport,
     A: Transport,
 {
-    type Output = EitherOutput<A::Output, B::Output>;
+    type Output = future::Either<A::Output, B::Output>;
     type Error = Either<A::Error, B::Error>;
     type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
     type Dial = EitherFuture<A::Dial, B::Dial>;

core/src/transport/choice.rs

@@ -18,9 +18,10 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
-use crate::either::{EitherFuture, EitherOutput};
+use crate::either::EitherFuture;
 use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
 use either::Either;
+use futures::future;
 use multiaddr::Multiaddr;
 use std::{pin::Pin, task::Context, task::Poll};
 
@@ -40,7 +41,7 @@ where
     B: Transport,
     A: Transport,
 {
-    type Output = EitherOutput<A::Output, B::Output>;
+    type Output = future::Either<A::Output, B::Output>;
     type Error = Either<A::Error, B::Error>;
     type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
     type Dial = EitherFuture<A::Dial, B::Dial>;

core/src/upgrade/either.rs

@@ -19,10 +19,11 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{
-    either::{EitherFuture, EitherName, EitherOutput},
+    either::{EitherFuture, EitherName},
     upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo},
 };
 use either::Either;
+use futures::future;
 
 impl<A, B> UpgradeInfo for Either<A, B>
 where
@@ -48,7 +49,7 @@ where
     A: InboundUpgrade<C, Output = TA, Error = EA>,
     B: InboundUpgrade<C, Output = TB, Error = EB>,
 {
-    type Output = EitherOutput<TA, TB>;
+    type Output = future::Either<TA, TB>;
     type Error = Either<EA, EB>;
     type Future = EitherFuture<A::Future, B::Future>;
 
@@ -70,7 +71,7 @@ where
     A: OutboundUpgrade<C, Output = TA, Error = EA>,
     B: OutboundUpgrade<C, Output = TB, Error = EB>,
 {
-    type Output = EitherOutput<TA, TB>;
+    type Output = future::Either<TA, TB>;
     type Error = Either<EA, EB>;
     type Future = EitherFuture<A::Future, B::Future>;
 

core/src/upgrade/select.rs

@@ -20,10 +20,11 @@
 
 use crate::either::EitherFuture;
 use crate::{
-    either::{EitherName, EitherOutput},
+    either::EitherName,
     upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo},
 };
 use either::Either;
+use futures::future;
 
 /// Upgrade that combines two upgrades into one. Supports all the protocols supported by either
 /// sub-upgrade.
@@ -65,7 +66,7 @@ where
     A: InboundUpgrade<C, Output = TA, Error = EA>,
     B: InboundUpgrade<C, Output = TB, Error = EB>,
 {
-    type Output = EitherOutput<TA, TB>;
+    type Output = future::Either<TA, TB>;
     type Error = Either<EA, EB>;
     type Future = EitherFuture<A::Future, B::Future>;
 
@@ -82,7 +83,7 @@ where
     A: OutboundUpgrade<C, Output = TA, Error = EA>,
     B: OutboundUpgrade<C, Output = TB, Error = EB>,
 {
-    type Output = EitherOutput<TA, TB>;
+    type Output = future::Either<TA, TB>;
     type Error = Either<EA, EB>;
     type Future = EitherFuture<A::Future, B::Future>;
 

protocols/dcutr/src/handler/relayed.rs

@@ -22,9 +22,9 @@
 
 use crate::protocol;
 use either::Either;
+use futures::future;
 use futures::future::{BoxFuture, FutureExt};
 use instant::Instant;
-use libp2p_core::either::EitherOutput;
 use libp2p_core::multiaddr::Multiaddr;
 use libp2p_core::upgrade::{DeniedUpgrade, NegotiationError, UpgradeError};
 use libp2p_core::ConnectedPoint;
@@ -174,7 +174,7 @@ impl Handler {
         >,
     ) {
         match output {
-            EitherOutput::First(inbound_connect) => {
+            future::Either::Left(inbound_connect) => {
                 let remote_addr = match &self.endpoint {
                     ConnectedPoint::Dialer { address, role_override: _ } => address.clone(),
                     ConnectedPoint::Listener { ..} => unreachable!("`<Handler as ConnectionHandler>::listen_protocol` denies all incoming substreams as a listener."),
@@ -187,7 +187,7 @@ impl Handler {
                 ));
             }
             // A connection listener denies all incoming substreams, thus none can ever be fully negotiated.
-            EitherOutput::Second(output) => void::unreachable(output),
+            future::Either::Right(output) => void::unreachable(output),
         }
     }