fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-20 05:16:35 +00:00
Code Issues Projects Releases Wiki Activity

refactor(core)!: remove EitherError in favor of either::Either (#3337)

Browse Source 
Defining our own `EitherError` type has no value now that `Either` provides the same implementation.

Related: https://github.com/libp2p/rust-libp2p/issues/3271
This commit is contained in:
Thomas Eizinger
2023-01-18 10:05:59 +11:00
committed by GitHub
parent 29a77164f1
commit f4fed3880b
17 changed files with 118 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.toml
View File

@ -132,6 +132,7 @@ libp2p-gossipsub = { version = "0.44.0", path = "protocols/gossipsub", optional
[dev-dependencies]
async-std = { version = "1.6.2", features = ["attributes"] }
async-trait = "0.1"
either = "1.8.0"
env_logger = "0.10.0"
clap = { version = "4.0.13", features = ["derive"] }
tokio = { version = "1.15", features = ["io-util", "io-std", "macros", "rt", "rt-multi-thread"] }

3
core/CHANGELOG.md
View File

@ -12,11 +12,14 @@
- Improve error messages in case keys cannot be decoded because of missing feature flags. See [PR 2972].
- Remove `EitherError` in favor of `either::Either`. See [PR 3337].
[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
[PR 3090]: https://github.com/libp2p/rust-libp2p/pull/3090
[PR 2972]: https://github.com/libp2p/rust-libp2p/pull/2972
[PR 3337]: https://github.com/libp2p/rust-libp2p/pull/3337
# 0.37.0

113
core/src/either.rs
View File

@ -24,44 +24,13 @@ use crate::{
transport::{ListenerId, Transport, TransportError, TransportEvent},
Multiaddr, ProtocolName,
};
use either::Either;
use futures::{
io::{IoSlice, IoSliceMut},
prelude::*,
};
use pin_project::pin_project;
use std::{fmt, io, pin::Pin, task::Context, task::Poll};
#[derive(Debug, Copy, Clone)]
pub enum EitherError<A, B> {
A(A),
B(B),
}
impl<A, B> fmt::Display for EitherError<A, B>
where
A: fmt::Display,
B: fmt::Display,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
EitherError::A(a) => a.fmt(f),
EitherError::B(b) => b.fmt(f),
}
}
}
impl<A, B> std::error::Error for EitherError<A, B>
where
A: std::error::Error,
B: std::error::Error,
{
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
EitherError::A(a) => a.source(),
EitherError::B(b) => b.source(),
}
}
}
use std::{io, pin::Pin, task::Context, task::Poll};
/// Implements `AsyncRead` and `AsyncWrite` and dispatches all method calls to
/// either `First` or `Second`.
@ -147,15 +116,15 @@ where
A: TryStream<Ok = I>,
B: TryStream<Ok = I>,
{
type Item = Result<I, EitherError<A::Error, B::Error>>;
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(EitherError::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(EitherError::B)))
TryStream::try_poll_next(b, cx).map(|v| v.map(|r| r.map_err(Either::Right)))
}
}
}
@ -166,33 +135,33 @@ where
A: Sink<I>,
B: Sink<I>,
{
type Error = EitherError<A::Error, B::Error>;
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(EitherError::A),
EitherOutputProj::Second(b) => Sink::poll_ready(b, cx).map_err(EitherError::B),
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(EitherError::A),
EitherOutputProj::Second(b) => Sink::start_send(b, item).map_err(EitherError::B),
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(EitherError::A),
EitherOutputProj::Second(b) => Sink::poll_flush(b, cx).map_err(EitherError::B),
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(EitherError::A),
EitherOutputProj::Second(b) => Sink::poll_close(b, cx).map_err(EitherError::B),
EitherOutputProj::First(a) => Sink::poll_close(a, cx).map_err(Either::Left),
EitherOutputProj::Second(b) => Sink::poll_close(b, cx).map_err(Either::Right),
}
}
}
@ -203,7 +172,7 @@ where
B: StreamMuxer,
{
type Substream = EitherOutput<A::Substream, B::Substream>;
type Error = EitherError<A::Error, B::Error>;
type Error = Either<A::Error, B::Error>;
fn poll_inbound(
self: Pin<&mut Self>,
@ -213,11 +182,11 @@ where
EitherOutputProj::First(inner) => inner
.poll_inbound(cx)
.map_ok(EitherOutput::First)
.map_err(EitherError::A),
.map_err(Either::Left),
EitherOutputProj::Second(inner) => inner
.poll_inbound(cx)
.map_ok(EitherOutput::Second)
.map_err(EitherError::B),
.map_err(Either::Right),
}
}
@ -229,18 +198,18 @@ where
EitherOutputProj::First(inner) => inner
.poll_outbound(cx)
.map_ok(EitherOutput::First)
.map_err(EitherError::A),
.map_err(Either::Left),
EitherOutputProj::Second(inner) => inner
.poll_outbound(cx)
.map_ok(EitherOutput::Second)
.map_err(EitherError::B),
.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(EitherError::A),
EitherOutputProj::Second(inner) => inner.poll_close(cx).map_err(EitherError::B),
EitherOutputProj::First(inner) => inner.poll_close(cx).map_err(Either::Left),
EitherOutputProj::Second(inner) => inner.poll_close(cx).map_err(Either::Right),
}
}
@ -249,8 +218,8 @@ where
cx: &mut Context<'_>,
) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
match self.project() {
EitherOutputProj::First(inner) => inner.poll(cx).map_err(EitherError::A),
EitherOutputProj::Second(inner) => inner.poll(cx).map_err(EitherError::B),
EitherOutputProj::First(inner) => inner.poll(cx).map_err(Either::Left),
EitherOutputProj::Second(inner) => inner.poll(cx).map_err(Either::Right),
}
}
}
@ -269,16 +238,16 @@ where
AFuture: TryFuture<Ok = AInner>,
BFuture: TryFuture<Ok = BInner>,
{
type Output = Result<EitherOutput<AInner, BInner>, EitherError<AFuture::Error, BFuture::Error>>;
type Output = Result<EitherOutput<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_err(EitherError::A),
.map_err(Either::Left),
EitherFutureProj::Second(a) => TryFuture::try_poll(a, cx)
.map_ok(EitherOutput::Second)
.map_err(EitherError::B),
.map_err(Either::Right),
}
}
}
@ -296,16 +265,16 @@ where
AFut: TryFuture<Ok = AItem, Error = AError>,
BFut: TryFuture<Ok = BItem, Error = BError>,
{
type Output = Result<EitherOutput<AItem, BItem>, EitherError<AError, BError>>;
type Output = Result<EitherOutput<AItem, BItem>, Either<AError, BError>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.project() {
EitherFuture2Proj::A(a) => TryFuture::try_poll(a, cx)
.map_ok(EitherOutput::First)
.map_err(EitherError::A),
.map_err(Either::Left),
EitherFuture2Proj::B(a) => TryFuture::try_poll(a, cx)
.map_ok(EitherOutput::Second)
.map_err(EitherError::B),
.map_err(Either::Right),
}
}
}
@ -338,7 +307,7 @@ where
A: Transport,
{
type Output = EitherOutput<A::Output, B::Output>;
type Error = EitherError<A::Error, B::Error>;
type Error = Either<A::Error, B::Error>;
type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
type Dial = EitherFuture<A::Dial, B::Dial>;
@ -349,18 +318,16 @@ where
match self.project() {
EitherTransportProj::Left(a) => match a.poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(event) => Poll::Ready(
event
.map_upgrade(EitherFuture::First)
.map_err(EitherError::A),
),
Poll::Ready(event) => {
Poll::Ready(event.map_upgrade(EitherFuture::First).map_err(Either::Left))
}
},
EitherTransportProj::Right(b) => match b.poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(event) => Poll::Ready(
event
.map_upgrade(EitherFuture::Second)
.map_err(EitherError::B),
.map_err(Either::Right),
),
},
}
@ -378,11 +345,11 @@ where
match self {
EitherTransport::Left(a) => a.listen_on(addr).map_err(|e| match e {
MultiaddrNotSupported(addr) => MultiaddrNotSupported(addr),
Other(err) => Other(EitherError::A(err)),
Other(err) => Other(Either::Left(err)),
}),
EitherTransport::Right(b) => b.listen_on(addr).map_err(|e| match e {
MultiaddrNotSupported(addr) => MultiaddrNotSupported(addr),
Other(err) => Other(EitherError::B(err)),
Other(err) => Other(Either::Right(err)),
}),
}
}
@ -393,12 +360,12 @@ where
EitherTransport::Left(a) => match a.dial(addr) {
Ok(connec) => Ok(EitherFuture::First(connec)),
Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
Err(Other(err)) => Err(Other(EitherError::A(err))),
Err(Other(err)) => Err(Other(Either::Left(err))),
},
EitherTransport::Right(b) => match b.dial(addr) {
Ok(connec) => Ok(EitherFuture::Second(connec)),
Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
Err(Other(err)) => Err(Other(EitherError::B(err))),
Err(Other(err)) => Err(Other(Either::Right(err))),
},
}
}
@ -415,12 +382,12 @@ where
EitherTransport::Left(a) => match a.dial_as_listener(addr) {
Ok(connec) => Ok(EitherFuture::First(connec)),
Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
Err(Other(err)) => Err(Other(EitherError::A(err))),
Err(Other(err)) => Err(Other(Either::Left(err))),
},
EitherTransport::Right(b) => match b.dial_as_listener(addr) {
Ok(connec) => Ok(EitherFuture::Second(connec)),
Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
Err(Other(err)) => Err(Other(EitherError::B(err))),
Err(Other(err)) => Err(Other(Either::Right(err))),
},
}
}

20
core/src/transport/and_then.rs
View File

@ -20,10 +20,10 @@
use crate::{
connection::{ConnectedPoint, Endpoint},
either::EitherError,
transport::{ListenerId, Transport, TransportError, TransportEvent},
};
use futures::{future::Either, prelude::*};
use either::Either;
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::{error, marker::PhantomPinned, pin::Pin, task::Context, task::Poll};
@ -50,14 +50,14 @@ where
F::Error: error::Error,
{
type Output = O;
type Error = EitherError<T::Error, F::Error>;
type Error = Either<T::Error, F::Error>;
type ListenerUpgrade = AndThenFuture<T::ListenerUpgrade, C, F>;
type Dial = AndThenFuture<T::Dial, C, F>;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.transport
.listen_on(addr)
.map_err(|err| err.map(EitherError::A))
.map_err(|err| err.map(Either::Left))
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
@ -68,7 +68,7 @@ where
let dialed_fut = self
.transport
.dial(addr.clone())
.map_err(|err| err.map(EitherError::A))?;
.map_err(|err| err.map(Either::Left))?;
let future = AndThenFuture {
inner: Either::Left(Box::pin(dialed_fut)),
args: Some((
@ -90,7 +90,7 @@ where
let dialed_fut = self
.transport
.dial_as_listener(addr.clone())
.map_err(|err| err.map(EitherError::A))?;
.map_err(|err| err.map(Either::Left))?;
let future = AndThenFuture {
inner: Either::Left(Box::pin(dialed_fut)),
args: Some((
@ -139,7 +139,7 @@ where
Poll::Ready(other) => {
let mapped = other
.map_upgrade(|_upgrade| unreachable!("case already matched"))
.map_err(EitherError::A);
.map_err(Either::Left);
Poll::Ready(mapped)
}
Poll::Pending => Poll::Pending,
@ -163,7 +163,7 @@ where
TMap: FnOnce(TFut::Ok, ConnectedPoint) -> TMapOut,
TMapOut: TryFuture,
{
type Output = Result<TMapOut::Ok, EitherError<TFut::Error, TMapOut::Error>>;
type Output = Result<TMapOut::Ok, Either<TFut::Error, TMapOut::Error>>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
@ -171,7 +171,7 @@ where
Either::Left(future) => {
let item = match TryFuture::try_poll(future.as_mut(), cx) {
Poll::Ready(Ok(v)) => v,
Poll::Ready(Err(err)) => return Poll::Ready(Err(EitherError::A(err))),
Poll::Ready(Err(err)) => return Poll::Ready(Err(Either::Left(err))),
Poll::Pending => return Poll::Pending,
};
let (f, a) = self
@ -183,7 +183,7 @@ where
Either::Right(future) => {
return match TryFuture::try_poll(future.as_mut(), cx) {
Poll::Ready(Ok(v)) => Poll::Ready(Ok(v)),
Poll::Ready(Err(err)) => return Poll::Ready(Err(EitherError::B(err))),
Poll::Ready(Err(err)) => return Poll::Ready(Err(Either::Right(err))),
Poll::Pending => Poll::Pending,
}
}

21
core/src/transport/choice.rs
View File

@ -18,8 +18,9 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::either::{EitherError, EitherFuture, EitherOutput};
use crate::either::{EitherFuture, EitherOutput};
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use either::Either;
use multiaddr::Multiaddr;
use std::{pin::Pin, task::Context, task::Poll};
@ -40,19 +41,19 @@ where
A: Transport,
{
type Output = EitherOutput<A::Output, B::Output>;
type Error = EitherError<A::Error, B::Error>;
type Error = Either<A::Error, B::Error>;
type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
type Dial = EitherFuture<A::Dial, B::Dial>;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
let addr = match self.0.listen_on(addr) {
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
res => return res.map_err(|err| err.map(EitherError::A)),
res => return res.map_err(|err| err.map(Either::Left)),
};
let addr = match self.1.listen_on(addr) {
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
res => return res.map_err(|err| err.map(EitherError::B)),
res => return res.map_err(|err| err.map(Either::Right)),
};
Err(TransportError::MultiaddrNotSupported(addr))
@ -67,7 +68,7 @@ where
Ok(connec) => return Ok(EitherFuture::First(connec)),
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
Err(TransportError::Other(err)) => {
return Err(TransportError::Other(EitherError::A(err)))
return Err(TransportError::Other(Either::Left(err)))
}
};
@ -75,7 +76,7 @@ where
Ok(connec) => return Ok(EitherFuture::Second(connec)),
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
Err(TransportError::Other(err)) => {
return Err(TransportError::Other(EitherError::B(err)))
return Err(TransportError::Other(Either::Right(err)))
}
};
@ -90,7 +91,7 @@ where
Ok(connec) => return Ok(EitherFuture::First(connec)),
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
Err(TransportError::Other(err)) => {
return Err(TransportError::Other(EitherError::A(err)))
return Err(TransportError::Other(Either::Left(err)))
}
};
@ -98,7 +99,7 @@ where
Ok(connec) => return Ok(EitherFuture::Second(connec)),
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
Err(TransportError::Other(err)) => {
return Err(TransportError::Other(EitherError::B(err)))
return Err(TransportError::Other(Either::Right(err)))
}
};
@ -120,13 +121,13 @@ where
let this = self.project();
match this.0.poll(cx) {
Poll::Ready(ev) => {
return Poll::Ready(ev.map_upgrade(EitherFuture::First).map_err(EitherError::A))
return Poll::Ready(ev.map_upgrade(EitherFuture::First).map_err(Either::Left))
}
Poll::Pending => {}
}
match this.1.poll(cx) {
Poll::Ready(ev) => {
return Poll::Ready(ev.map_upgrade(EitherFuture::Second).map_err(EitherError::B))
return Poll::Ready(ev.map_upgrade(EitherFuture::Second).map_err(Either::Right))
}
Poll::Pending => {}
}

7
core/src/upgrade/either.rs
View File

@ -19,9 +19,10 @@
// DEALINGS IN THE SOFTWARE.
use crate::{
either::{EitherError, EitherFuture2, EitherName, EitherOutput},
either::{EitherFuture2, EitherName, EitherOutput},
upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo},
};
use either::Either;
/// A type to represent two possible upgrade types (inbound or outbound).
#[derive(Debug, Clone)]
@ -55,7 +56,7 @@ where
B: InboundUpgrade<C, Output = TB, Error = EB>,
{
type Output = EitherOutput<TA, TB>;
type Error = EitherError<EA, EB>;
type Error = Either<EA, EB>;
type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_inbound(self, sock: C, info: Self::Info) -> Self::Future {
@ -77,7 +78,7 @@ where
B: OutboundUpgrade<C, Output = TB, Error = EB>,
{
type Output = EitherOutput<TA, TB>;
type Error = EitherError<EA, EB>;
type Error = Either<EA, EB>;
type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_outbound(self, sock: C, info: Self::Info) -> Self::Future {

7
core/src/upgrade/select.rs
View File

@ -19,9 +19,10 @@
// DEALINGS IN THE SOFTWARE.
use crate::{
either::{EitherError, EitherFuture2, EitherName, EitherOutput},
either::{EitherFuture2, EitherName, EitherOutput},
upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo},
};
use either::Either;
/// Upgrade that combines two upgrades into one. Supports all the protocols supported by either
/// sub-upgrade.
@ -64,7 +65,7 @@ where
B: InboundUpgrade<C, Output = TB, Error = EB>,
{
type Output = EitherOutput<TA, TB>;
type Error = EitherError<EA, EB>;
type Error = Either<EA, EB>;
type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_inbound(self, sock: C, info: Self::Info) -> Self::Future {
@ -81,7 +82,7 @@ where
B: OutboundUpgrade<C, Output = TB, Error = EB>,
{
type Output = EitherOutput<TA, TB>;
type Error = EitherError<EA, EB>;
type Error = Either<EA, EB>;
type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_outbound(self, sock: C, info: Self::Info) -> Self::Future {

4
examples/file-sharing.rs
View File

@ -207,8 +207,8 @@ enum CliArgument {
mod network {
use super::*;
use async_trait::async_trait;
use either::Either;
use futures::channel::{mpsc, oneshot};
use libp2p::core::either::EitherError;
use libp2p::core::upgrade::{read_length_prefixed, write_length_prefixed, ProtocolName};
use libp2p::identity;
use libp2p::identity::ed25519;
@ -405,7 +405,7 @@ mod network {
&mut self,
event: SwarmEvent<
ComposedEvent,
EitherError<ConnectionHandlerUpgrErr<io::Error>, io::Error>,
Either<ConnectionHandlerUpgrErr<io::Error>, io::Error>,
>,
) {
match event {

15
protocols/dcutr/src/handler/relayed.rs
View File

@ -21,9 +21,10 @@
//! [`ConnectionHandler`] handling relayed connection potentially upgraded to a direct connection.
use crate::protocol;
use either::Either;
use futures::future::{BoxFuture, FutureExt};
use instant::Instant;
use libp2p_core::either::{EitherError, EitherOutput};
use libp2p_core::either::EitherOutput;
use libp2p_core::multiaddr::Multiaddr;
use libp2p_core::upgrade::{self, DeniedUpgrade, NegotiationError, UpgradeError};
use libp2p_core::ConnectedPoint;
@ -134,7 +135,7 @@ pub struct Handler {
/// A pending fatal error that results in the connection being closed.
pending_error: Option<
ConnectionHandlerUpgrErr<
EitherError<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
Either<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
>,
>,
/// Queue of events to return when polled.
@ -252,8 +253,8 @@ impl Handler {
// the remote peer and results in closing the connection.
self.pending_error = Some(error.map_upgrade_err(|e| {
e.map_err(|e| match e {
EitherError::A(e) => EitherError::A(e),
EitherError::B(v) => void::unreachable(v),
Either::Left(e) => Either::Left(e),
Either::Right(v) => void::unreachable(v),
})
}));
}
@ -292,7 +293,7 @@ impl Handler {
_ => {
// Anything else is considered a fatal error or misbehaviour of
// the remote peer and results in closing the connection.
self.pending_error = Some(error.map_upgrade_err(|e| e.map_err(EitherError::B)));
self.pending_error = Some(error.map_upgrade_err(|e| e.map_err(Either::Right)));
}
}
}
@ -302,7 +303,7 @@ impl ConnectionHandler for Handler {
type InEvent = Command;
type OutEvent = Event;
type Error = ConnectionHandlerUpgrErr<
EitherError<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
Either<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
>;
type InboundProtocol = upgrade::EitherUpgrade<protocol::inbound::Upgrade, DeniedUpgrade>;
type OutboundProtocol = protocol::outbound::Upgrade;
@ -393,7 +394,7 @@ impl ConnectionHandler for Handler {
}
Err(e) => {
return Poll::Ready(ConnectionHandlerEvent::Close(
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(e))),
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(e))),
))
}
}

1
protocols/identify/Cargo.toml
View File

@ -23,6 +23,7 @@ prost = "0.11"
smallvec = "1.6.1"
thiserror = "1.0"
void = "1.0"
either = "1.8.0"
[dev-dependencies]
async-std = { version = "1.6.2", features = ["attributes"] }

7
protocols/identify/src/handler.rs
View File

@ -21,11 +21,12 @@
use crate::protocol::{
self, Identify, InboundPush, Info, OutboundPush, Protocol, Push, UpgradeError,
};
use either::Either;
use futures::future::BoxFuture;
use futures::prelude::*;
use futures::stream::FuturesUnordered;
use futures_timer::Delay;
use libp2p_core::either::{EitherError, EitherOutput};
use libp2p_core::either::EitherOutput;
use libp2p_core::upgrade::{EitherUpgrade, SelectUpgrade};
use libp2p_core::{ConnectedPoint, Multiaddr, PeerId, PublicKey};
use libp2p_swarm::handler::{
@ -259,8 +260,8 @@ impl Handler {
let err = err.map_upgrade_err(|e| match e {
UpgradeError::Select(e) => UpgradeError::Select(e),
UpgradeError::Apply(EitherError::A(ioe)) => UpgradeError::Apply(ioe),
UpgradeError::Apply(EitherError::B(ioe)) => UpgradeError::Apply(ioe),
UpgradeError::Apply(Either::Left(ioe)) => UpgradeError::Apply(ioe),
UpgradeError::Apply(Either::Right(ioe)) => UpgradeError::Apply(ioe),
});
self.events
.push(ConnectionHandlerEvent::Custom(Event::IdentificationError(

9
protocols/relay/src/behaviour/handler.rs
View File

@ -31,7 +31,6 @@ use futures::stream::{FuturesUnordered, StreamExt};
use futures_timer::Delay;
use instant::Instant;
use libp2p_core::connection::ConnectionId;
use libp2p_core::either::EitherError;
use libp2p_core::{upgrade, ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
@ -397,7 +396,7 @@ pub struct Handler {
/// A pending fatal error that results in the connection being closed.
pending_error: Option<
ConnectionHandlerUpgrErr<
EitherError<inbound_hop::FatalUpgradeError, outbound_stop::FatalUpgradeError>,
Either<inbound_hop::FatalUpgradeError, outbound_stop::FatalUpgradeError>,
>,
>,
@ -521,7 +520,7 @@ impl Handler {
inbound_hop::UpgradeError::Fatal(error),
)) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(EitherError::A(error)),
upgrade::UpgradeError::Apply(Either::Left(error)),
));
return;
}
@ -572,7 +571,7 @@ impl Handler {
ConnectionHandlerUpgrErr::Upgrade(upgrade::UpgradeError::Apply(error)) => match error {
outbound_stop::UpgradeError::Fatal(error) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(EitherError::B(error)),
upgrade::UpgradeError::Apply(Either::Right(error)),
));
return;
}
@ -624,7 +623,7 @@ impl ConnectionHandler for Handler {
type InEvent = In;
type OutEvent = Event;
type Error = ConnectionHandlerUpgrErr<
EitherError<inbound_hop::FatalUpgradeError, outbound_stop::FatalUpgradeError>,
Either<inbound_hop::FatalUpgradeError, outbound_stop::FatalUpgradeError>,
>;
type InboundProtocol = inbound_hop::Upgrade;
type OutboundProtocol = outbound_stop::Upgrade;

11
protocols/relay/src/priv_client/handler.rs
View File

@ -28,7 +28,6 @@ use futures::sink::SinkExt;
use futures::stream::{FuturesUnordered, StreamExt};
use futures_timer::Delay;
use instant::Instant;
use libp2p_core::either::EitherError;
use libp2p_core::multiaddr::Protocol;
use libp2p_core::{upgrade, ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::handler::{
@ -174,7 +173,7 @@ pub struct Handler {
/// A pending fatal error that results in the connection being closed.
pending_error: Option<
ConnectionHandlerUpgrErr<
EitherError<inbound_stop::FatalUpgradeError, outbound_hop::FatalUpgradeError>,
Either<inbound_stop::FatalUpgradeError, outbound_hop::FatalUpgradeError>,
>,
>,
/// Until when to keep the connection alive.
@ -366,7 +365,7 @@ impl Handler {
inbound_stop::UpgradeError::Fatal(error),
)) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(EitherError::A(error)),
upgrade::UpgradeError::Apply(Either::Left(error)),
));
return;
}
@ -413,7 +412,7 @@ impl Handler {
match error {
outbound_hop::UpgradeError::Fatal(error) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(EitherError::B(error)),
upgrade::UpgradeError::Apply(Either::Right(error)),
));
return;
}
@ -476,7 +475,7 @@ impl Handler {
match error {
outbound_hop::UpgradeError::Fatal(error) => {
self.pending_error = Some(ConnectionHandlerUpgrErr::Upgrade(
upgrade::UpgradeError::Apply(EitherError::B(error)),
upgrade::UpgradeError::Apply(Either::Right(error)),
));
return;
}
@ -510,7 +509,7 @@ impl ConnectionHandler for Handler {
type InEvent = In;
type OutEvent = Event;
type Error = ConnectionHandlerUpgrErr<
EitherError<inbound_stop::FatalUpgradeError, outbound_hop::FatalUpgradeError>,
Either<inbound_stop::FatalUpgradeError, outbound_hop::FatalUpgradeError>,
>;
type InboundProtocol = inbound_stop::Upgrade;
type OutboundProtocol = outbound_hop::Upgrade;

6
swarm/src/behaviour/toggle.rs
View File

@ -28,7 +28,7 @@ use crate::upgrade::SendWrapper;
use crate::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use either::Either;
use libp2p_core::{
either::{EitherError, EitherOutput},
either::EitherOutput,
upgrade::{DeniedUpgrade, EitherUpgrade},
ConnectedPoint, Multiaddr, PeerId,
};
@ -215,8 +215,8 @@ where
ConnectionHandlerUpgrErr::Timer => ConnectionHandlerUpgrErr::Timer,
ConnectionHandlerUpgrErr::Upgrade(err) => {
ConnectionHandlerUpgrErr::Upgrade(err.map_err(|err| match err {
EitherError::A(e) => e,
EitherError::B(v) => void::unreachable(v),
Either::Left(e) => e,
Either::Right(v) => void::unreachable(v),
}))
}
};

10
swarm/src/handler/either.rs
View File

@ -25,7 +25,7 @@ use crate::handler::{
};
use crate::upgrade::SendWrapper;
use either::Either;
use libp2p_core::either::{EitherError, EitherOutput};
use libp2p_core::either::EitherOutput;
use libp2p_core::upgrade::{EitherUpgrade, UpgradeError};
use libp2p_core::{ConnectedPoint, PeerId};
use std::task::{Context, Poll};
@ -145,14 +145,14 @@ where
fn transpose(self) -> Either<DialUpgradeError<LOOI, LOP>, DialUpgradeError<ROOI, ROP>> {
match self {
DialUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(error))),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(error))),
info: Either::Left(info),
} => Either::Left(DialUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(error)),
info,
}),
DialUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(error))),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Right(error))),
info: Either::Right(info),
} => Either::Right(DialUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(error)),
@ -214,14 +214,14 @@ where
fn transpose(self) -> Either<ListenUpgradeError<LIOI, LIP>, ListenUpgradeError<RIOI, RIP>> {
match self {
ListenUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(error))),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(error))),
info: Either::Left(info),
} => Either::Left(ListenUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(error)),
info,
}),
ListenUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(error))),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Right(error))),
info: Either::Right(info),
} => Either::Right(ListenUpgradeError {
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(error)),

16
swarm/src/handler/select.rs
View File

@ -28,7 +28,7 @@ use crate::upgrade::SendWrapper;
use either::Either;
use libp2p_core::{
either::{EitherError, EitherOutput},
either::EitherOutput,
upgrade::{EitherUpgrade, NegotiationError, ProtocolError, SelectUpgrade, UpgradeError},
ConnectedPoint, PeerId,
};
@ -186,7 +186,7 @@ where
}),
DialUpgradeError {
info: EitherOutput::First(info),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(err))),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(err))),
} => Either::Left(DialUpgradeError {
info,
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(err)),
@ -214,7 +214,7 @@ where
}),
DialUpgradeError {
info: EitherOutput::Second(info),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(err))),
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Right(err))),
} => Either::Right(DialUpgradeError {
info,
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(err)),
@ -318,14 +318,14 @@ where
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Select(e2)),
}));
}
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(e))) => {
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Left(e))) => {
self.proto1
.on_connection_event(ConnectionEvent::ListenUpgradeError(ListenUpgradeError {
info: i1,
error: ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)),
}));
}
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(e))) => {
ConnectionHandlerUpgrErr::Upgrade(UpgradeError::Apply(Either::Right(e))) => {
self.proto2
.on_connection_event(ConnectionEvent::ListenUpgradeError(ListenUpgradeError {
info: i2,
@ -343,7 +343,7 @@ where
{
type InEvent = EitherOutput<TProto1::InEvent, TProto2::InEvent>;
type OutEvent = EitherOutput<TProto1::OutEvent, TProto2::OutEvent>;
type Error = EitherError<TProto1::Error, TProto2::Error>;
type Error = Either<TProto1::Error, TProto2::Error>;
type InboundProtocol = SelectUpgrade<
SendWrapper<<TProto1 as ConnectionHandler>::InboundProtocol>,
SendWrapper<<TProto2 as ConnectionHandler>::InboundProtocol>,
@ -395,7 +395,7 @@ where
return Poll::Ready(ConnectionHandlerEvent::Custom(EitherOutput::First(event)));
}
Poll::Ready(ConnectionHandlerEvent::Close(event)) => {
return Poll::Ready(ConnectionHandlerEvent::Close(EitherError::A(event)));
return Poll::Ready(ConnectionHandlerEvent::Close(Either::Left(event)));
}
Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => {
return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {
@ -412,7 +412,7 @@ where
return Poll::Ready(ConnectionHandlerEvent::Custom(EitherOutput::Second(event)));
}
Poll::Ready(ConnectionHandlerEvent::Close(event)) => {
return Poll::Ready(ConnectionHandlerEvent::Close(EitherError::B(event)));
return Poll::Ready(ConnectionHandlerEvent::Close(Either::Right(event)));
}
Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => {
return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {

11
swarm/src/lib.rs
View File

@ -1750,12 +1750,11 @@ fn p2p_addr(peer: Option<PeerId>, addr: Multiaddr) -> Result<Multiaddr, Multiadd
mod tests {
use super::*;
use crate::test::{CallTraceBehaviour, MockBehaviour};
use either::Either;
use futures::executor::block_on;
use futures::executor::ThreadPool;
use futures::future::poll_fn;
use futures::future::Either;
use futures::{executor, future, ready};
use libp2p_core::either::EitherError;
use libp2p_core::multiaddr::multiaddr;
use libp2p_core::transport::memory::MemoryTransportError;
use libp2p_core::transport::TransportEvent;
@ -2225,13 +2224,13 @@ mod tests {
match futures::future::select(transport.select_next_some(), swarm.next())
.await
{
Either::Left((TransportEvent::Incoming { .. }, _)) => {
future::Either::Left((TransportEvent::Incoming { .. }, _)) => {
break;
}
Either::Left(_) => {
future::Either::Left(_) => {
panic!("Unexpected transport event.")
}
Either::Right((e, _)) => {
future::Either::Right((e, _)) => {
panic!("Expect swarm to not emit any event {e:?}")
}
}
@ -2633,7 +2632,7 @@ mod tests {
"/ip4/127.0.0.1/tcp/80".parse().unwrap(),
TransportError::Other(io::Error::new(
io::ErrorKind::Other,
EitherError::<_, Void>::A(EitherError::<Void, _>::B(UpgradeError::Apply(
Either::<_, Void>::Left(Either::<Void, _>::Right(UpgradeError::Apply(
MemoryTransportError::Unreachable,
))),
)),