fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-04-24 18:52:14 +00:00
Code Issues Projects Releases Wiki Activity

Add ProtocolsHandler::InboundOpenInfo. (#1714)

Browse Source 
* Add `ProtocolsHandler::InboundOpenInfo`.

* Update swarm/src/protocols_handler/multi.rs

Co-authored-by: Roman Borschel <romanb@users.noreply.github.com>

* Update CHANGELOGs and versions.

Co-authored-by: Roman Borschel <romanb@users.noreply.github.com>
This commit is contained in:
Toralf Wittner 2020-08-23 16:57:20 +02:00 committed by GitHub
parent 740a2e0235
commit 0b05b21093
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
32 changed files with 337 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGELOG.md
View File

@ -23,6 +23,13 @@
- [`parity-multiaddr` CHANGELOG](misc/multiaddr/CHANGELOG.md)
- [`libp2p-core-derive` CHANGELOG](misc/core-derive/CHANGELOG.md)
# Version 0.25.0 [unreleased]
- The `ProtocolsHandler` in `libp2p-swarm` has a new associated type
`InboundOpenInfo` ([PR 1714]).
[PR 1714]: https://github.com/libp2p/rust-libp2p/pull/1714
# Version 0.24.0 [2020-08-18]
- Update `libp2p-core`, `libp2p-gossipsub`, `libp2p-kad`, `libp2p-mdns`,

18
Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p"
edition = "2018"
description = "Peer-to-peer networking library"
version = "0.24.0"
version = "0.25.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -65,18 +65,18 @@ futures = "0.3.1"
lazy_static = "1.2"
libp2p-core = { version = "0.21.0", path = "core" }
libp2p-core-derive = { version = "0.20.2", path = "misc/core-derive" }
libp2p-floodsub = { version = "0.21.0", path = "protocols/floodsub", optional = true }
libp2p-gossipsub = { version = "0.21.0", path = "./protocols/gossipsub", optional = true }
libp2p-identify = { version = "0.21.0", path = "protocols/identify", optional = true }
libp2p-kad = { version = "0.22.0", path = "protocols/kad", optional = true }
libp2p-floodsub = { version = "0.22.0", path = "protocols/floodsub", optional = true }
libp2p-gossipsub = { version = "0.22.0", path = "./protocols/gossipsub", optional = true }
libp2p-identify = { version = "0.22.0", path = "protocols/identify", optional = true }
libp2p-kad = { version = "0.23.0", path = "protocols/kad", optional = true }
libp2p-mplex = { version = "0.21.0", path = "muxers/mplex", optional = true }
libp2p-noise = { version = "0.23.0", path = "protocols/noise", optional = true }
libp2p-ping = { version = "0.21.0", path = "protocols/ping", optional = true }
libp2p-ping = { version = "0.22.0", path = "protocols/ping", optional = true }
libp2p-plaintext = { version = "0.21.0", path = "protocols/plaintext", optional = true }
libp2p-pnet = { version = "0.19.1", path = "protocols/pnet", optional = true }
libp2p-request-response = { version = "0.2.0", path = "protocols/request-response", optional = true }
libp2p-request-response = { version = "0.3.0", path = "protocols/request-response", optional = true }
libp2p-secio = { version = "0.21.0", path = "protocols/secio", default-features = false, optional = true }
libp2p-swarm = { version = "0.21.0", path = "swarm" }
libp2p-swarm = { version = "0.22.0", path = "swarm" }
libp2p-uds = { version = "0.21.0", path = "transports/uds", optional = true }
libp2p-wasm-ext = { version = "0.21.0", path = "transports/wasm-ext", optional = true }
libp2p-yamux = { version = "0.21.0", path = "muxers/yamux", optional = true }
@ -90,7 +90,7 @@ wasm-timer = "0.2.4"
[target.'cfg(not(any(target_os = "emscripten", target_os = "wasi", target_os = "unknown")))'.dependencies]
libp2p-deflate = { version = "0.21.0", path = "protocols/deflate", optional = true }
libp2p-dns = { version = "0.21.0", path = "transports/dns", optional = true }
libp2p-mdns = { version = "0.21.0", path = "protocols/mdns", optional = true }
libp2p-mdns = { version = "0.22.0", path = "protocols/mdns", optional = true }
libp2p-tcp = { version = "0.21.0", path = "transports/tcp", optional = true }
libp2p-websocket = { version = "0.22.0", path = "transports/websocket", optional = true }

4
protocols/floodsub/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.22.0 [unreleased]
- Update `libp2p-swarm`.
# 0.21.0 [2020-08-18]
- Bump `libp2p-core` and `libp2p-swarm` dependency.

4
protocols/floodsub/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-floodsub"
edition = "2018"
description = "Floodsub protocol for libp2p"
version = "0.21.0"
version = "0.22.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -14,7 +14,7 @@ cuckoofilter = "0.3.2"
fnv = "1.0"
futures = "0.3.1"
libp2p-core = { version = "0.21.0", path = "../../core" }
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
prost = "0.6.1"
rand = "0.7"
smallvec = "1.0"

4
protocols/gossipsub/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.22.0 [unreleased]
- Update `libp2p-swarm`.
# 0.21.0 [2020-08-18]
- Add public API to list topics and peers. [PR 1677](https://github.com/libp2p/rust-libp2p/pull/1677).

4
protocols/gossipsub/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-gossipsub"
edition = "2018"
description = "Gossipsub protocol for libp2p"
version = "0.21.0"
version = "0.22.0"
authors = ["Age Manning <Age@AgeManning.com>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -10,7 +10,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
categories = ["network-programming", "asynchronous"]
[dependencies]
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
libp2p-core = { version = "0.21.0", path = "../../core" }
bytes = "0.5.4"
byteorder = "1.3.2"

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

@ -40,7 +40,7 @@ use std::{
/// Protocol Handler that manages a single long-lived substream with a peer.
pub struct GossipsubHandler {
/// Upgrade configuration for the gossipsub protocol.
listen_protocol: SubstreamProtocol<ProtocolConfig>,
listen_protocol: SubstreamProtocol<ProtocolConfig, ()>,
/// The single long-lived outbound substream.
outbound_substream: Option<OutboundSubstreamState>,
@ -95,7 +95,7 @@ impl GossipsubHandler {
protocol_id,
max_transmit_size,
validation_mode,
)),
), ()),
inbound_substream: None,
outbound_substream: None,
outbound_substream_establishing: false,
@ -112,14 +112,16 @@ impl ProtocolsHandler for GossipsubHandler {
type InboundProtocol = ProtocolConfig;
type OutboundProtocol = ProtocolConfig;
type OutboundOpenInfo = GossipsubRpc;
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
self.listen_protocol.clone()
}
fn inject_fully_negotiated_inbound(
&mut self,
substream: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
_info: Self::InboundOpenInfo
) {
// new inbound substream. Replace the current one, if it exists.
trace!("New inbound substream request");
@ -184,8 +186,7 @@ impl ProtocolsHandler for GossipsubHandler {
self.send_queue.shrink_to_fit();
self.outbound_substream_establishing = true;
return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: self.listen_protocol.clone(),
info: message,
protocol: self.listen_protocol.clone().map_info(|()| message)
});
}

4
protocols/identify/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.22.0 [unreleased]
- Update `libp2p-swarm`.
# 0.21.0 [2020-08-18]
- Bump `libp2p-core` and `libp2p-swarm` dependencies.

4
protocols/identify/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-identify"
edition = "2018"
description = "Nodes identifcation protocol for libp2p"
version = "0.21.0"
version = "0.22.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -12,7 +12,7 @@ categories = ["network-programming", "asynchronous"]
[dependencies]
futures = "0.3.1"
libp2p-core = { version = "0.21.0", path = "../../core" }
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
log = "0.4.1"
prost = "0.6.1"
smallvec = "1.0"

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

@ -93,14 +93,16 @@ impl ProtocolsHandler for IdentifyHandler {
type InboundProtocol = IdentifyProtocolConfig;
type OutboundProtocol = IdentifyProtocolConfig;
type OutboundOpenInfo = ();
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
SubstreamProtocol::new(self.config.clone())
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
SubstreamProtocol::new(self.config.clone(), ())
}
fn inject_fully_negotiated_inbound(
&mut self,
protocol: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output
protocol: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
_info: Self::InboundOpenInfo
) {
self.events.push(IdentifyHandlerEvent::Identify(protocol))
}
@ -152,8 +154,7 @@ impl ProtocolsHandler for IdentifyHandler {
Poll::Ready(Ok(())) => {
self.next_id.reset(DELAY_TO_NEXT_ID);
let ev = ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(self.config.clone()),
info: (),
protocol: SubstreamProtocol::new(self.config.clone(), ())
};
Poll::Ready(ev)
}

4
protocols/kad/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.23.0 [unreleased]
- Update `libp2p-swarm`.
# 0.22.1 [2020-08-19]
- Explicitly convert from u8 to usize in `BucketIndex::range` to prevent type

4
protocols/kad/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-kad"
edition = "2018"
description = "Kademlia protocol for libp2p"
version = "0.22.1"
version = "0.23.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -18,7 +18,7 @@ futures_codec = "0.4"
futures = "0.3.1"
log = "0.4"
libp2p-core = { version = "0.21.0", path = "../../core" }
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
multihash = "0.11.0"
prost = "0.6.1"
rand = "0.7.2"

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

@ -261,7 +261,6 @@ impl error::Error for KademliaHandlerQueryErr {
}
impl From<ProtocolsHandlerUpgrErr<io::Error>> for KademliaHandlerQueryErr {
#[inline]
fn from(err: ProtocolsHandlerUpgrErr<io::Error>) -> Self {
KademliaHandlerQueryErr::Upgrade(err)
}
@ -409,13 +408,13 @@ where
type OutboundProtocol = KademliaProtocolConfig;
// Message of the request to send to the remote, and user data if we expect an answer.
type OutboundOpenInfo = (KadRequestMsg, Option<TUserData>);
type InboundOpenInfo = ();
#[inline]
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
if self.config.allow_listening {
SubstreamProtocol::new(self.config.protocol_config.clone()).map_upgrade(upgrade::EitherUpgrade::A)
SubstreamProtocol::new(self.config.protocol_config.clone(), ()).map_upgrade(upgrade::EitherUpgrade::A)
} else {
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade))
SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
}
}
@ -431,6 +430,7 @@ where
fn inject_fully_negotiated_inbound(
&mut self,
protocol: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
(): Self::InboundOpenInfo
) {
// If `self.allow_listening` is false, then we produced a `DeniedUpgrade` and `protocol`
// is a `Void`.
@ -591,7 +591,6 @@ where
}
}
#[inline]
fn inject_dial_upgrade_error(
&mut self,
(_, user_data): Self::OutboundOpenInfo,
@ -605,7 +604,6 @@ where
}
}
#[inline]
fn connection_keep_alive(&self) -> KeepAlive {
self.keep_alive
}
@ -696,8 +694,7 @@ fn advance_substream<TUserData>(
match state {
SubstreamState::OutPendingOpen(msg, user_data) => {
let ev = ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(upgrade),
info: (msg, user_data),
protocol: SubstreamProtocol::new(upgrade, (msg, user_data))
};
(None, Some(ev), false)
}

4
protocols/mdns/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.22.0 [unreleased]
- Update `libp2p-swarm`.
# 0.21.0 [2020-08-18]
- Bump `libp2p-core` and `libp2p-swarm` dependencies.

4
protocols/mdns/Cargo.toml
View File

@ -1,7 +1,7 @@
[package]
name = "libp2p-mdns"
edition = "2018"
version = "0.21.0"
version = "0.22.0"
description = "Implementation of the libp2p mDNS discovery method"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
@ -17,7 +17,7 @@ either = "1.5.3"
futures = "0.3.1"
lazy_static = "1.2"
libp2p-core = { version = "0.21.0", path = "../../core" }
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
log = "0.4"
net2 = "0.2"
rand = "0.7"

4
protocols/ping/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.22.0 [unreleased]
- Update `libp2p-swarm`.
# 0.21.0 [2020-08-18]
- Refactor the ping protocol for conformity by (re)using

4
protocols/ping/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-ping"
edition = "2018"
description = "Ping protocol for libp2p"
version = "0.21.0"
version = "0.22.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -12,7 +12,7 @@ categories = ["network-programming", "asynchronous"]
[dependencies]
futures = "0.3.1"
libp2p-core = { version = "0.21.0", path = "../../core" }
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
log = "0.4.1"
rand = "0.7.2"
void = "1.0"

12
protocols/ping/src/handler.rs
View File

@ -207,12 +207,13 @@ impl ProtocolsHandler for PingHandler {
type InboundProtocol = protocol::Ping;
type OutboundProtocol = protocol::Ping;
type OutboundOpenInfo = ();
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<protocol::Ping> {
SubstreamProtocol::new(protocol::Ping)
fn listen_protocol(&self) -> SubstreamProtocol<protocol::Ping, ()> {
SubstreamProtocol::new(protocol::Ping, ())
}
fn inject_fully_negotiated_inbound(&mut self, stream: NegotiatedSubstream) {
fn inject_fully_negotiated_inbound(&mut self, stream: NegotiatedSubstream, (): ()) {
self.inbound = Some(protocol::recv_ping(stream).boxed());
}
@ -329,11 +330,10 @@ impl ProtocolsHandler for PingHandler {
}
None => {
self.outbound = Some(PingState::OpenStream);
let protocol = SubstreamProtocol::new(protocol::Ping)
let protocol = SubstreamProtocol::new(protocol::Ping, ())
.with_timeout(self.config.timeout);
return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol,
info: (),
protocol
})
}
}

4
protocols/request-response/CHANGELOG.md
View File

@ -1,3 +1,7 @@
# 0.3.0 [unreleased]
- Update `libp2p-swarm`.
# 0.2.0 [2020-08-18]
- Fixed connection keep-alive, permitting connections to close due

4
protocols/request-response/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-request-response"
edition = "2018"
description = "Generic Request/Response Protocols"
version = "0.2.0"
version = "0.3.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -13,7 +13,7 @@ categories = ["network-programming", "asynchronous"]
async-trait = "0.1"
futures = "0.3.1"
libp2p-core = { version = "0.21.0", path = "../../core" }
libp2p-swarm = { version = "0.21.0", path = "../../swarm" }
libp2p-swarm = { version = "0.22.0", path = "../../swarm" }
log = "0.4.11"
lru = "0.6"
rand = "0.7"

12
protocols/request-response/src/handler.rs
View File

@ -145,8 +145,9 @@ where
type InboundProtocol = ResponseProtocol<TCodec>;
type OutboundProtocol = RequestProtocol<TCodec>;
type OutboundOpenInfo = RequestId;
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
// A channel for notifying the handler when the inbound
// upgrade received the request.
let (rq_send, rq_recv) = oneshot::channel();
@ -173,12 +174,13 @@ where
// `ResponseChannel`.
self.inbound.push(rq_recv.map_ok(move |rq| (rq, rs_send)).boxed());
SubstreamProtocol::new(proto).with_timeout(self.substream_timeout)
SubstreamProtocol::new(proto, ()).with_timeout(self.substream_timeout)
}
fn inject_fully_negotiated_inbound(
&mut self,
(): (),
(): ()
) {
// Nothing to do, as the response has already been sent
// as part of the upgrade.
@ -229,6 +231,7 @@ where
fn inject_listen_upgrade_error(
&mut self,
(): Self::InboundOpenInfo,
error: ProtocolsHandlerUpgrErr<io::Error>
) {
match error {
@ -300,9 +303,8 @@ where
let info = request.request_id;
return Poll::Ready(
ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(request)
.with_timeout(self.substream_timeout),
info,
protocol: SubstreamProtocol::new(request, info)
.with_timeout(self.substream_timeout)
},
)
}

11
swarm/CHANGELOG.md
View File

@ -1,3 +1,14 @@
# 0.22.0 [unreleased]
- Adds `ProtocolsHandler::InboundOpenInfo` type which mirrors the existing
`OutboundOpenInfo` type. A value of this type is passed as an extra argument
to `ProtocolsHandler::inject_fully_negotiated_inbound` and
`ProtocolsHandler::inject_listen_upgrade_error`.
- `SubstreamProtocol` now has a second type parameter corresponding to
inbound or outbound information, a value of which is part of `SubstreamProtocol`
now. Consequently `ProtocolsHandlerEvent::OutboundSubstreamRequest` no longer
has a separate `info` field.
# 0.21.0 [2020-08-18]
- Add missing delegation calls in some `ProtocolsHandler` wrappers.

3
swarm/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "libp2p-swarm"
edition = "2018"
description = "The libp2p swarm"
version = "0.21.0"
version = "0.22.0"
authors = ["Parity Technologies <admin@parity.io>"]
license = "MIT"
repository = "https://github.com/libp2p/rust-libp2p"
@ -10,6 +10,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
categories = ["network-programming", "asynchronous"]
[dependencies]
either = "1.6.0"
futures = "0.3.1"
libp2p-core = { version = "0.21.0", path = "../core" }
log = "0.4"

69
swarm/src/protocols_handler.rs
View File

@ -110,6 +110,8 @@ pub trait ProtocolsHandler: Send + 'static {
type InboundProtocol: InboundUpgradeSend + Send + 'static;
/// The outbound upgrade for the protocol(s) used by the handler.
type OutboundProtocol: OutboundUpgradeSend;
/// The type of additional information returned from `listen_protocol`.
type InboundOpenInfo: Send + 'static;
/// The type of additional information passed to an `OutboundSubstreamRequest`.
type OutboundOpenInfo: Send + 'static;
@ -120,12 +122,13 @@ pub trait ProtocolsHandler: Send + 'static {
/// > supported protocols, even if in a specific context a particular one is
/// > not supported, (eg. when only allowing one substream at a time for a protocol).
/// > This allows a remote to put the list of supported protocols in a cache.
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol>;
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo>;
/// Injects the output of a successful upgrade on a new inbound substream.
fn inject_fully_negotiated_inbound(
&mut self,
protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output
protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
info: Self::InboundOpenInfo
);
/// Injects the output of a successful upgrade on a new outbound substream.
@ -156,6 +159,7 @@ pub trait ProtocolsHandler: Send + 'static {
/// Indicates to the handler that upgrading an inbound substream to the given protocol has failed.
fn inject_listen_upgrade_error(
&mut self,
_: Self::InboundOpenInfo,
_: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>
) {}
@ -235,20 +239,22 @@ pub trait ProtocolsHandler: Send + 'static {
/// The inbound substream protocol(s) are defined by [`ProtocolsHandler::listen_protocol`]
/// and the outbound substream protocol(s) by [`ProtocolsHandlerEvent::OutboundSubstreamRequest`].
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct SubstreamProtocol<TUpgrade> {
pub struct SubstreamProtocol<TUpgrade, TInfo> {
upgrade: TUpgrade,
info: TInfo,
upgrade_protocol: upgrade::Version,
timeout: Duration,
}
impl<TUpgrade> SubstreamProtocol<TUpgrade> {
impl<TUpgrade, TInfo> SubstreamProtocol<TUpgrade, TInfo> {
/// Create a new `SubstreamProtocol` from the given upgrade.
///
/// The default timeout for applying the given upgrade on a substream is
/// 10 seconds.
pub fn new(upgrade: TUpgrade) -> SubstreamProtocol<TUpgrade> {
pub fn new(upgrade: TUpgrade, info: TInfo) -> Self {
SubstreamProtocol {
upgrade,
info,
upgrade_protocol: upgrade::Version::V1,
timeout: Duration::from_secs(10),
}
@ -262,12 +268,26 @@ impl<TUpgrade> SubstreamProtocol<TUpgrade> {
}
/// Maps a function over the protocol upgrade.
pub fn map_upgrade<U, F>(self, f: F) -> SubstreamProtocol<U>
pub fn map_upgrade<U, F>(self, f: F) -> SubstreamProtocol<U, TInfo>
where
F: FnOnce(TUpgrade) -> U,
{
SubstreamProtocol {
upgrade: f(self.upgrade),
info: self.info,
upgrade_protocol: self.upgrade_protocol,
timeout: self.timeout,
}
}
/// Maps a function over the protocol info.
pub fn map_info<U, F>(self, f: F) -> SubstreamProtocol<TUpgrade, U>
where
F: FnOnce(TInfo) -> U,
{
SubstreamProtocol {
upgrade: self.upgrade,
info: f(self.info),
upgrade_protocol: self.upgrade_protocol,
timeout: self.timeout,
}
@ -284,20 +304,19 @@ impl<TUpgrade> SubstreamProtocol<TUpgrade> {
&self.upgrade
}
/// Borrows the contained protocol info.
pub fn info(&self) -> &TInfo {
&self.info
}
/// Borrows the timeout for the protocol upgrade.
pub fn timeout(&self) -> &Duration {
&self.timeout
}
/// Converts the substream protocol configuration into the contained upgrade.
pub fn into_upgrade(self) -> (upgrade::Version, TUpgrade) {
(self.upgrade_protocol, self.upgrade)
}
}
impl<TUpgrade> From<TUpgrade> for SubstreamProtocol<TUpgrade> {
fn from(upgrade: TUpgrade) -> SubstreamProtocol<TUpgrade> {
SubstreamProtocol::new(upgrade)
pub fn into_upgrade(self) -> (upgrade::Version, TUpgrade, TInfo) {
(self.upgrade_protocol, self.upgrade, self.info)
}
}
@ -307,9 +326,7 @@ pub enum ProtocolsHandlerEvent<TConnectionUpgrade, TOutboundOpenInfo, TCustom, T
/// Request a new outbound substream to be opened with the remote.
OutboundSubstreamRequest {
/// The protocol(s) to apply on the substream.
protocol: SubstreamProtocol<TConnectionUpgrade>,
/// User-defined information, passed back when the substream is open.
info: TOutboundOpenInfo,
protocol: SubstreamProtocol<TConnectionUpgrade, TOutboundOpenInfo>
},
/// Close the connection for the given reason.
@ -333,10 +350,9 @@ impl<TConnectionUpgrade, TOutboundOpenInfo, TCustom, TErr>
F: FnOnce(TOutboundOpenInfo) -> I,
{
match self {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol,
info: map(info),
protocol: protocol.map_info(map)
}
}
ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(val),
@ -354,10 +370,9 @@ impl<TConnectionUpgrade, TOutboundOpenInfo, TCustom, TErr>
F: FnOnce(TConnectionUpgrade) -> I,
{
match self {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: protocol.map_upgrade(map),
info,
protocol: protocol.map_upgrade(map)
}
}
ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(val),
@ -374,8 +389,8 @@ impl<TConnectionUpgrade, TOutboundOpenInfo, TCustom, TErr>
F: FnOnce(TCustom) -> I,
{
match self {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }
}
ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(map(val)),
ProtocolsHandlerEvent::Close(val) => ProtocolsHandlerEvent::Close(val),
@ -391,8 +406,8 @@ impl<TConnectionUpgrade, TOutboundOpenInfo, TCustom, TErr>
F: FnOnce(TErr) -> I,
{
match self {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }
}
ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(val),
ProtocolsHandlerEvent::Close(val) => ProtocolsHandlerEvent::Close(map(val)),

10
swarm/src/protocols_handler/dummy.rs
View File

@ -51,14 +51,16 @@ impl ProtocolsHandler for DummyProtocolsHandler {
type InboundProtocol = DeniedUpgrade;
type OutboundProtocol = DeniedUpgrade;
type OutboundOpenInfo = Void;
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
SubstreamProtocol::new(DeniedUpgrade)
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
SubstreamProtocol::new(DeniedUpgrade, ())
}
fn inject_fully_negotiated_inbound(
&mut self,
_: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output
_: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
_: Self::InboundOpenInfo
) {
}
@ -75,7 +77,7 @@ impl ProtocolsHandler for DummyProtocolsHandler {
fn inject_dial_upgrade_error(&mut self, _: Self::OutboundOpenInfo, _: ProtocolsHandlerUpgrErr<<Self::OutboundProtocol as OutboundUpgrade<NegotiatedSubstream>>::Error>) {}
fn inject_listen_upgrade_error(&mut self, _: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Error>) {}
fn inject_listen_upgrade_error(&mut self, _: Self::InboundOpenInfo, _: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Error>) {}
fn connection_keep_alive(&self) -> KeepAlive {
self.keep_alive

15
swarm/src/protocols_handler/map_in.rs
View File

@ -59,17 +59,19 @@ where
type Error = TProtoHandler::Error;
type InboundProtocol = TProtoHandler::InboundProtocol;
type OutboundProtocol = TProtoHandler::OutboundProtocol;
type InboundOpenInfo = TProtoHandler::InboundOpenInfo;
type OutboundOpenInfo = TProtoHandler::OutboundOpenInfo;
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
self.inner.listen_protocol()
}
fn inject_fully_negotiated_inbound(
&mut self,
protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output
protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
info: Self::InboundOpenInfo
) {
self.inner.inject_fully_negotiated_inbound(protocol)
self.inner.inject_fully_negotiated_inbound(protocol, info)
}
fn inject_fully_negotiated_outbound(
@ -94,11 +96,8 @@ where
self.inner.inject_dial_upgrade_error(info, error)
}
fn inject_listen_upgrade_error(
&mut self,
error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>
) {
self.inner.inject_listen_upgrade_error(error)
fn inject_listen_upgrade_error(&mut self, info: Self::InboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>) {
self.inner.inject_listen_upgrade_error(info, error)
}
fn connection_keep_alive(&self) -> KeepAlive {

19
swarm/src/protocols_handler/map_out.rs
View File

@ -57,17 +57,19 @@ where
type Error = TProtoHandler::Error;
type InboundProtocol = TProtoHandler::InboundProtocol;
type OutboundProtocol = TProtoHandler::OutboundProtocol;
type InboundOpenInfo = TProtoHandler::InboundOpenInfo;
type OutboundOpenInfo = TProtoHandler::OutboundOpenInfo;
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
self.inner.listen_protocol()
}
fn inject_fully_negotiated_inbound(
&mut self,
protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output
protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
info: Self::InboundOpenInfo
) {
self.inner.inject_fully_negotiated_inbound(protocol)
self.inner.inject_fully_negotiated_inbound(protocol, info)
}
fn inject_fully_negotiated_outbound(
@ -90,11 +92,8 @@ where
self.inner.inject_dial_upgrade_error(info, error)
}
fn inject_listen_upgrade_error(
&mut self,
error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>
) {
self.inner.inject_listen_upgrade_error(error)
fn inject_listen_upgrade_error(&mut self, info: Self::InboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>) {
self.inner.inject_listen_upgrade_error(info, error)
}
fn connection_keep_alive(&self) -> KeepAlive {
@ -111,8 +110,8 @@ where
match ev {
ProtocolsHandlerEvent::Custom(ev) => ProtocolsHandlerEvent::Custom((self.map)(ev)),
ProtocolsHandlerEvent::Close(err) => ProtocolsHandlerEvent::Close(err),
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol } => {
ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }
}
}
})

106
swarm/src/protocols_handler/multi.rs
View File

@ -96,13 +96,21 @@ where
type Error = <H as ProtocolsHandler>::Error;
type InboundProtocol = Upgrade<K, <H as ProtocolsHandler>::InboundProtocol>;
type OutboundProtocol = <H as ProtocolsHandler>::OutboundProtocol;
type InboundOpenInfo = Info<K, <H as ProtocolsHandler>::InboundOpenInfo>;
type OutboundOpenInfo = (K, <H as ProtocolsHandler>::OutboundOpenInfo);
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
let upgrades = self.handlers.iter()
.map(|(k, h)| (k.clone(), h.listen_protocol().into_upgrade().1))
.collect();
SubstreamProtocol::new(Upgrade { upgrades })
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
let (upgrade, info) = self.handlers.iter()
.map(|(k, h)| {
let (_, u, i) = h.listen_protocol().into_upgrade();
(k.clone(), (u, i))
})
.fold((Upgrade::new(), Info::new()), |(mut upg, mut inf), (k, (u, i))| {
upg.upgrades.push((k.clone(), u));
inf.infos.push((k, i));
(upg, inf)
});
SubstreamProtocol::new(upgrade, info)
}
fn inject_fully_negotiated_outbound (
@ -119,10 +127,13 @@ where
fn inject_fully_negotiated_inbound (
&mut self,
(key, arg): <Self::InboundProtocol as InboundUpgradeSend>::Output
(key, arg): <Self::InboundProtocol as InboundUpgradeSend>::Output,
mut info: Self::InboundOpenInfo
) {
if let Some(h) = self.handlers.get_mut(&key) {
h.inject_fully_negotiated_inbound(arg)
if let Some(i) = info.take(&key) {
h.inject_fully_negotiated_inbound(arg, i)
}
} else {
log::error!("inject_fully_negotiated_inbound: no handler for key")
}
@ -156,47 +167,64 @@ where
fn inject_listen_upgrade_error(
&mut self,
mut info: Self::InboundOpenInfo,
error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>
) {
match error {
ProtocolsHandlerUpgrErr::Timer =>
for h in self.handlers.values_mut() {
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Timer)
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Timer)
}
}
ProtocolsHandlerUpgrErr::Timeout =>
for h in self.handlers.values_mut() {
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Timeout)
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Timeout)
}
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) =>
for h in self.handlers.values_mut() {
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)))
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)))
}
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::ProtocolError(e))) =>
match e {
ProtocolError::IoError(e) =>
for h in self.handlers.values_mut() {
let e = NegotiationError::ProtocolError(ProtocolError::IoError(e.kind().into()));
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
let e = NegotiationError::ProtocolError(ProtocolError::IoError(e.kind().into()));
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
}
}
ProtocolError::InvalidMessage =>
for h in self.handlers.values_mut() {
let e = NegotiationError::ProtocolError(ProtocolError::InvalidMessage);
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
let e = NegotiationError::ProtocolError(ProtocolError::InvalidMessage);
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
}
}
ProtocolError::InvalidProtocol =>
for h in self.handlers.values_mut() {
let e = NegotiationError::ProtocolError(ProtocolError::InvalidProtocol);
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
let e = NegotiationError::ProtocolError(ProtocolError::InvalidProtocol);
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
}
}
ProtocolError::TooManyProtocols =>
for h in self.handlers.values_mut() {
let e = NegotiationError::ProtocolError(ProtocolError::TooManyProtocols);
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
for (k, h) in &mut self.handlers {
if let Some(i) = info.take(k) {
let e = NegotiationError::ProtocolError(ProtocolError::TooManyProtocols);
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e)))
}
}
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply((k, e))) =>
if let Some(h) = self.handlers.get_mut(&k) {
h.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)))
if let Some(i) = info.take(&k) {
h.inject_listen_upgrade_error(i, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)))
}
}
}
}
@ -309,12 +337,37 @@ impl<H: ProtocolName> ProtocolName for IndexedProtoName<H> {
}
}
/// The aggregated `InboundOpenInfo`s of supported inbound substream protocols.
#[derive(Clone)]
pub struct Info<K, I> {
infos: Vec<(K, I)>
}
impl<K: Eq, I> Info<K, I> {
fn new() -> Self {
Info { infos: Vec::new() }
}
pub fn take(&mut self, k: &K) -> Option<I> {
if let Some(p) = self.infos.iter().position(|(key, _)| key == k) {
return Some(self.infos.remove(p).1)
}
None
}
}
/// Inbound and outbound upgrade for all `ProtocolsHandler`s.
#[derive(Clone)]
pub struct Upgrade<K, H> {
upgrades: Vec<(K, H)>
}
impl<K, H> Upgrade<K, H> {
fn new() -> Self {
Upgrade { upgrades: Vec::new() }
}
}
impl<K, H> fmt::Debug for Upgrade<K, H>
where
K: fmt::Debug + Eq + Hash,
@ -433,4 +486,3 @@ impl fmt::Display for DuplicateProtonameError {
}
impl error::Error for DuplicateProtonameError {}

31
swarm/src/protocols_handler/node_handler.rs
View File

@ -95,8 +95,11 @@ where
/// The underlying handler.
handler: TProtoHandler,
/// Futures that upgrade incoming substreams.
negotiating_in:
Vec<(InboundUpgradeApply<Substream<StreamMuxerBox>, SendWrapper<TProtoHandler::InboundProtocol>>, Delay)>,
negotiating_in: Vec<(
TProtoHandler::InboundOpenInfo,
InboundUpgradeApply<Substream<StreamMuxerBox>, SendWrapper<TProtoHandler::InboundProtocol>>,
Delay
)>,
/// Futures that upgrade outgoing substreams. The first element of the tuple is the userdata
/// to pass back once successfully opened.
negotiating_out: Vec<(
@ -192,9 +195,10 @@ where
SubstreamEndpoint::Listener => {
let protocol = self.handler.listen_protocol();
let timeout = protocol.timeout().clone();
let upgrade = upgrade::apply_inbound(substream, SendWrapper(protocol.into_upgrade().1));
let (_, upgrade, info) = protocol.into_upgrade();
let upgrade = upgrade::apply_inbound(substream, SendWrapper(upgrade));
let timeout = Delay::new(timeout);
self.negotiating_in.push((upgrade, timeout));
self.negotiating_in.push((info, upgrade, timeout));
}
SubstreamEndpoint::Dialer((upgrade_id, user_data, timeout)) => {
let pos = match self
@ -231,27 +235,27 @@ where
// Continue negotiation of newly-opened substreams on the listening side.
// We remove each element from `negotiating_in` one by one and add them back if not ready.
for n in (0..self.negotiating_in.len()).rev() {
let (mut in_progress, mut timeout) = self.negotiating_in.swap_remove(n);
let (info, mut in_progress, mut timeout) = self.negotiating_in.swap_remove(n);
match Future::poll(Pin::new(&mut timeout), cx) {
Poll::Ready(Ok(_)) => {
let err = ProtocolsHandlerUpgrErr::Timeout;
self.handler.inject_listen_upgrade_error(err);
self.handler.inject_listen_upgrade_error(info, err);
continue
}
Poll::Ready(Err(_)) => {
let err = ProtocolsHandlerUpgrErr::Timer;
self.handler.inject_listen_upgrade_error(err);
self.handler.inject_listen_upgrade_error(info, err);
continue;
}
Poll::Pending => {},
}
match Future::poll(Pin::new(&mut in_progress), cx) {
Poll::Ready(Ok(upgrade)) =>
self.handler.inject_fully_negotiated_inbound(upgrade),
Poll::Pending => self.negotiating_in.push((in_progress, timeout)),
self.handler.inject_fully_negotiated_inbound(upgrade, info),
Poll::Pending => self.negotiating_in.push((info, in_progress, timeout)),
Poll::Ready(Err(err)) => {
let err = ProtocolsHandlerUpgrErr::Upgrade(err);
self.handler.inject_listen_upgrade_error(err);
self.handler.inject_listen_upgrade_error(info, err);
}
}
}
@ -308,14 +312,11 @@ where
Poll::Ready(ProtocolsHandlerEvent::Custom(event)) => {
return Poll::Ready(Ok(ConnectionHandlerEvent::Custom(event)));
}
Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol,
info,
}) => {
Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
let id = self.unique_dial_upgrade_id;
let timeout = protocol.timeout().clone();
self.unique_dial_upgrade_id += 1;
let (version, upgrade) = protocol.into_upgrade();
let (version, upgrade, info) = protocol.into_upgrade();
self.queued_dial_upgrades.push((id, (version, SendWrapper(upgrade))));
return Poll::Ready(Ok(
ConnectionHandlerEvent::OutboundSubstreamRequest((id, info, timeout)),

26
swarm/src/protocols_handler/one_shot.rs
View File

@ -38,7 +38,7 @@ where
TOutbound: OutboundUpgradeSend,
{
/// The upgrade for inbound substreams.
listen_protocol: SubstreamProtocol<TInbound>,
listen_protocol: SubstreamProtocol<TInbound, ()>,
/// If `Some`, something bad happened and we should shut down the handler with an error.
pending_error: Option<ProtocolsHandlerUpgrErr<<TOutbound as OutboundUpgradeSend>::Error>>,
/// Queue of events to produce in `poll()`.
@ -62,7 +62,7 @@ where
{
/// Creates a `OneShotHandler`.
pub fn new(
listen_protocol: SubstreamProtocol<TInbound>,
listen_protocol: SubstreamProtocol<TInbound, ()>,
config: OneShotHandlerConfig,
) -> Self {
OneShotHandler {
@ -86,7 +86,7 @@ where
///
/// > **Note**: If you modify the protocol, modifications will only applies to future inbound
/// > substreams, not the ones already being negotiated.
pub fn listen_protocol_ref(&self) -> &SubstreamProtocol<TInbound> {
pub fn listen_protocol_ref(&self) -> &SubstreamProtocol<TInbound, ()> {
&self.listen_protocol
}
@ -94,7 +94,7 @@ where
///
/// > **Note**: If you modify the protocol, modifications will only applies to future inbound
/// > substreams, not the ones already being negotiated.
pub fn listen_protocol_mut(&mut self) -> &mut SubstreamProtocol<TInbound> {
pub fn listen_protocol_mut(&mut self) -> &mut SubstreamProtocol<TInbound, ()> {
&mut self.listen_protocol
}
@ -113,21 +113,20 @@ where
{
fn default() -> Self {
OneShotHandler::new(
SubstreamProtocol::new(Default::default()),
SubstreamProtocol::new(Default::default(), ()),
OneShotHandlerConfig::default()
)
}
}
impl<TInbound, TOutbound, TEvent> ProtocolsHandler
for OneShotHandler<TInbound, TOutbound, TEvent>
impl<TInbound, TOutbound, TEvent> ProtocolsHandler for OneShotHandler<TInbound, TOutbound, TEvent>
where
TInbound: InboundUpgradeSend + Send + 'static,
TOutbound: OutboundUpgradeSend,
TInbound::Output: Into<TEvent>,
TOutbound::Output: Into<TEvent>,
TOutbound::Error: error::Error + Send + 'static,
SubstreamProtocol<TInbound>: Clone,
SubstreamProtocol<TInbound, ()>: Clone,
TEvent: Send + 'static,
{
type InEvent = TOutbound;
@ -138,14 +137,16 @@ where
type InboundProtocol = TInbound;
type OutboundProtocol = TOutbound;
type OutboundOpenInfo = ();
type InboundOpenInfo = ();
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
self.listen_protocol.clone()
}
fn inject_fully_negotiated_inbound(
&mut self,
out: <Self::InboundProtocol as InboundUpgradeSend>::Output,
(): Self::InboundOpenInfo
) {
// If we're shutting down the connection for inactivity, reset the timeout.
if !self.keep_alive.is_yes() {
@ -208,9 +209,8 @@ where
let upgrade = self.dial_queue.remove(0);
return Poll::Ready(
ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(upgrade)
.with_timeout(self.config.outbound_substream_timeout),
info: (),
protocol: SubstreamProtocol::new(upgrade, ())
.with_timeout(self.config.outbound_substream_timeout)
},
);
}
@ -256,7 +256,7 @@ mod tests {
#[test]
fn do_not_keep_idle_connection_alive() {
let mut handler: OneShotHandler<_, DeniedUpgrade, Void> = OneShotHandler::new(
SubstreamProtocol::new(DeniedUpgrade{}),
SubstreamProtocol::new(DeniedUpgrade{}, ()),
Default::default(),
);

64
swarm/src/protocols_handler/select.rs
View File

@ -105,13 +105,16 @@ where
type InboundProtocol = SelectUpgrade<SendWrapper<<TProto1 as ProtocolsHandler>::InboundProtocol>, SendWrapper<<TProto2 as ProtocolsHandler>::InboundProtocol>>;
type OutboundProtocol = EitherUpgrade<SendWrapper<TProto1::OutboundProtocol>, SendWrapper<TProto2::OutboundProtocol>>;
type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>;
type InboundOpenInfo = (TProto1::InboundOpenInfo, TProto2::InboundOpenInfo);
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
let proto1 = self.proto1.listen_protocol();
let proto2 = self.proto2.listen_protocol();
let timeout = std::cmp::max(proto1.timeout(), proto2.timeout()).clone();
let choice = SelectUpgrade::new(SendWrapper(proto1.into_upgrade().1), SendWrapper(proto2.into_upgrade().1));
SubstreamProtocol::new(choice).with_timeout(timeout)
let (_, u1, i1) = proto1.into_upgrade();
let (_, u2, i2) = proto2.into_upgrade();
let choice = SelectUpgrade::new(SendWrapper(u1), SendWrapper(u2));
SubstreamProtocol::new(choice, (i1, i2)).with_timeout(timeout)
}
fn inject_fully_negotiated_outbound(&mut self, protocol: <Self::OutboundProtocol as OutboundUpgradeSend>::Output, endpoint: Self::OutboundOpenInfo) {
@ -127,12 +130,12 @@ where
}
}
fn inject_fully_negotiated_inbound(&mut self, protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output) {
fn inject_fully_negotiated_inbound(&mut self, protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output, (i1, i2): Self::InboundOpenInfo) {
match protocol {
EitherOutput::First(protocol) =>
self.proto1.inject_fully_negotiated_inbound(protocol),
self.proto1.inject_fully_negotiated_inbound(protocol, i1),
EitherOutput::Second(protocol) =>
self.proto2.inject_fully_negotiated_inbound(protocol)
self.proto2.inject_fully_negotiated_inbound(protocol, i2)
}
}
@ -143,9 +146,9 @@ where
}
}
fn inject_address_change(&mut self, addr: &Multiaddr) {
self.proto1.inject_address_change(addr);
self.proto2.inject_address_change(addr)
fn inject_address_change(&mut self, new_address: &Multiaddr) {
self.proto1.inject_address_change(new_address);
self.proto2.inject_address_change(new_address)
}
fn inject_dial_upgrade_error(&mut self, info: Self::OutboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::OutboundProtocol as OutboundUpgradeSend>::Error>) {
@ -183,19 +186,19 @@ where
}
}
fn inject_listen_upgrade_error(&mut self, error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>) {
fn inject_listen_upgrade_error(&mut self, (i1, i2): Self::InboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>) {
match error {
ProtocolsHandlerUpgrErr::Timer => {
self.proto1.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Timer);
self.proto2.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Timer);
self.proto1.inject_listen_upgrade_error(i1, ProtocolsHandlerUpgrErr::Timer);
self.proto2.inject_listen_upgrade_error(i2, ProtocolsHandlerUpgrErr::Timer)
}
ProtocolsHandlerUpgrErr::Timeout => {
self.proto1.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Timeout);
self.proto2.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Timeout);
self.proto1.inject_listen_upgrade_error(i1, ProtocolsHandlerUpgrErr::Timeout);
self.proto2.inject_listen_upgrade_error(i2, ProtocolsHandlerUpgrErr::Timeout)
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)) => {
self.proto1.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)));
self.proto2.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)));
self.proto1.inject_listen_upgrade_error(i1, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)));
self.proto2.inject_listen_upgrade_error(i2, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::Failed)));
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(NegotiationError::ProtocolError(e))) => {
let (e1, e2);
@ -217,14 +220,14 @@ where
e2 = NegotiationError::ProtocolError(ProtocolError::TooManyProtocols)
}
}
self.proto1.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e1)));
self.proto2.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e2)))
self.proto1.inject_listen_upgrade_error(i1, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e1)));
self.proto2.inject_listen_upgrade_error(i2, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(e2)))
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(e))) => {
self.proto1.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)))
self.proto1.inject_listen_upgrade_error(i1, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)))
}
ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(e))) => {
self.proto2.inject_listen_upgrade_error(ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)))
self.proto2.inject_listen_upgrade_error(i2, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(e)))
}
}
}
@ -234,7 +237,6 @@ where
}
fn poll(&mut self, cx: &mut Context<'_>) -> Poll<ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent, Self::Error>> {
match self.proto1.poll(cx) {
Poll::Ready(ProtocolsHandlerEvent::Custom(event)) => {
return Poll::Ready(ProtocolsHandlerEvent::Custom(EitherOutput::First(event)));
@ -242,13 +244,11 @@ where
Poll::Ready(ProtocolsHandlerEvent::Close(event)) => {
return Poll::Ready(ProtocolsHandlerEvent::Close(EitherError::A(event)));
},
Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol,
info,
}) => {
Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: protocol.map_upgrade(|u| EitherUpgrade::A(SendWrapper(u))),
info: EitherOutput::First(info),
protocol: protocol
.map_upgrade(|u| EitherUpgrade::A(SendWrapper(u)))
.map_info(EitherOutput::First)
});
},
Poll::Pending => ()
@ -261,13 +261,11 @@ where
Poll::Ready(ProtocolsHandlerEvent::Close(event)) => {
return Poll::Ready(ProtocolsHandlerEvent::Close(EitherError::B(event)));
},
Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol,
info,
}) => {
Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol }) => {
return Poll::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: protocol.map_upgrade(|u| EitherUpgrade::B(SendWrapper(u))),
info: EitherOutput::Second(info),
protocol: protocol
.map_upgrade(|u| EitherUpgrade::B(SendWrapper(u)))
.map_info(EitherOutput::Second)
});
},
Poll::Pending => ()

45
swarm/src/toggle.rs
View File

@ -28,6 +28,7 @@ use crate::protocols_handler::{
ProtocolsHandlerUpgrErr,
IntoProtocolsHandler
};
use either::Either;
use libp2p_core::{
ConnectedPoint,
PeerId,
@ -213,26 +214,35 @@ where
type InboundProtocol = EitherUpgrade<SendWrapper<TInner::InboundProtocol>, SendWrapper<DeniedUpgrade>>;
type OutboundProtocol = TInner::OutboundProtocol;
type OutboundOpenInfo = TInner::OutboundOpenInfo;
type InboundOpenInfo = Either<TInner::InboundOpenInfo, ()>;
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
if let Some(inner) = self.inner.as_ref() {
inner.listen_protocol().map_upgrade(|u| EitherUpgrade::A(SendWrapper(u)))
inner.listen_protocol()
.map_upgrade(|u| EitherUpgrade::A(SendWrapper(u)))
.map_info(Either::Left)
} else {
SubstreamProtocol::new(EitherUpgrade::B(SendWrapper(DeniedUpgrade)))
SubstreamProtocol::new(EitherUpgrade::B(SendWrapper(DeniedUpgrade)), Either::Right(()))
}
}
fn inject_fully_negotiated_inbound(
&mut self,
out: <Self::InboundProtocol as InboundUpgradeSend>::Output
out: <Self::InboundProtocol as InboundUpgradeSend>::Output,
info: Self::InboundOpenInfo
) {
let out = match out {
EitherOutput::First(out) => out,
EitherOutput::Second(v) => void::unreachable(v),
};
self.inner.as_mut().expect("Can't receive an inbound substream if disabled; QED")
.inject_fully_negotiated_inbound(out)
if let Either::Left(info) = info {
self.inner.as_mut()
.expect("Can't receive an inbound substream if disabled; QED")
.inject_fully_negotiated_inbound(out, info)
} else {
panic!("Unpexpected Either::Right in enabled `inject_fully_negotiated_inbound`.")
}
}
fn inject_fully_negotiated_outbound(
@ -260,13 +270,22 @@ where
.inject_dial_upgrade_error(info, err)
}
fn inject_listen_upgrade_error(&mut self, err: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>) {
if let Some(inner) = self.inner.as_mut() {
let err = err.map_upgrade_err(|e| e.map_err(|e| match e {
EitherError::A(e) => e,
EitherError::B(v) => void::unreachable(v)
}));
inner.inject_listen_upgrade_error(err)
fn inject_listen_upgrade_error(&mut self, info: Self::InboundOpenInfo, err: ProtocolsHandlerUpgrErr<<Self::InboundProtocol as InboundUpgradeSend>::Error>) {
let err = match err {
ProtocolsHandlerUpgrErr::Timeout => ProtocolsHandlerUpgrErr::Timeout,
ProtocolsHandlerUpgrErr::Timer => ProtocolsHandlerUpgrErr::Timer,
ProtocolsHandlerUpgrErr::Upgrade(err) =>
ProtocolsHandlerUpgrErr::Upgrade(err.map_err(|err| match err {
EitherError::A(e) => e,
EitherError::B(v) => void::unreachable(v)
}))
};
if let Either::Left(info) = info {
self.inner.as_mut()
.expect("Can't receive an inbound substream if disabled; QED")
.inject_listen_upgrade_error(info, err)
} else {
panic!("Unexpected Either::Right on enabled `inject_listen_upgrade_error`.")
}
}