fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-29 18:51:22 +00:00
Code Issues Projects Releases Wiki Activity

refactor(core)!: remove EitherUpgrade (#3339)

Browse Source 
We don't need to define our own type here, we can simply implement `UpgradeInfo`, `InboundUpgrade` and `OutboundUpgrade` on `either::Either`.
This commit is contained in:
Thomas Eizinger 2023-01-18 13:35:07 +11:00 committed by GitHub
parent 8cd14e6a3a
commit db2cd43826
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 64 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
core/CHANGELOG.md
View File

@ -16,6 +16,8 @@
- Remove `EitherTransport` in favor of implementing `Transport` on `either::Either`. See [PR 3338]. - Remove `EitherTransport` in favor of implementing `Transport` on `either::Either`. See [PR 3338].
- Remove `EitherUpgrade` in favor of implementing `UpgradeInfo`, `InboundUpgrade` and `OutboundUpgrade` on `either::Either`. See [PR 3339].
[PR 3031]: https://github.com/libp2p/rust-libp2p/pull/3031 [PR 3031]: https://github.com/libp2p/rust-libp2p/pull/3031
[PR 3058]: https://github.com/libp2p/rust-libp2p/pull/3058 [PR 3058]: https://github.com/libp2p/rust-libp2p/pull/3058
[PR 3097]: https://github.com/libp2p/rust-libp2p/pull/3097 [PR 3097]: https://github.com/libp2p/rust-libp2p/pull/3097
@ -23,6 +25,7 @@
[PR 2972]: https://github.com/libp2p/rust-libp2p/pull/2972 [PR 2972]: https://github.com/libp2p/rust-libp2p/pull/2972
[PR 3337]: https://github.com/libp2p/rust-libp2p/pull/3337 [PR 3337]: https://github.com/libp2p/rust-libp2p/pull/3337
[PR 3338]: https://github.com/libp2p/rust-libp2p/pull/3338 [PR 3338]: https://github.com/libp2p/rust-libp2p/pull/3338
[PR 3339]: https://github.com/libp2p/rust-libp2p/pull/3339
# 0.37.0 # 0.37.0

1
core/src/upgrade.rs
View File

@ -74,7 +74,6 @@ use futures::future::Future;
pub use self::{ pub use self::{
apply::{apply, apply_inbound, apply_outbound, InboundUpgradeApply, OutboundUpgradeApply}, apply::{apply, apply_inbound, apply_outbound, InboundUpgradeApply, OutboundUpgradeApply},
denied::DeniedUpgrade, denied::DeniedUpgrade,
either::EitherUpgrade,
error::UpgradeError, error::UpgradeError,
from_fn::{from_fn, FromFnUpgrade}, from_fn::{from_fn, FromFnUpgrade},
map::{MapInboundUpgrade, MapInboundUpgradeErr, MapOutboundUpgrade, MapOutboundUpgradeErr}, map::{MapInboundUpgrade, MapInboundUpgradeErr, MapOutboundUpgrade, MapOutboundUpgradeErr},

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

@ -24,14 +24,7 @@ use crate::{
}; };
use either::Either; use either::Either;
/// A type to represent two possible upgrade types (inbound or outbound). impl<A, B> UpgradeInfo for Either<A, B>
#[derive(Debug, Clone)]
pub enum EitherUpgrade<A, B> {
A(A),
B(B),
}
impl<A, B> UpgradeInfo for EitherUpgrade<A, B>
where where
A: UpgradeInfo, A: UpgradeInfo,
B: UpgradeInfo, B: UpgradeInfo,
@ -44,13 +37,13 @@ where
fn protocol_info(&self) -> Self::InfoIter { fn protocol_info(&self) -> Self::InfoIter {
match self { match self {
EitherUpgrade::A(a) => EitherIter::A(a.protocol_info().into_iter()), Either::Left(a) => EitherIter::A(a.protocol_info().into_iter()),
EitherUpgrade::B(b) => EitherIter::B(b.protocol_info().into_iter()), Either::Right(b) => EitherIter::B(b.protocol_info().into_iter()),
} }
} }
} }
impl<C, A, B, TA, TB, EA, EB> InboundUpgrade<C> for EitherUpgrade<A, B> impl<C, A, B, TA, TB, EA, EB> InboundUpgrade<C> for Either<A, B>
where where
A: InboundUpgrade<C, Output = TA, Error = EA>, A: InboundUpgrade<C, Output = TA, Error = EA>,
B: InboundUpgrade<C, Output = TB, Error = EB>, B: InboundUpgrade<C, Output = TB, Error = EB>,
@ -61,10 +54,10 @@ where
fn upgrade_inbound(self, sock: C, info: Self::Info) -> Self::Future { fn upgrade_inbound(self, sock: C, info: Self::Info) -> Self::Future {
match (self, info) { match (self, info) {
(EitherUpgrade::A(a), EitherName::A(info)) => { (Either::Left(a), EitherName::A(info)) => {
EitherFuture2::A(a.upgrade_inbound(sock, info)) EitherFuture2::A(a.upgrade_inbound(sock, info))
} }
(EitherUpgrade::B(b), EitherName::B(info)) => { (Either::Right(b), EitherName::B(info)) => {
EitherFuture2::B(b.upgrade_inbound(sock, info)) EitherFuture2::B(b.upgrade_inbound(sock, info))
} }
_ => panic!("Invalid invocation of EitherUpgrade::upgrade_inbound"), _ => panic!("Invalid invocation of EitherUpgrade::upgrade_inbound"),
@ -72,7 +65,7 @@ where
} }
} }
impl<C, A, B, TA, TB, EA, EB> OutboundUpgrade<C> for EitherUpgrade<A, B> impl<C, A, B, TA, TB, EA, EB> OutboundUpgrade<C> for Either<A, B>
where where
A: OutboundUpgrade<C, Output = TA, Error = EA>, A: OutboundUpgrade<C, Output = TA, Error = EA>,
B: OutboundUpgrade<C, Output = TB, Error = EB>, B: OutboundUpgrade<C, Output = TB, Error = EB>,
@ -83,10 +76,10 @@ where
fn upgrade_outbound(self, sock: C, info: Self::Info) -> Self::Future { fn upgrade_outbound(self, sock: C, info: Self::Info) -> Self::Future {
match (self, info) { match (self, info) {
(EitherUpgrade::A(a), EitherName::A(info)) => { (Either::Left(a), EitherName::A(info)) => {
EitherFuture2::A(a.upgrade_outbound(sock, info)) EitherFuture2::A(a.upgrade_outbound(sock, info))
} }
(EitherUpgrade::B(b), EitherName::B(info)) => { (Either::Right(b), EitherName::B(info)) => {
EitherFuture2::B(b.upgrade_outbound(sock, info)) EitherFuture2::B(b.upgrade_outbound(sock, info))
} }
_ => panic!("Invalid invocation of EitherUpgrade::upgrade_outbound"), _ => panic!("Invalid invocation of EitherUpgrade::upgrade_outbound"),

10
protocols/dcutr/src/handler.rs
View File

@ -21,7 +21,7 @@
use crate::protocol; use crate::protocol;
use either::Either; use either::Either;
use libp2p_core::connection::ConnectionId; use libp2p_core::connection::ConnectionId;
use libp2p_core::upgrade::{self, DeniedUpgrade}; use libp2p_core::upgrade::DeniedUpgrade;
use libp2p_core::{ConnectedPoint, PeerId}; use libp2p_core::{ConnectedPoint, PeerId};
use libp2p_swarm::dummy; use libp2p_swarm::dummy;
use libp2p_swarm::handler::SendWrapper; use libp2p_swarm::handler::SendWrapper;
@ -70,11 +70,11 @@ impl IntoConnectionHandler for Prototype {
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol { fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
match self { match self {
Prototype::UnknownConnection => upgrade::EitherUpgrade::A(SendWrapper( Prototype::UnknownConnection => {
upgrade::EitherUpgrade::A(protocol::inbound::Upgrade {}), Either::Left(SendWrapper(Either::Left(protocol::inbound::Upgrade {})))
)), }
Prototype::DirectConnection { .. } => { Prototype::DirectConnection { .. } => {
upgrade::EitherUpgrade::A(SendWrapper(upgrade::EitherUpgrade::B(DeniedUpgrade))) Either::Left(SendWrapper(Either::Right(DeniedUpgrade)))
} }
} }
} }

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

@ -26,7 +26,7 @@ use futures::future::{BoxFuture, FutureExt};
use instant::Instant; use instant::Instant;
use libp2p_core::either::EitherOutput; use libp2p_core::either::EitherOutput;
use libp2p_core::multiaddr::Multiaddr; use libp2p_core::multiaddr::Multiaddr;
use libp2p_core::upgrade::{self, DeniedUpgrade, NegotiationError, UpgradeError}; use libp2p_core::upgrade::{DeniedUpgrade, NegotiationError, UpgradeError};
use libp2p_core::ConnectedPoint; use libp2p_core::ConnectedPoint;
use libp2p_swarm::handler::{ use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound, ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
@ -305,7 +305,7 @@ impl ConnectionHandler for Handler {
type Error = ConnectionHandlerUpgrErr< type Error = ConnectionHandlerUpgrErr<
Either<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>, Either<protocol::inbound::UpgradeError, protocol::outbound::UpgradeError>,
>; >;
type InboundProtocol = upgrade::EitherUpgrade<protocol::inbound::Upgrade, DeniedUpgrade>; type InboundProtocol = Either<protocol::inbound::Upgrade, DeniedUpgrade>;
type OutboundProtocol = protocol::outbound::Upgrade; type OutboundProtocol = protocol::outbound::Upgrade;
type OutboundOpenInfo = u8; // Number of upgrade attempts. type OutboundOpenInfo = u8; // Number of upgrade attempts.
type InboundOpenInfo = (); type InboundOpenInfo = ();
@ -313,7 +313,7 @@ impl ConnectionHandler for Handler {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> { fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
match self.endpoint { match self.endpoint {
ConnectedPoint::Dialer { .. } => { ConnectedPoint::Dialer { .. } => {
SubstreamProtocol::new(upgrade::EitherUpgrade::A(protocol::inbound::Upgrade {}), ()) SubstreamProtocol::new(Either::Left(protocol::inbound::Upgrade {}), ())
} }
ConnectedPoint::Listener { .. } => { ConnectedPoint::Listener { .. } => {
// By the protocol specification the listening side of a relayed connection // By the protocol specification the listening side of a relayed connection
@ -321,7 +321,7 @@ impl ConnectionHandler for Handler {
// the relayed connection opens a substream to the dialing side. (Connection roles // the relayed connection opens a substream to the dialing side. (Connection roles
// and substream roles are reversed.) The listening side on a relayed connection // and substream roles are reversed.) The listening side on a relayed connection
// never expects incoming substreams, hence the denied upgrade below. // never expects incoming substreams, hence the denied upgrade below.
SubstreamProtocol::new(upgrade::EitherUpgrade::B(DeniedUpgrade), ()) SubstreamProtocol::new(Either::Right(DeniedUpgrade), ())
} }
} }
} }

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

@ -27,7 +27,7 @@ use futures::prelude::*;
use futures::stream::FuturesUnordered; use futures::stream::FuturesUnordered;
use futures_timer::Delay; use futures_timer::Delay;
use libp2p_core::either::EitherOutput; use libp2p_core::either::EitherOutput;
use libp2p_core::upgrade::{EitherUpgrade, SelectUpgrade}; use libp2p_core::upgrade::SelectUpgrade;
use libp2p_core::{ConnectedPoint, Multiaddr, PeerId, PublicKey}; use libp2p_core::{ConnectedPoint, Multiaddr, PeerId, PublicKey};
use libp2p_swarm::handler::{ use libp2p_swarm::handler::{
ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound, ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound,
@ -102,8 +102,7 @@ pub struct Handler {
inbound_identify_push: Option<BoxFuture<'static, Result<Info, UpgradeError>>>, inbound_identify_push: Option<BoxFuture<'static, Result<Info, UpgradeError>>>,
/// Pending events to yield. /// Pending events to yield.
events: SmallVec< events: SmallVec<
[ConnectionHandlerEvent<EitherUpgrade<Identify, Push<OutboundPush>>, (), Event, io::Error>; [ConnectionHandlerEvent<Either<Identify, Push<OutboundPush>>, (), Event, io::Error>; 4],
4],
>, >,
/// Streams awaiting `BehaviourInfo` to then send identify requests. /// Streams awaiting `BehaviourInfo` to then send identify requests.
@ -277,7 +276,7 @@ impl ConnectionHandler for Handler {
type OutEvent = Event; type OutEvent = Event;
type Error = io::Error; type Error = io::Error;
type InboundProtocol = SelectUpgrade<Identify, Push<InboundPush>>; type InboundProtocol = SelectUpgrade<Identify, Push<InboundPush>>;
type OutboundProtocol = EitherUpgrade<Identify, Push<OutboundPush>>; type OutboundProtocol = Either<Identify, Push<OutboundPush>>;
type OutboundOpenInfo = (); type OutboundOpenInfo = ();
type InboundOpenInfo = (); type InboundOpenInfo = ();
@ -306,10 +305,7 @@ impl ConnectionHandler for Handler {
Protocol::Push => { Protocol::Push => {
self.events self.events
.push(ConnectionHandlerEvent::OutboundSubstreamRequest { .push(ConnectionHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new( protocol: SubstreamProtocol::new(Either::Right(Push::outbound(info)), ()),
EitherUpgrade::B(Push::outbound(info)),
(),
),
}); });
} }
Protocol::Identify(_) => { Protocol::Identify(_) => {
@ -347,7 +343,7 @@ impl ConnectionHandler for Handler {
Poll::Ready(()) => { Poll::Ready(()) => {
self.trigger_next_identify.reset(self.interval); self.trigger_next_identify.reset(self.interval);
let ev = ConnectionHandlerEvent::OutboundSubstreamRequest { let ev = ConnectionHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(EitherUpgrade::A(Identify), ()), protocol: SubstreamProtocol::new(Either::Left(Identify), ()),
}; };
return Poll::Ready(ev); return Poll::Ready(ev);
} }

11
protocols/kad/src/handler.rs
View File

@ -23,6 +23,7 @@ use crate::protocol::{
KademliaProtocolConfig, KademliaProtocolConfig,
}; };
use crate::record::{self, Record}; use crate::record::{self, Record};
use either::Either;
use futures::prelude::*; use futures::prelude::*;
use futures::stream::SelectAll; use futures::stream::SelectAll;
use instant::Instant; use instant::Instant;
@ -68,9 +69,9 @@ impl<T: Clone + fmt::Debug + Send + 'static + Unpin> IntoConnectionHandler
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol { fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
if self.config.allow_listening { if self.config.allow_listening {
upgrade::EitherUpgrade::A(self.config.protocol_config.clone()) Either::Left(self.config.protocol_config.clone())
} else { } else {
upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade) Either::Right(upgrade::DeniedUpgrade)
} }
} }
} }
@ -633,7 +634,7 @@ where
type InEvent = KademliaHandlerIn<TUserData>; type InEvent = KademliaHandlerIn<TUserData>;
type OutEvent = KademliaHandlerEvent<TUserData>; type OutEvent = KademliaHandlerEvent<TUserData>;
type Error = io::Error; // TODO: better error type? type Error = io::Error; // TODO: better error type?
type InboundProtocol = upgrade::EitherUpgrade<KademliaProtocolConfig, upgrade::DeniedUpgrade>; type InboundProtocol = Either<KademliaProtocolConfig, upgrade::DeniedUpgrade>;
type OutboundProtocol = KademliaProtocolConfig; type OutboundProtocol = KademliaProtocolConfig;
// Message of the request to send to the remote, and user data if we expect an answer. // Message of the request to send to the remote, and user data if we expect an answer.
type OutboundOpenInfo = (KadRequestMsg, Option<TUserData>); type OutboundOpenInfo = (KadRequestMsg, Option<TUserData>);
@ -642,9 +643,9 @@ where
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> { fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
if self.config.allow_listening { if self.config.allow_listening {
SubstreamProtocol::new(self.config.protocol_config.clone(), ()) SubstreamProtocol::new(self.config.protocol_config.clone(), ())
.map_upgrade(upgrade::EitherUpgrade::A) .map_upgrade(Either::Left)
} else { } else {
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ()) SubstreamProtocol::new(Either::Right(upgrade::DeniedUpgrade), ())
} }
} }

1
protocols/ping/Cargo.toml
View File

@ -11,6 +11,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
categories = ["network-programming", "asynchronous"] categories = ["network-programming", "asynchronous"]
[dependencies] [dependencies]
either = "1.8.0"
futures = "0.3.1" futures = "0.3.1"
futures-timer = "3.0.2" futures-timer = "3.0.2"
instant = "0.1.11" instant = "0.1.11"

5
protocols/ping/tests/ping.rs
View File

@ -20,6 +20,7 @@
//! Integration tests for the `Ping` network behaviour. //! Integration tests for the `Ping` network behaviour.
use either::Either;
use futures::{channel::mpsc, prelude::*}; use futures::{channel::mpsc, prelude::*};
use libp2p_core::{ use libp2p_core::{
identity, identity,
@ -251,8 +252,8 @@ fn mk_transport(muxer: MuxerChoice) -> (PeerId, transport::Boxed<(PeerId, Stream
.upgrade(upgrade::Version::V1) .upgrade(upgrade::Version::V1)
.authenticate(noise::NoiseAuthenticated::xx(&id_keys).unwrap()) .authenticate(noise::NoiseAuthenticated::xx(&id_keys).unwrap())
.multiplex(match muxer { .multiplex(match muxer {
MuxerChoice::Yamux => upgrade::EitherUpgrade::A(yamux::YamuxConfig::default()), MuxerChoice::Yamux => Either::Left(yamux::YamuxConfig::default()),
MuxerChoice::Mplex => upgrade::EitherUpgrade::B(mplex::MplexConfig::default()), MuxerChoice::Mplex => Either::Right(mplex::MplexConfig::default()),
}) })
.boxed(), .boxed(),
) )

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

@ -355,7 +355,7 @@ impl IntoConnectionHandler for Prototype {
} }
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol { fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
upgrade::EitherUpgrade::A(SendWrapper(inbound_hop::Upgrade { Either::Left(SendWrapper(inbound_hop::Upgrade {
reservation_duration: self.config.reservation_duration, reservation_duration: self.config.reservation_duration,
max_circuit_duration: self.config.max_circuit_duration, max_circuit_duration: self.config.max_circuit_duration,
max_circuit_bytes: self.config.max_circuit_bytes, max_circuit_bytes: self.config.max_circuit_bytes,

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

@ -155,7 +155,7 @@ impl IntoConnectionHandler for Prototype {
} }
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol { fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
upgrade::EitherUpgrade::A(SendWrapper(inbound_stop::Upgrade {})) Either::Left(SendWrapper(inbound_stop::Upgrade {}))
} }
} }

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

@ -28,9 +28,7 @@ use crate::upgrade::SendWrapper;
use crate::{NetworkBehaviour, NetworkBehaviourAction, PollParameters}; use crate::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use either::Either; use either::Either;
use libp2p_core::{ use libp2p_core::{
either::EitherOutput, either::EitherOutput, upgrade::DeniedUpgrade, ConnectedPoint, Multiaddr, PeerId,
upgrade::{DeniedUpgrade, EitherUpgrade},
ConnectedPoint, Multiaddr, PeerId,
}; };
use std::{task::Context, task::Poll}; use std::{task::Context, task::Poll};
@ -143,9 +141,9 @@ where
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol { fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
if let Some(inner) = self.inner.as_ref() { if let Some(inner) = self.inner.as_ref() {
EitherUpgrade::A(SendWrapper(inner.inbound_protocol())) Either::Left(SendWrapper(inner.inbound_protocol()))
} else { } else {
EitherUpgrade::B(SendWrapper(DeniedUpgrade)) Either::Right(SendWrapper(DeniedUpgrade))
} }
} }
} }
@ -235,8 +233,7 @@ where
type InEvent = TInner::InEvent; type InEvent = TInner::InEvent;
type OutEvent = TInner::OutEvent; type OutEvent = TInner::OutEvent;
type Error = TInner::Error; type Error = TInner::Error;
type InboundProtocol = type InboundProtocol = Either<SendWrapper<TInner::InboundProtocol>, SendWrapper<DeniedUpgrade>>;
EitherUpgrade<SendWrapper<TInner::InboundProtocol>, SendWrapper<DeniedUpgrade>>;
type OutboundProtocol = TInner::OutboundProtocol; type OutboundProtocol = TInner::OutboundProtocol;
type OutboundOpenInfo = TInner::OutboundOpenInfo; type OutboundOpenInfo = TInner::OutboundOpenInfo;
type InboundOpenInfo = Either<TInner::InboundOpenInfo, ()>; type InboundOpenInfo = Either<TInner::InboundOpenInfo, ()>;
@ -245,13 +242,10 @@ where
if let Some(inner) = self.inner.as_ref() { if let Some(inner) = self.inner.as_ref() {
inner inner
.listen_protocol() .listen_protocol()
.map_upgrade(|u| EitherUpgrade::A(SendWrapper(u))) .map_upgrade(|u| Either::Left(SendWrapper(u)))
.map_info(Either::Left) .map_info(Either::Left)
} else { } else {
SubstreamProtocol::new( SubstreamProtocol::new(Either::Right(SendWrapper(DeniedUpgrade)), Either::Right(()))
EitherUpgrade::B(SendWrapper(DeniedUpgrade)),
Either::Right(()),
)
} }
} }

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

@ -26,7 +26,7 @@ use crate::handler::{
use crate::upgrade::SendWrapper; use crate::upgrade::SendWrapper;
use either::Either; use either::Either;
use libp2p_core::either::EitherOutput; use libp2p_core::either::EitherOutput;
use libp2p_core::upgrade::{EitherUpgrade, UpgradeError}; use libp2p_core::upgrade::UpgradeError;
use libp2p_core::{ConnectedPoint, PeerId}; use libp2p_core::{ConnectedPoint, PeerId};
use std::task::{Context, Poll}; use std::task::{Context, Poll};
@ -58,10 +58,10 @@ where
fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol { fn inbound_protocol(&self) -> <Self::Handler as ConnectionHandler>::InboundProtocol {
match self { match self {
IntoEitherHandler::Left(into_handler) => { IntoEitherHandler::Left(into_handler) => {
EitherUpgrade::A(SendWrapper(into_handler.inbound_protocol())) Either::Left(SendWrapper(into_handler.inbound_protocol()))
} }
IntoEitherHandler::Right(into_handler) => { IntoEitherHandler::Right(into_handler) => {
EitherUpgrade::B(SendWrapper(into_handler.inbound_protocol())) Either::Right(SendWrapper(into_handler.inbound_protocol()))
} }
} }
} }
@ -91,7 +91,7 @@ impl<L, R> IntoEitherHandler<L, R> {
} }
impl<LIP, RIP, LIOI, RIOI> impl<LIP, RIP, LIOI, RIOI>
FullyNegotiatedInbound<EitherUpgrade<SendWrapper<LIP>, SendWrapper<RIP>>, Either<LIOI, RIOI>> FullyNegotiatedInbound<Either<SendWrapper<LIP>, SendWrapper<RIP>>, Either<LIOI, RIOI>>
where where
RIP: InboundUpgradeSend, RIP: InboundUpgradeSend,
LIP: InboundUpgradeSend, LIP: InboundUpgradeSend,
@ -114,7 +114,7 @@ where
} }
impl<LOP, ROP, LOOI, ROOI> impl<LOP, ROP, LOOI, ROOI>
FullyNegotiatedOutbound<EitherUpgrade<SendWrapper<LOP>, SendWrapper<ROP>>, Either<LOOI, ROOI>> FullyNegotiatedOutbound<Either<SendWrapper<LOP>, SendWrapper<ROP>>, Either<LOOI, ROOI>>
where where
LOP: OutboundUpgradeSend, LOP: OutboundUpgradeSend,
ROP: OutboundUpgradeSend, ROP: OutboundUpgradeSend,
@ -137,7 +137,7 @@ where
} }
impl<LOP, ROP, LOOI, ROOI> impl<LOP, ROP, LOOI, ROOI>
DialUpgradeError<Either<LOOI, ROOI>, EitherUpgrade<SendWrapper<LOP>, SendWrapper<ROP>>> DialUpgradeError<Either<LOOI, ROOI>, Either<SendWrapper<LOP>, SendWrapper<ROP>>>
where where
LOP: OutboundUpgradeSend, LOP: OutboundUpgradeSend,
ROP: OutboundUpgradeSend, ROP: OutboundUpgradeSend,
@ -206,7 +206,7 @@ where
} }
impl<LIP, RIP, LIOI, RIOI> impl<LIP, RIP, LIOI, RIOI>
ListenUpgradeError<Either<LIOI, RIOI>, EitherUpgrade<SendWrapper<LIP>, SendWrapper<RIP>>> ListenUpgradeError<Either<LIOI, RIOI>, Either<SendWrapper<LIP>, SendWrapper<RIP>>>
where where
RIP: InboundUpgradeSend, RIP: InboundUpgradeSend,
LIP: InboundUpgradeSend, LIP: InboundUpgradeSend,
@ -284,10 +284,9 @@ where
type InEvent = Either<L::InEvent, R::InEvent>; type InEvent = Either<L::InEvent, R::InEvent>;
type OutEvent = Either<L::OutEvent, R::OutEvent>; type OutEvent = Either<L::OutEvent, R::OutEvent>;
type Error = Either<L::Error, R::Error>; type Error = Either<L::Error, R::Error>;
type InboundProtocol = type InboundProtocol = Either<SendWrapper<L::InboundProtocol>, SendWrapper<R::InboundProtocol>>;
EitherUpgrade<SendWrapper<L::InboundProtocol>, SendWrapper<R::InboundProtocol>>;
type OutboundProtocol = type OutboundProtocol =
EitherUpgrade<SendWrapper<L::OutboundProtocol>, SendWrapper<R::OutboundProtocol>>; Either<SendWrapper<L::OutboundProtocol>, SendWrapper<R::OutboundProtocol>>;
type InboundOpenInfo = Either<L::InboundOpenInfo, R::InboundOpenInfo>; type InboundOpenInfo = Either<L::InboundOpenInfo, R::InboundOpenInfo>;
type OutboundOpenInfo = Either<L::OutboundOpenInfo, R::OutboundOpenInfo>; type OutboundOpenInfo = Either<L::OutboundOpenInfo, R::OutboundOpenInfo>;
@ -295,11 +294,11 @@ where
match self { match self {
Either::Left(a) => a Either::Left(a) => a
.listen_protocol() .listen_protocol()
.map_upgrade(|u| EitherUpgrade::A(SendWrapper(u))) .map_upgrade(|u| Either::Left(SendWrapper(u)))
.map_info(Either::Left), .map_info(Either::Left),
Either::Right(b) => b Either::Right(b) => b
.listen_protocol() .listen_protocol()
.map_upgrade(|u| EitherUpgrade::B(SendWrapper(u))) .map_upgrade(|u| Either::Right(SendWrapper(u)))
.map_info(Either::Right), .map_info(Either::Right),
} }
} }
@ -334,12 +333,12 @@ where
Either::Left(handler) => futures::ready!(handler.poll(cx)) Either::Left(handler) => futures::ready!(handler.poll(cx))
.map_custom(Either::Left) .map_custom(Either::Left)
.map_close(Either::Left) .map_close(Either::Left)
.map_protocol(|p| EitherUpgrade::A(SendWrapper(p))) .map_protocol(|p| Either::Left(SendWrapper(p)))
.map_outbound_open_info(Either::Left), .map_outbound_open_info(Either::Left),
Either::Right(handler) => futures::ready!(handler.poll(cx)) Either::Right(handler) => futures::ready!(handler.poll(cx))
.map_custom(Either::Right) .map_custom(Either::Right)
.map_close(Either::Right) .map_close(Either::Right)
.map_protocol(|p| EitherUpgrade::B(SendWrapper(p))) .map_protocol(|p| Either::Right(SendWrapper(p)))
.map_outbound_open_info(Either::Right), .map_outbound_open_info(Either::Right),
}; };

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

@ -29,7 +29,7 @@ use crate::upgrade::SendWrapper;
use either::Either; use either::Either;
use libp2p_core::{ use libp2p_core::{
either::EitherOutput, either::EitherOutput,
upgrade::{EitherUpgrade, NegotiationError, ProtocolError, SelectUpgrade, UpgradeError}, upgrade::{NegotiationError, ProtocolError, SelectUpgrade, UpgradeError},
ConnectedPoint, PeerId, ConnectedPoint, PeerId,
}; };
use std::{cmp, task::Context, task::Poll}; use std::{cmp, task::Context, task::Poll};
@ -102,7 +102,7 @@ impl<TProto1, TProto2> ConnectionHandlerSelect<TProto1, TProto2> {
impl<S1OOI, S2OOI, S1OP, S2OP> impl<S1OOI, S2OOI, S1OP, S2OP>
FullyNegotiatedOutbound< FullyNegotiatedOutbound<
EitherUpgrade<SendWrapper<S1OP>, SendWrapper<S2OP>>, Either<SendWrapper<S1OP>, SendWrapper<S2OP>>,
EitherOutput<S1OOI, S2OOI>, EitherOutput<S1OOI, S2OOI>,
> >
where where
@ -151,10 +151,7 @@ where
} }
impl<S1OOI, S2OOI, S1OP, S2OP> impl<S1OOI, S2OOI, S1OP, S2OP>
DialUpgradeError< DialUpgradeError<EitherOutput<S1OOI, S2OOI>, Either<SendWrapper<S1OP>, SendWrapper<S2OP>>>
EitherOutput<S1OOI, S2OOI>,
EitherUpgrade<SendWrapper<S1OP>, SendWrapper<S2OP>>,
>
where where
S1OP: OutboundUpgradeSend, S1OP: OutboundUpgradeSend,
S2OP: OutboundUpgradeSend, S2OP: OutboundUpgradeSend,
@ -348,10 +345,8 @@ where
SendWrapper<<TProto1 as ConnectionHandler>::InboundProtocol>, SendWrapper<<TProto1 as ConnectionHandler>::InboundProtocol>,
SendWrapper<<TProto2 as ConnectionHandler>::InboundProtocol>, SendWrapper<<TProto2 as ConnectionHandler>::InboundProtocol>,
>; >;
type OutboundProtocol = EitherUpgrade< type OutboundProtocol =
SendWrapper<TProto1::OutboundProtocol>, Either<SendWrapper<TProto1::OutboundProtocol>, SendWrapper<TProto2::OutboundProtocol>>;
SendWrapper<TProto2::OutboundProtocol>,
>;
type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>; type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>;
type InboundOpenInfo = (TProto1::InboundOpenInfo, TProto2::InboundOpenInfo); type InboundOpenInfo = (TProto1::InboundOpenInfo, TProto2::InboundOpenInfo);
@ -400,7 +395,7 @@ where
Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => { Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => {
return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {
protocol: protocol protocol: protocol
.map_upgrade(|u| EitherUpgrade::A(SendWrapper(u))) .map_upgrade(|u| Either::Left(SendWrapper(u)))
.map_info(EitherOutput::First), .map_info(EitherOutput::First),
}); });
} }
@ -417,7 +412,7 @@ where
Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => { Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => {
return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {
protocol: protocol protocol: protocol
.map_upgrade(|u| EitherUpgrade::B(SendWrapper(u))) .map_upgrade(|u| Either::Right(SendWrapper(u)))
.map_info(EitherOutput::Second), .map_info(EitherOutput::Second),
}); });
} }