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

Add EitherUpgrade and generalise OrUpgrade. (#662)

Browse Source
This commit is contained in:
Toralf Wittner 2018-11-20 15:09:59 +01:00 committed by GitHub
parent b213fd7bd7
commit 938b91742f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 179 additions and 368 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
core/src/either.rs
View File

@ -316,3 +316,29 @@ where
} }
} }
} }
#[derive(Debug, Copy, Clone)]
#[must_use = "futures do nothing unless polled"]
pub enum EitherFuture2<A, B> { A(A), B(B) }
impl<AFut, BFut, AItem, BItem, AError, BError> Future for EitherFuture2<AFut, BFut>
where
AFut: Future<Item = AItem, Error = AError>,
BFut: Future<Item = BItem, Error = BError>
{
type Item = EitherOutput<AItem, BItem>;
type Error = EitherError<AError, BError>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self {
EitherFuture2::A(a) => a.poll()
.map(|v| v.map(EitherOutput::First))
.map_err(|e| EitherError::A(e)),
EitherFuture2::B(b) => b.poll()
.map(|v| v.map(EitherOutput::Second))
.map_err(|e| EitherError::B(e))
}
}
}

170
core/src/protocols_handler.rs
View File

@ -19,18 +19,19 @@
// DEALINGS IN THE SOFTWARE. // DEALINGS IN THE SOFTWARE.
use crate::{ use crate::{
either::{EitherError, EitherOutput}, either::EitherOutput,
nodes::handled_node::{NodeHandler, NodeHandlerEndpoint, NodeHandlerEvent}, nodes::handled_node::{NodeHandler, NodeHandlerEndpoint, NodeHandlerEvent},
upgrade::{ upgrade::{
self, self,
InboundUpgrade, InboundUpgrade,
InboundUpgradeExt, InboundUpgradeExt,
OutboundUpgrade, OutboundUpgrade,
OutboundUpgradeExt,
UpgradeInfo, UpgradeInfo,
InboundUpgradeApply, InboundUpgradeApply,
OutboundUpgradeApply, OutboundUpgradeApply,
DeniedUpgrade DeniedUpgrade,
EitherUpgrade,
OrUpgrade
} }
}; };
use futures::prelude::*; use futures::prelude::*;
@ -107,8 +108,6 @@ pub trait ProtocolsHandler {
/// > list of supported protocols in a cache in order to avoid spurious queries. /// > list of supported protocols in a cache in order to avoid spurious queries.
fn listen_protocol(&self) -> Self::InboundProtocol; fn listen_protocol(&self) -> Self::InboundProtocol;
fn dialer_protocol(&self) -> Self::OutboundProtocol;
/// Injects a fully-negotiated substream in the handler. /// Injects a fully-negotiated substream in the handler.
/// ///
/// This method is called when a substream has been successfully opened and negotiated. /// This method is called when a substream has been successfully opened and negotiated.
@ -320,11 +319,6 @@ where
DeniedUpgrade DeniedUpgrade
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
DeniedUpgrade
}
#[inline] #[inline]
fn inject_fully_negotiated_inbound( fn inject_fully_negotiated_inbound(
&mut self, &mut self,
@ -393,11 +387,6 @@ where
self.inner.listen_protocol() self.inner.listen_protocol()
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
self.inner.dialer_protocol()
}
#[inline] #[inline]
fn inject_fully_negotiated_inbound( fn inject_fully_negotiated_inbound(
&mut self, &mut self,
@ -471,11 +460,6 @@ where
self.inner.listen_protocol() self.inner.listen_protocol()
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
self.inner.dialer_protocol()
}
#[inline] #[inline]
fn inject_fully_negotiated_inbound( fn inject_fully_negotiated_inbound(
&mut self, &mut self,
@ -766,105 +750,15 @@ where
type InEvent = EitherOutput<TProto1::InEvent, TProto2::InEvent>; type InEvent = EitherOutput<TProto1::InEvent, TProto2::InEvent>;
type OutEvent = EitherOutput<TProto1::OutEvent, TProto2::OutEvent>; type OutEvent = EitherOutput<TProto1::OutEvent, TProto2::OutEvent>;
type Substream = TSubstream; type Substream = TSubstream;
type InboundProtocol = OrUpgrade<TProto1::InboundProtocol, TProto2::InboundProtocol>;
type InboundProtocol = type OutboundProtocol = EitherUpgrade<TProto1::OutboundProtocol, TProto2::OutboundProtocol>;
upgrade::OrUpgrade<
upgrade::Toggleable<
upgrade::MapInboundUpgradeErr<
upgrade::MapInboundUpgrade<
TProto1::InboundProtocol,
fn(TProto1Out) -> EitherOutput<TProto1Out, TProto2Out>
>,
fn(<TProto1::InboundProtocol as InboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::InboundProtocol as InboundUpgrade<TSubstream>>::Error,
<TProto2::InboundProtocol as InboundUpgrade<TSubstream>>::Error
>
>
>,
upgrade::Toggleable<
upgrade::MapInboundUpgradeErr<
upgrade::MapInboundUpgrade<
TProto2::InboundProtocol,
fn(TProto2Out) -> EitherOutput<TProto1Out, TProto2Out>
>,
fn(<TProto2::InboundProtocol as InboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::InboundProtocol as InboundUpgrade<TSubstream>>::Error,
<TProto2::InboundProtocol as InboundUpgrade<TSubstream>>::Error
>
>
>
>;
type OutboundProtocol =
upgrade::OrUpgrade<
upgrade::Toggleable<
upgrade::MapOutboundUpgradeErr<
upgrade::MapOutboundUpgrade<
TProto1::OutboundProtocol,
fn(TProto1Out) -> EitherOutput<TProto1Out, TProto2Out>
>,
fn(<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>
>
>,
upgrade::Toggleable<
upgrade::MapOutboundUpgradeErr<
upgrade::MapOutboundUpgrade<
TProto2::OutboundProtocol,
fn(TProto2Out) -> EitherOutput<TProto1Out, TProto2Out>
>,
fn(<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>
>
>
>;
type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>; type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>;
#[inline] #[inline]
fn listen_protocol(&self) -> Self::InboundProtocol { fn listen_protocol(&self) -> Self::InboundProtocol {
let proto1 = self.proto1.listen_protocol() let proto1 = self.proto1.listen_protocol();
.map_inbound(EitherOutput::First as fn(TProto1Out) -> EitherOutput<TProto1Out, TProto2Out>) let proto2 = self.proto2.listen_protocol();
.map_inbound_err(EitherError::A as fn(<TProto1::InboundProtocol as InboundUpgrade<TSubstream>>::Error) -> proto1.or_inbound(proto2)
EitherError<
<TProto1::InboundProtocol as InboundUpgrade<TSubstream>>::Error,
<TProto2::InboundProtocol as InboundUpgrade<TSubstream>>::Error
>);
let proto2 = self.proto2.listen_protocol()
.map_inbound(EitherOutput::Second as fn(TProto2Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_inbound_err(EitherError::B as fn(<TProto2::InboundProtocol as InboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::InboundProtocol as InboundUpgrade<TSubstream>>::Error,
<TProto2::InboundProtocol as InboundUpgrade<TSubstream>>::Error
>);
upgrade::toggleable(proto1).or_inbound(upgrade::toggleable(proto2))
}
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
let proto1 = self.proto1.dialer_protocol()
.map_outbound(EitherOutput::First as fn(TProto1Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_outbound_err(EitherError::A as fn(<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>);
let proto2 = self.proto2.dialer_protocol()
.map_outbound(EitherOutput::Second as fn(TProto2Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_outbound_err(EitherError::B as fn(<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>);
upgrade::toggleable(proto1).or_outbound(upgrade::toggleable(proto2))
} }
fn inject_fully_negotiated_outbound(&mut self, protocol: <Self::OutboundProtocol as OutboundUpgrade<TSubstream>>::Output, endpoint: Self::OutboundOpenInfo) { fn inject_fully_negotiated_outbound(&mut self, protocol: <Self::OutboundProtocol as OutboundUpgrade<TSubstream>>::Output, endpoint: Self::OutboundOpenInfo) {
@ -923,29 +817,8 @@ where
return Ok(Async::Ready(Some(ProtocolsHandlerEvent::Custom(EitherOutput::First(event))))); return Ok(Async::Ready(Some(ProtocolsHandlerEvent::Custom(EitherOutput::First(event)))));
}, },
Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest { upgrade, info})) => { Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest { upgrade, info})) => {
let upgrade = {
let proto1 = upgrade
.map_outbound(EitherOutput::First as fn(TProto1Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_outbound_err(EitherError::A as fn(<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>);
let proto2 = self.proto2.dialer_protocol()
.map_outbound(EitherOutput::Second as fn(TProto2Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_outbound_err(EitherError::B as fn(<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>);
let proto1 = upgrade::toggleable(proto1);
let mut proto2 = upgrade::toggleable(proto2);
proto2.disable();
proto1.or_outbound(proto2)
};
return Ok(Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest { return Ok(Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest {
upgrade, upgrade: EitherUpgrade::A(upgrade),
info: EitherOutput::First(info), info: EitherOutput::First(info),
}))); })));
}, },
@ -958,29 +831,8 @@ where
return Ok(Async::Ready(Some(ProtocolsHandlerEvent::Custom(EitherOutput::Second(event))))); return Ok(Async::Ready(Some(ProtocolsHandlerEvent::Custom(EitherOutput::Second(event)))));
}, },
Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest { upgrade, info })) => { Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest { upgrade, info })) => {
let upgrade = {
let proto1 = self.proto1.dialer_protocol()
.map_outbound(EitherOutput::First as fn(TProto1Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_outbound_err(EitherError::A as fn(<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>);
let proto2 = upgrade
.map_outbound(EitherOutput::Second as fn(TProto2Out) -> EitherOutput<TProto1Out, TProto2Out>)
.map_outbound_err(EitherError::B as fn(<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error) ->
EitherError<
<TProto1::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error,
<TProto2::OutboundProtocol as OutboundUpgrade<TSubstream>>::Error
>);
let mut proto1 = upgrade::toggleable(proto1);
proto1.disable();
let proto2 = upgrade::toggleable(proto2);
proto1.or_outbound(proto2)
};
return Ok(Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest { return Ok(Async::Ready(Some(ProtocolsHandlerEvent::OutboundSubstreamRequest {
upgrade, upgrade: EitherUpgrade::B(upgrade),
info: EitherOutput::Second(info), info: EitherOutput::Second(info),
}))); })));
}, },

109
core/src/upgrade/either.rs Normal file
View File

@ -0,0 +1,109 @@
// Copyright 2018 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use bytes::Bytes;
use futures::future::Either;
use crate::{
either::{EitherOutput, EitherError, EitherFuture2},
upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}
};
/// A type to represent two possible upgrade types (inbound or outbound).
#[derive(Debug, Clone)]
pub enum EitherUpgrade<A, B> { A(A), B(B) }
impl<A, B> UpgradeInfo for EitherUpgrade<A, B>
where
A: UpgradeInfo,
B: UpgradeInfo
{
type UpgradeId = Either<A::UpgradeId, B::UpgradeId>;
type NamesIter = EitherIter<A::NamesIter, B::NamesIter>;
fn protocol_names(&self) -> Self::NamesIter {
match self {
EitherUpgrade::A(a) => EitherIter::A(a.protocol_names()),
EitherUpgrade::B(b) => EitherIter::B(b.protocol_names())
}
}
}
impl<C, A, B, TA, TB, EA, EB> InboundUpgrade<C> for EitherUpgrade<A, B>
where
A: InboundUpgrade<C, Output = TA, Error = EA>,
B: InboundUpgrade<C, Output = TB, Error = EB>,
{
type Output = EitherOutput<TA, TB>;
type Error = EitherError<EA, EB>;
type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_inbound(self, sock: C, id: Self::UpgradeId) -> Self::Future {
match (self, id) {
(EitherUpgrade::A(a), Either::A(id)) => EitherFuture2::A(a.upgrade_inbound(sock, id)),
(EitherUpgrade::B(b), Either::B(id)) => EitherFuture2::B(b.upgrade_inbound(sock, id)),
_ => panic!("Invalid invocation of EitherUpgrade::upgrade_inbound")
}
}
}
impl<C, A, B, TA, TB, EA, EB> OutboundUpgrade<C> for EitherUpgrade<A, B>
where
A: OutboundUpgrade<C, Output = TA, Error = EA>,
B: OutboundUpgrade<C, Output = TB, Error = EB>,
{
type Output = EitherOutput<TA, TB>;
type Error = EitherError<EA, EB>;
type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_outbound(self, sock: C, id: Self::UpgradeId) -> Self::Future {
match (self, id) {
(EitherUpgrade::A(a), Either::A(id)) => EitherFuture2::A(a.upgrade_outbound(sock, id)),
(EitherUpgrade::B(b), Either::B(id)) => EitherFuture2::B(b.upgrade_outbound(sock, id)),
_ => panic!("Invalid invocation of EitherUpgrade::upgrade_outbound")
}
}
}
/// A type to represent two possible `Iterator` types.
#[derive(Debug, Clone)]
pub enum EitherIter<A, B> { A(A), B(B) }
impl<A, B, AId, BId> Iterator for EitherIter<A, B>
where
A: Iterator<Item = (Bytes, AId)>,
B: Iterator<Item = (Bytes, BId)>,
{
type Item = (Bytes, Either<AId, BId>);
fn next(&mut self) -> Option<Self::Item> {
match self {
EitherIter::A(a) => a.next().map(|(name, id)| (name, Either::A(id))),
EitherIter::B(b) => b.next().map(|(name, id)| (name, Either::B(id)))
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
match self {
EitherIter::A(a) => a.size_hint(),
EitherIter::B(b) => b.size_hint()
}
}
}

8
core/src/upgrade/mod.rs
View File

@ -59,10 +59,10 @@
mod apply; mod apply;
mod denied; mod denied;
mod either;
mod error; mod error;
mod map; mod map;
mod or; mod or;
mod toggleable;
use bytes::Bytes; use bytes::Bytes;
use futures::future::Future; use futures::future::Future;
@ -70,10 +70,10 @@ 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,
map::{MapInboundUpgrade, MapOutboundUpgrade, MapInboundUpgradeErr, MapOutboundUpgradeErr}, map::{MapInboundUpgrade, MapOutboundUpgrade, MapInboundUpgradeErr, MapOutboundUpgradeErr},
or::OrUpgrade, or::OrUpgrade,
toggleable::{toggleable, Toggleable}
}; };
/// Common trait for upgrades that can be applied on inbound substreams, outbound substreams, /// Common trait for upgrades that can be applied on inbound substreams, outbound substreams,
@ -132,7 +132,7 @@ pub trait InboundUpgradeExt<C>: InboundUpgrade<C> {
fn or_inbound<U>(self, upgrade: U) -> OrUpgrade<Self, U> fn or_inbound<U>(self, upgrade: U) -> OrUpgrade<Self, U>
where where
Self: Sized, Self: Sized,
U: InboundUpgrade<C, Output = Self::Output, Error = Self::Error> U: InboundUpgrade<C>
{ {
OrUpgrade::new(self, upgrade) OrUpgrade::new(self, upgrade)
} }
@ -182,7 +182,7 @@ pub trait OutboundUpgradeExt<C>: OutboundUpgrade<C> {
fn or_outbound<U>(self, upgrade: U) -> OrUpgrade<Self, U> fn or_outbound<U>(self, upgrade: U) -> OrUpgrade<Self, U>
where where
Self: Sized, Self: Sized,
U: OutboundUpgrade<C, Output = Self::Output, Error = Self::Error> U: OutboundUpgrade<C>
{ {
OrUpgrade::new(self, upgrade) OrUpgrade::new(self, upgrade)
} }

37
core/src/upgrade/or.rs
View File

@ -20,7 +20,10 @@
use bytes::Bytes; use bytes::Bytes;
use futures::future::Either; use futures::future::Either;
use crate::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}; use crate::{
either::{EitherOutput, EitherError, EitherFuture2},
upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}
};
/// Upgrade that combines two upgrades into one. Supports all the protocols supported by either /// Upgrade that combines two upgrades into one. Supports all the protocols supported by either
/// sub-upgrade. /// sub-upgrade.
@ -51,36 +54,36 @@ where
} }
} }
impl<C, A, B, T, E> InboundUpgrade<C> for OrUpgrade<A, B> impl<C, A, B, TA, TB, EA, EB> InboundUpgrade<C> for OrUpgrade<A, B>
where where
A: InboundUpgrade<C, Output = T, Error = E>, A: InboundUpgrade<C, Output = TA, Error = EA>,
B: InboundUpgrade<C, Output = T, Error = E>, B: InboundUpgrade<C, Output = TB, Error = EB>,
{ {
type Output = T; // TODO: different output types type Output = EitherOutput<TA, TB>;
type Error = E; // TODO: different error types type Error = EitherError<EA, EB>;
type Future = Either<A::Future, B::Future>; type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_inbound(self, sock: C, id: Self::UpgradeId) -> Self::Future { fn upgrade_inbound(self, sock: C, id: Self::UpgradeId) -> Self::Future {
match id { match id {
Either::A(id) => Either::A(self.0.upgrade_inbound(sock, id)), Either::A(id) => EitherFuture2::A(self.0.upgrade_inbound(sock, id)),
Either::B(id) => Either::B(self.1.upgrade_inbound(sock, id)) Either::B(id) => EitherFuture2::B(self.1.upgrade_inbound(sock, id))
} }
} }
} }
impl<C, A, B, T, E> OutboundUpgrade<C> for OrUpgrade<A, B> impl<C, A, B, TA, TB, EA, EB> OutboundUpgrade<C> for OrUpgrade<A, B>
where where
A: OutboundUpgrade<C, Output = T, Error = E>, A: OutboundUpgrade<C, Output = TA, Error = EA>,
B: OutboundUpgrade<C, Output = T, Error = E>, B: OutboundUpgrade<C, Output = TB, Error = EB>,
{ {
type Output = T; // TODO: different output types type Output = EitherOutput<TA, TB>;
type Error = E; // TODO: different error types type Error = EitherError<EA, EB>;
type Future = Either<A::Future, B::Future>; type Future = EitherFuture2<A::Future, B::Future>;
fn upgrade_outbound(self, sock: C, id: Self::UpgradeId) -> Self::Future { fn upgrade_outbound(self, sock: C, id: Self::UpgradeId) -> Self::Future {
match id { match id {
Either::A(id) => Either::A(self.0.upgrade_outbound(sock, id)), Either::A(id) => EitherFuture2::A(self.0.upgrade_outbound(sock, id)),
Either::B(id) => Either::B(self.1.upgrade_outbound(sock, id)) Either::B(id) => EitherFuture2::B(self.1.upgrade_outbound(sock, id))
} }
} }
} }

153
core/src/upgrade/toggleable.rs
View File

@ -1,153 +0,0 @@
// Copyright 2018 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
use futures::future;
/// Wraps around a `InboundUpgrade` or `OutboundUpgrade` and makes it possible
/// to enable or disable the upgrade.
#[inline]
pub fn toggleable<U>(upgrade: U) -> Toggleable<U> {
Toggleable {
inner: upgrade,
enabled: true,
}
}
/// See `toggleable`.
#[derive(Debug, Copy, Clone)]
pub struct Toggleable<U> {
inner: U,
enabled: bool,
}
impl<U> Toggleable<U> {
/// Toggles the upgrade.
#[inline]
pub fn toggle(&mut self) {
self.enabled = !self.enabled;
}
/// Returns true if the upgrade is enabled.
#[inline]
pub fn enabled(&self) -> bool {
self.enabled
}
/// Enables the upgrade.
#[inline]
pub fn enable(&mut self) {
self.enabled = true;
}
/// Disables the upgrade.
#[inline]
pub fn disable(&mut self) {
self.enabled = false;
}
}
impl<U> UpgradeInfo for Toggleable<U>
where
U: UpgradeInfo
{
type UpgradeId = U::UpgradeId;
type NamesIter = ToggleableIter<U::NamesIter>;
#[inline]
fn protocol_names(&self) -> Self::NamesIter {
ToggleableIter {
inner: self.inner.protocol_names(),
enabled: self.enabled,
}
}
}
impl<C, U> InboundUpgrade<C> for Toggleable<U>
where
U: InboundUpgrade<C>
{
type Output = U::Output;
type Error = U::Error;
type Future = future::Either<future::Empty<Self::Output, Self::Error>, U::Future>;
#[inline]
fn upgrade_inbound(self, socket: C, id: Self::UpgradeId) -> Self::Future {
if self.enabled {
future::Either::B(self.inner.upgrade_inbound(socket, id))
} else {
future::Either::A(future::empty())
}
}
}
impl<C, U> OutboundUpgrade<C> for Toggleable<U>
where
U: OutboundUpgrade<C>
{
type Output = U::Output;
type Error = U::Error;
type Future = future::Either<future::Empty<Self::Output, Self::Error>, U::Future>;
#[inline]
fn upgrade_outbound(self, socket: C, id: Self::UpgradeId) -> Self::Future {
if self.enabled {
future::Either::B(self.inner.upgrade_outbound(socket, id))
} else {
future::Either::A(future::empty())
}
}
}
/// Iterator that is toggleable.
#[derive(Debug, Clone)]
pub struct ToggleableIter<I> {
inner: I,
// It is expected that `enabled` doesn't change once the iterator has been created.
enabled: bool,
}
impl<I> Iterator for ToggleableIter<I>
where I: Iterator
{
type Item = I::Item;
fn next(&mut self) -> Option<Self::Item> {
if self.enabled {
self.inner.next()
} else {
None
}
}
fn size_hint(&self) -> (usize, Option<usize>) {
if self.enabled {
self.inner.size_hint()
} else {
(0, Some(0))
}
}
}
impl<I> ExactSizeIterator for ToggleableIter<I>
where
I: ExactSizeIterator
{}

5
protocols/floodsub/src/handler.rs
View File

@ -114,11 +114,6 @@ where
self.config.clone() self.config.clone()
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
self.config.clone()
}
fn inject_fully_negotiated_inbound( fn inject_fully_negotiated_inbound(
&mut self, &mut self,
protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output

5
protocols/identify/src/listen_handler.rs
View File

@ -69,11 +69,6 @@ where
self.config.clone() self.config.clone()
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
DeniedUpgrade
}
fn inject_fully_negotiated_inbound( fn inject_fully_negotiated_inbound(
&mut self, &mut self,
protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output

16
protocols/identify/src/periodic_id_handler.rs
View File

@ -22,7 +22,7 @@ use crate::{RemoteInfo, IdentifyProtocolConfig};
use futures::prelude::*; use futures::prelude::*;
use libp2p_core::{ use libp2p_core::{
protocols_handler::{ProtocolsHandler, ProtocolsHandlerEvent}, protocols_handler::{ProtocolsHandler, ProtocolsHandlerEvent},
upgrade::{self, DeniedUpgrade, OutboundUpgrade, Toggleable} upgrade::{DeniedUpgrade, OutboundUpgrade}
}; };
use std::{io, marker::PhantomData, time::{Duration, Instant}}; use std::{io, marker::PhantomData, time::{Duration, Instant}};
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
@ -39,7 +39,7 @@ const TRY_AGAIN_ON_ERR: Duration = Duration::from_secs(60 * 60);
/// Protocol handler that identifies the remote at a regular period. /// Protocol handler that identifies the remote at a regular period.
pub struct PeriodicIdentification<TSubstream> { pub struct PeriodicIdentification<TSubstream> {
/// Configuration for the protocol. /// Configuration for the protocol.
config: Toggleable<IdentifyProtocolConfig>, config: IdentifyProtocolConfig,
/// If `Some`, we successfully generated an `PeriodicIdentificationEvent` and we will produce /// If `Some`, we successfully generated an `PeriodicIdentificationEvent` and we will produce
/// it the next time `poll()` is invoked. /// it the next time `poll()` is invoked.
@ -67,7 +67,7 @@ impl<TSubstream> PeriodicIdentification<TSubstream> {
#[inline] #[inline]
pub fn new() -> Self { pub fn new() -> Self {
PeriodicIdentification { PeriodicIdentification {
config: upgrade::toggleable(IdentifyProtocolConfig), config: IdentifyProtocolConfig,
pending_result: None, pending_result: None,
next_id: Some(Delay::new(Instant::now() + DELAY_TO_FIRST_ID)), next_id: Some(Delay::new(Instant::now() + DELAY_TO_FIRST_ID)),
marker: PhantomData, marker: PhantomData,
@ -83,7 +83,7 @@ where
type OutEvent = PeriodicIdentificationEvent; type OutEvent = PeriodicIdentificationEvent;
type Substream = TSubstream; type Substream = TSubstream;
type InboundProtocol = DeniedUpgrade; type InboundProtocol = DeniedUpgrade;
type OutboundProtocol = Toggleable<IdentifyProtocolConfig>; type OutboundProtocol = IdentifyProtocolConfig;
type OutboundOpenInfo = (); type OutboundOpenInfo = ();
#[inline] #[inline]
@ -91,11 +91,6 @@ where
DeniedUpgrade DeniedUpgrade
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
self.config.clone()
}
fn inject_fully_negotiated_inbound(&mut self, protocol: Void) { fn inject_fully_negotiated_inbound(&mut self, protocol: Void) {
unreachable(protocol) unreachable(protocol)
} }
@ -155,8 +150,7 @@ where
Ok(Async::NotReady) => Ok(Async::NotReady), Ok(Async::NotReady) => Ok(Async::NotReady),
Ok(Async::Ready(())) => { Ok(Async::Ready(())) => {
next_id.reset(Instant::now() + DELAY_TO_NEXT_ID); next_id.reset(Instant::now() + DELAY_TO_NEXT_ID);
let mut upgrade = self.config.clone(); let upgrade = self.config.clone();
upgrade.enable();
let ev = ProtocolsHandlerEvent::OutboundSubstreamRequest { upgrade, info: () }; let ev = ProtocolsHandlerEvent::OutboundSubstreamRequest { upgrade, info: () };
Ok(Async::Ready(Some(ev))) Ok(Async::Ready(Some(ev)))
} }

13
protocols/ping/src/dial_handler.rs
View File

@ -23,7 +23,7 @@ use libp2p_core::{
OutboundUpgrade, OutboundUpgrade,
ProtocolsHandler, ProtocolsHandler,
ProtocolsHandlerEvent, ProtocolsHandlerEvent,
upgrade::{self, DeniedUpgrade} upgrade::DeniedUpgrade
}; };
use log::warn; use log::warn;
use protocol::{Ping, PingDialer}; use protocol::{Ping, PingDialer};
@ -40,7 +40,7 @@ use void::{Void, unreachable};
/// If the remote doesn't respond, produces `Unresponsive` and closes the connection. /// If the remote doesn't respond, produces `Unresponsive` and closes the connection.
pub struct PeriodicPingHandler<TSubstream> { pub struct PeriodicPingHandler<TSubstream> {
/// Configuration for the ping protocol. /// Configuration for the ping protocol.
ping_config: upgrade::Toggleable<Ping<Instant>>, ping_config: Ping<Instant>,
/// State of the outgoing ping. /// State of the outgoing ping.
out_state: OutState<TSubstream>, out_state: OutState<TSubstream>,
@ -128,7 +128,7 @@ impl<TSubstream> PeriodicPingHandler<TSubstream> {
let ping_timeout = Duration::from_secs(30); let ping_timeout = Duration::from_secs(30);
PeriodicPingHandler { PeriodicPingHandler {
ping_config: upgrade::toggleable(Default::default()), ping_config: Default::default(),
out_state: OutState::NeedToOpen { out_state: OutState::NeedToOpen {
expires: Delay::new(Instant::now() + ping_timeout), expires: Delay::new(Instant::now() + ping_timeout),
}, },
@ -154,7 +154,7 @@ where
type OutEvent = OutEvent; type OutEvent = OutEvent;
type Substream = TSubstream; type Substream = TSubstream;
type InboundProtocol = DeniedUpgrade; type InboundProtocol = DeniedUpgrade;
type OutboundProtocol = upgrade::Toggleable<Ping<Instant>>; type OutboundProtocol = Ping<Instant>;
type OutboundOpenInfo = (); type OutboundOpenInfo = ();
#[inline] #[inline]
@ -162,11 +162,6 @@ where
DeniedUpgrade DeniedUpgrade
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
self.ping_config
}
fn inject_fully_negotiated_inbound(&mut self, protocol: Void) { fn inject_fully_negotiated_inbound(&mut self, protocol: Void) {
unreachable(protocol) unreachable(protocol)
} }

5
protocols/ping/src/listen_handler.rs
View File

@ -80,11 +80,6 @@ where
self.ping_config self.ping_config
} }
#[inline]
fn dialer_protocol(&self) -> Self::OutboundProtocol {
DeniedUpgrade
}
fn inject_fully_negotiated_inbound( fn inject_fully_negotiated_inbound(
&mut self, &mut self,
protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output