Move swarm and protocols handler into swarm crate. (#1188)

Move swarm and protocols handler into swarm crate.

swarm/src/protocols_handler/dummy.rs

@@ -0,0 +1,98 @@
+// 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::protocols_handler::{
+    KeepAlive,
+    SubstreamProtocol,
+    ProtocolsHandler,
+    ProtocolsHandlerEvent,
+    ProtocolsHandlerUpgrErr
+};
+use futures::prelude::*;
+use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade, DeniedUpgrade};
+use std::marker::PhantomData;
+use tokio_io::{AsyncRead, AsyncWrite};
+use void::Void;
+
+/// Implementation of `ProtocolsHandler` that doesn't handle anything.
+pub struct DummyProtocolsHandler<TSubstream> {
+    marker: PhantomData<TSubstream>,
+}
+
+impl<TSubstream> Default for DummyProtocolsHandler<TSubstream> {
+    #[inline]
+    fn default() -> Self {
+        DummyProtocolsHandler {
+            marker: PhantomData,
+        }
+    }
+}
+
+impl<TSubstream> ProtocolsHandler for DummyProtocolsHandler<TSubstream>
+where
+    TSubstream: AsyncRead + AsyncWrite,
+{
+    type InEvent = Void;
+    type OutEvent = Void;
+    type Error = Void;
+    type Substream = TSubstream;
+    type InboundProtocol = DeniedUpgrade;
+    type OutboundProtocol = DeniedUpgrade;
+    type OutboundOpenInfo = Void;
+
+    #[inline]
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+        SubstreamProtocol::new(DeniedUpgrade)
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_inbound(
+        &mut self,
+        _: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output
+    ) {
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_outbound(
+        &mut self,
+        _: <Self::OutboundProtocol as OutboundUpgrade<TSubstream>>::Output,
+        _: Self::OutboundOpenInfo
+    ) {
+    }
+
+    #[inline]
+    fn inject_event(&mut self, _: Self::InEvent) {}
+
+    #[inline]
+    fn inject_dial_upgrade_error(&mut self, _: Self::OutboundOpenInfo, _: ProtocolsHandlerUpgrErr<<Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error>) {}
+
+    #[inline]
+    fn connection_keep_alive(&self) -> KeepAlive { KeepAlive::No }
+
+    #[inline]
+    fn poll(
+        &mut self,
+    ) -> Poll<
+        ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent>,
+        Void,
+    > {
+        Ok(Async::NotReady)
+    }
+}

swarm/src/protocols_handler/map_in.rs

@@ -0,0 +1,112 @@
+// 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::protocols_handler::{
+    KeepAlive,
+    SubstreamProtocol,
+    ProtocolsHandler,
+    ProtocolsHandlerEvent,
+    ProtocolsHandlerUpgrErr
+};
+use futures::prelude::*;
+use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade};
+use std::marker::PhantomData;
+
+/// Wrapper around a protocol handler that turns the input event into something else.
+pub struct MapInEvent<TProtoHandler, TNewIn, TMap> {
+    inner: TProtoHandler,
+    map: TMap,
+    marker: PhantomData<TNewIn>,
+}
+
+impl<TProtoHandler, TMap, TNewIn> MapInEvent<TProtoHandler, TNewIn, TMap> {
+    /// Creates a `MapInEvent`.
+    #[inline]
+    pub(crate) fn new(inner: TProtoHandler, map: TMap) -> Self {
+        MapInEvent {
+            inner,
+            map,
+            marker: PhantomData,
+        }
+    }
+}
+
+impl<TProtoHandler, TMap, TNewIn> ProtocolsHandler for MapInEvent<TProtoHandler, TNewIn, TMap>
+where
+    TProtoHandler: ProtocolsHandler,
+    TMap: Fn(TNewIn) -> Option<TProtoHandler::InEvent>,
+{
+    type InEvent = TNewIn;
+    type OutEvent = TProtoHandler::OutEvent;
+    type Error = TProtoHandler::Error;
+    type Substream = TProtoHandler::Substream;
+    type InboundProtocol = TProtoHandler::InboundProtocol;
+    type OutboundProtocol = TProtoHandler::OutboundProtocol;
+    type OutboundOpenInfo = TProtoHandler::OutboundOpenInfo;
+
+    #[inline]
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+        self.inner.listen_protocol()
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_inbound(
+        &mut self,
+        protocol: <Self::InboundProtocol as InboundUpgrade<Self::Substream>>::Output
+    ) {
+        self.inner.inject_fully_negotiated_inbound(protocol)
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_outbound(
+        &mut self,
+        protocol: <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Output,
+        info: Self::OutboundOpenInfo
+    ) {
+        self.inner.inject_fully_negotiated_outbound(protocol, info)
+    }
+
+    #[inline]
+    fn inject_event(&mut self, event: TNewIn) {
+        if let Some(event) = (self.map)(event) {
+            self.inner.inject_event(event);
+        }
+    }
+
+    #[inline]
+    fn inject_dial_upgrade_error(&mut self, info: Self::OutboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error>) {
+        self.inner.inject_dial_upgrade_error(info, error)
+    }
+
+    #[inline]
+    fn connection_keep_alive(&self) -> KeepAlive {
+        self.inner.connection_keep_alive()
+    }
+
+    #[inline]
+    fn poll(
+        &mut self,
+    ) -> Poll<
+        ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent>,
+        Self::Error,
+    > {
+        self.inner.poll()
+    }
+}

swarm/src/protocols_handler/map_out.rs

@@ -0,0 +1,114 @@
+// 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::protocols_handler::{
+    KeepAlive,
+    SubstreamProtocol,
+    ProtocolsHandler,
+    ProtocolsHandlerEvent,
+    ProtocolsHandlerUpgrErr
+};
+use futures::prelude::*;
+use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade};
+
+/// Wrapper around a protocol handler that turns the output event into something else.
+pub struct MapOutEvent<TProtoHandler, TMap> {
+    inner: TProtoHandler,
+    map: TMap,
+}
+
+impl<TProtoHandler, TMap> MapOutEvent<TProtoHandler, TMap> {
+    /// Creates a `MapOutEvent`.
+    #[inline]
+    pub(crate) fn new(inner: TProtoHandler, map: TMap) -> Self {
+        MapOutEvent {
+            inner,
+            map,
+        }
+    }
+}
+
+impl<TProtoHandler, TMap, TNewOut> ProtocolsHandler for MapOutEvent<TProtoHandler, TMap>
+where
+    TProtoHandler: ProtocolsHandler,
+    TMap: FnMut(TProtoHandler::OutEvent) -> TNewOut,
+{
+    type InEvent = TProtoHandler::InEvent;
+    type OutEvent = TNewOut;
+    type Error = TProtoHandler::Error;
+    type Substream = TProtoHandler::Substream;
+    type InboundProtocol = TProtoHandler::InboundProtocol;
+    type OutboundProtocol = TProtoHandler::OutboundProtocol;
+    type OutboundOpenInfo = TProtoHandler::OutboundOpenInfo;
+
+    #[inline]
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+        self.inner.listen_protocol()
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_inbound(
+        &mut self,
+        protocol: <Self::InboundProtocol as InboundUpgrade<Self::Substream>>::Output
+    ) {
+        self.inner.inject_fully_negotiated_inbound(protocol)
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_outbound(
+        &mut self,
+        protocol: <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Output,
+        info: Self::OutboundOpenInfo
+    ) {
+        self.inner.inject_fully_negotiated_outbound(protocol, info)
+    }
+
+    #[inline]
+    fn inject_event(&mut self, event: Self::InEvent) {
+        self.inner.inject_event(event)
+    }
+
+    #[inline]
+    fn inject_dial_upgrade_error(&mut self, info: Self::OutboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error>) {
+        self.inner.inject_dial_upgrade_error(info, error)
+    }
+
+    #[inline]
+    fn connection_keep_alive(&self) -> KeepAlive {
+        self.inner.connection_keep_alive()
+    }
+
+    #[inline]
+    fn poll(
+        &mut self,
+    ) -> Poll<
+        ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent>,
+        Self::Error,
+    > {
+        Ok(self.inner.poll()?.map(|ev| {
+            match ev {
+                ProtocolsHandlerEvent::Custom(ev) => ProtocolsHandlerEvent::Custom((self.map)(ev)),
+                ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
+                    ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }
+                }
+            }
+        }))
+    }
+}

swarm/src/protocols_handler/mod.rs

@@ -0,0 +1,505 @@
+// 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.
+
+//! Once a connection to a remote peer is established, a `ProtocolsHandler` negotiates
+//! and handles one or more specific protocols on the connection.
+//!
+//! Protocols are negotiated and used on individual substreams of the connection.
+//! Thus a `ProtocolsHandler` defines the inbound and outbound upgrades to apply
+//! when creating a new inbound or outbound substream, respectively, and is notified
+//! by a `Swarm` when these upgrades have been successfully applied, including the
+//! final output of the upgrade. A `ProtocolsHandler` can then continue communicating
+//! with the peer over the substream using the negotiated protocol(s).
+//!
+//! Two `ProtocolsHandler`s can be composed with [`ProtocolsHandler::select()`]
+//! in order to build a new handler supporting the combined set of protocols,
+//! with methods being dispatched to the appropriate handler according to the
+//! used protocol(s) determined by the associated types of the handlers.
+//!
+//! > **Note**: A `ProtocolsHandler` handles one or more protocols in the context of a single
+//! >           connection with a remote. In order to handle a protocol that requires knowledge of
+//! >           the network as a whole, see the `NetworkBehaviour` trait.
+
+mod dummy;
+mod map_in;
+mod map_out;
+mod node_handler;
+mod one_shot;
+mod select;
+
+use futures::prelude::*;
+use libp2p_core::{
+    ConnectedPoint,
+    PeerId,
+    upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeError},
+};
+use std::{cmp::Ordering, error, fmt, time::Duration};
+use tokio_io::{AsyncRead, AsyncWrite};
+use wasm_timer::Instant;
+
+pub use dummy::DummyProtocolsHandler;
+pub use map_in::MapInEvent;
+pub use map_out::MapOutEvent;
+pub use node_handler::{NodeHandlerWrapper, NodeHandlerWrapperBuilder, NodeHandlerWrapperError};
+pub use one_shot::OneShotHandler;
+pub use select::{IntoProtocolsHandlerSelect, ProtocolsHandlerSelect};
+
+/// A handler for a set of protocols used on a connection with a remote.
+///
+/// This trait should be implemented for a type that maintains the state for
+/// the execution of a specific protocol with a remote.
+///
+/// # Handling a protocol
+///
+/// Communication with a remote over a set of protocols is initiated in one of two ways:
+///
+///   1. Dialing by initiating a new outbound substream. In order to do so,
+///      [`ProtocolsHandler::poll()`] must return an [`OutboundSubstreamRequest`], providing an
+///      instance of [`ProtocolsHandler::OutboundUpgrade`] that is used to negotiate the
+///      protocol(s). Upon success, [`ProtocolsHandler::inject_fully_negotiated_outbound`]
+///      is called with the final output of the upgrade.
+///
+///   2. Listening by accepting a new inbound substream. When a new inbound substream
+///      is created on a connection, [`ProtocolsHandler::listen_protocol`] is called
+///      to obtain an instance of [`ProtocolsHandler::InboundUpgrade`] that is used to
+///      negotiate the protocol(s). Upon success,
+///      [`ProtocolsHandler::inject_fully_negotiated_inbound`] is called with the final
+///      output of the upgrade.
+///
+/// # Connection Keep-Alive
+///
+/// A `ProtocolsHandler` can influence the lifetime of the underlying connection
+/// through [`ProtocolsHandler::connection_keep_alive`]. That is, the protocol
+/// implemented by the handler can include conditions for terminating the connection.
+/// The lifetime of successfully negotiated substreams is fully controlled by the handler.
+///
+/// Implementors of this trait should keep in mind that the connection can be closed at any time.
+/// When a connection is closed gracefully, the substreams used by the handler may still
+/// continue reading data until the remote closes its side of the connection.
+pub trait ProtocolsHandler {
+    /// Custom event that can be received from the outside.
+    type InEvent;
+    /// Custom event that can be produced by the handler and that will be returned to the outside.
+    type OutEvent;
+    /// The type of errors returned by [`ProtocolsHandler::poll`].
+    type Error: error::Error;
+    /// The type of substreams on which the protocol(s) are negotiated.
+    type Substream: AsyncRead + AsyncWrite;
+    /// The inbound upgrade for the protocol(s) used by the handler.
+    type InboundProtocol: InboundUpgrade<Self::Substream>;
+    /// The outbound upgrade for the protocol(s) used by the handler.
+    type OutboundProtocol: OutboundUpgrade<Self::Substream>;
+    /// The type of additional information passed to an `OutboundSubstreamRequest`.
+    type OutboundOpenInfo;
+
+    /// The [`InboundUpgrade`] to apply on inbound substreams to negotiate the
+    /// desired protocols.
+    ///
+    /// > **Note**: The returned `InboundUpgrade` should always accept all the generally
+    /// >           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>;
+
+    /// Injects the output of a successful upgrade on a new inbound substream.
+    fn inject_fully_negotiated_inbound(
+        &mut self,
+        protocol: <Self::InboundProtocol as InboundUpgrade<Self::Substream>>::Output
+    );
+
+    /// Injects the output of a successful upgrade on a new outbound substream.
+    ///
+    /// The second argument is the information that was previously passed to
+    /// [`ProtocolsHandlerEvent::OutboundSubstreamRequest`].
+    fn inject_fully_negotiated_outbound(
+        &mut self,
+        protocol: <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Output,
+        info: Self::OutboundOpenInfo
+    );
+
+    /// Injects an event coming from the outside in the handler.
+    fn inject_event(&mut self, event: Self::InEvent);
+
+    /// Indicates to the handler that upgrading a substream to the given protocol has failed.
+    fn inject_dial_upgrade_error(
+        &mut self,
+        info: Self::OutboundOpenInfo,
+        error: ProtocolsHandlerUpgrErr<
+            <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error
+        >
+    );
+
+    /// Returns until when the connection should be kept alive.
+    ///
+    /// This method is called by the `Swarm` after each invocation of
+    /// [`ProtocolsHandler::poll`] to determine if the connection and the associated
+    /// `ProtocolsHandler`s should be kept alive as far as this handler is concerned
+    /// and if so, for how long.
+    ///
+    /// Returning [`KeepAlive::No`] indicates that the connection should be
+    /// closed and this handler destroyed immediately.
+    ///
+    /// Returning [`KeepAlive::Until`] indicates that the connection may be closed
+    /// and this handler destroyed after the specified `Instant`.
+    ///
+    /// Returning [`KeepAlive::Yes`] indicates that the connection should
+    /// be kept alive until the next call to this method.
+    ///
+    /// > **Note**: The connection is always closed and the handler destroyed
+    /// > when [`ProtocolsHandler::poll`] returns an error. Furthermore, the
+    /// > connection may be closed for reasons outside of the control
+    /// > of the handler.
+    fn connection_keep_alive(&self) -> KeepAlive;
+
+    /// Should behave like `Stream::poll()`.
+    ///
+    /// Returning an error will close the connection to the remote.
+    fn poll(&mut self) -> Poll<
+        ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent>,
+        Self::Error
+    >;
+
+    /// Adds a closure that turns the input event into something else.
+    #[inline]
+    fn map_in_event<TNewIn, TMap>(self, map: TMap) -> MapInEvent<Self, TNewIn, TMap>
+    where
+        Self: Sized,
+        TMap: Fn(&TNewIn) -> Option<&Self::InEvent>,
+    {
+        MapInEvent::new(self, map)
+    }
+
+    /// Adds a closure that turns the output event into something else.
+    #[inline]
+    fn map_out_event<TMap, TNewOut>(self, map: TMap) -> MapOutEvent<Self, TMap>
+    where
+        Self: Sized,
+        TMap: FnMut(Self::OutEvent) -> TNewOut,
+    {
+        MapOutEvent::new(self, map)
+    }
+
+    /// Creates a new `ProtocolsHandler` that selects either this handler or
+    /// `other` by delegating methods calls appropriately.
+    ///
+    /// > **Note**: The largest `KeepAlive` returned by the two handlers takes precedence,
+    /// > i.e. is returned from [`ProtocolsHandler::connection_keep_alive`] by the returned
+    /// > handler.
+    #[inline]
+    fn select<TProto2>(self, other: TProto2) -> ProtocolsHandlerSelect<Self, TProto2>
+    where
+        Self: Sized,
+    {
+        ProtocolsHandlerSelect::new(self, other)
+    }
+
+    /// Creates a builder that allows creating a `NodeHandler` that handles this protocol
+    /// exclusively.
+    ///
+    /// > **Note**: This method should not be redefined in a custom `ProtocolsHandler`.
+    #[inline]
+    fn into_node_handler_builder(self) -> NodeHandlerWrapperBuilder<Self>
+    where
+        Self: Sized,
+    {
+        IntoProtocolsHandler::into_node_handler_builder(self)
+    }
+
+    /// Builds an implementation of `NodeHandler` that handles this protocol exclusively.
+    ///
+    /// > **Note**: This is a shortcut for `self.into_node_handler_builder().build()`.
+    #[inline]
+    #[deprecated(note = "Use into_node_handler_builder instead")]
+    fn into_node_handler(self) -> NodeHandlerWrapper<Self>
+    where
+        Self: Sized,
+    {
+        #![allow(deprecated)]
+        self.into_node_handler_builder().build()
+    }
+}
+
+/// Configuration of inbound or outbound substream protocol(s)
+/// for a [`ProtocolsHandler`].
+///
+/// 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> {
+    upgrade: TUpgrade,
+    timeout: Duration,
+}
+
+impl<TUpgrade> SubstreamProtocol<TUpgrade> {
+    /// Create a new `ListenProtocol` 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> {
+        SubstreamProtocol {
+            upgrade,
+            timeout: Duration::from_secs(10),
+        }
+    }
+
+    /// Maps a function over the protocol upgrade.
+    pub fn map_upgrade<U, F>(self, f: F) -> SubstreamProtocol<U>
+    where
+        F: FnOnce(TUpgrade) -> U,
+    {
+        SubstreamProtocol {
+            upgrade: f(self.upgrade),
+            timeout: self.timeout,
+        }
+    }
+
+    /// Sets a new timeout for the protocol upgrade.
+    pub fn with_timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = timeout;
+        self
+    }
+
+    /// Borrows the contained protocol upgrade.
+    pub fn upgrade(&self) -> &TUpgrade {
+        &self.upgrade
+    }
+
+    /// 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) -> TUpgrade {
+        self.upgrade
+    }
+}
+
+impl<TUpgrade> From<TUpgrade> for SubstreamProtocol<TUpgrade> {
+    fn from(upgrade: TUpgrade) -> SubstreamProtocol<TUpgrade> {
+        SubstreamProtocol::new(upgrade)
+    }
+}
+
+/// Event produced by a handler.
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum ProtocolsHandlerEvent<TConnectionUpgrade, TOutboundOpenInfo, TCustom> {
+    /// 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,
+    },
+
+    /// Other event.
+    Custom(TCustom),
+}
+
+/// Event produced by a handler.
+impl<TConnectionUpgrade, TOutboundOpenInfo, TCustom>
+    ProtocolsHandlerEvent<TConnectionUpgrade, TOutboundOpenInfo, TCustom>
+{
+    /// If this is an `OutboundSubstreamRequest`, maps the `info` member from a
+    /// `TOutboundOpenInfo` to something else.
+    #[inline]
+    pub fn map_outbound_open_info<F, I>(
+        self,
+        map: F,
+    ) -> ProtocolsHandlerEvent<TConnectionUpgrade, I, TCustom>
+    where
+        F: FnOnce(TOutboundOpenInfo) -> I,
+    {
+        match self {
+            ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
+                ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                    protocol,
+                    info: map(info),
+                }
+            }
+            ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(val),
+        }
+    }
+
+    /// If this is an `OutboundSubstreamRequest`, maps the protocol (`TConnectionUpgrade`)
+    /// to something else.
+    #[inline]
+    pub fn map_protocol<F, I>(
+        self,
+        map: F,
+    ) -> ProtocolsHandlerEvent<I, TOutboundOpenInfo, TCustom>
+    where
+        F: FnOnce(TConnectionUpgrade) -> I,
+    {
+        match self {
+            ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
+                ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                    protocol: protocol.map_upgrade(map),
+                    info,
+                }
+            }
+            ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(val),
+        }
+    }
+
+    /// If this is a `Custom` event, maps the content to something else.
+    #[inline]
+    pub fn map_custom<F, I>(
+        self,
+        map: F,
+    ) -> ProtocolsHandlerEvent<TConnectionUpgrade, TOutboundOpenInfo, I>
+    where
+        F: FnOnce(TCustom) -> I,
+    {
+        match self {
+            ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info } => {
+                ProtocolsHandlerEvent::OutboundSubstreamRequest { protocol, info }
+            }
+            ProtocolsHandlerEvent::Custom(val) => ProtocolsHandlerEvent::Custom(map(val)),
+        }
+    }
+}
+
+/// Error that can happen on an outbound substream opening attempt.
+#[derive(Debug)]
+pub enum ProtocolsHandlerUpgrErr<TUpgrErr> {
+    /// The opening attempt timed out before the negotiation was fully completed.
+    Timeout,
+    /// There was an error in the timer used.
+    Timer,
+    /// Error while upgrading the substream to the protocol we want.
+    Upgrade(UpgradeError<TUpgrErr>),
+}
+
+impl<TUpgrErr> fmt::Display for ProtocolsHandlerUpgrErr<TUpgrErr>
+where
+    TUpgrErr: fmt::Display,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ProtocolsHandlerUpgrErr::Timeout => {
+                write!(f, "Timeout error while opening a substream")
+            },
+            ProtocolsHandlerUpgrErr::Timer => {
+                write!(f, "Timer error while opening a substream")
+            },
+            ProtocolsHandlerUpgrErr::Upgrade(err) => write!(f, "{}", err),
+        }
+    }
+}
+
+impl<TUpgrErr> error::Error for ProtocolsHandlerUpgrErr<TUpgrErr>
+where
+    TUpgrErr: error::Error + 'static
+{
+    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
+        match self {
+            ProtocolsHandlerUpgrErr::Timeout => None,
+            ProtocolsHandlerUpgrErr::Timer => None,
+            ProtocolsHandlerUpgrErr::Upgrade(err) => Some(err),
+        }
+    }
+}
+
+/// Prototype for a `ProtocolsHandler`.
+pub trait IntoProtocolsHandler {
+    /// The protocols handler.
+    type Handler: ProtocolsHandler;
+
+    /// Builds the protocols handler.
+    ///
+    /// The `PeerId` is the id of the node the handler is going to handle.
+    fn into_handler(self, remote_peer_id: &PeerId, connected_point: &ConnectedPoint) -> Self::Handler;
+
+    /// Return the handler's inbound protocol.
+    fn inbound_protocol(&self) -> <Self::Handler as ProtocolsHandler>::InboundProtocol;
+
+    /// Builds an implementation of `IntoProtocolsHandler` that handles both this protocol and the
+    /// other one together.
+    fn select<TProto2>(self, other: TProto2) -> IntoProtocolsHandlerSelect<Self, TProto2>
+    where
+        Self: Sized,
+    {
+        IntoProtocolsHandlerSelect::new(self, other)
+    }
+
+    /// Creates a builder that will allow creating a `NodeHandler` that handles this protocol
+    /// exclusively.
+    fn into_node_handler_builder(self) -> NodeHandlerWrapperBuilder<Self>
+    where
+        Self: Sized,
+    {
+        NodeHandlerWrapperBuilder::new(self)
+    }
+}
+
+impl<T> IntoProtocolsHandler for T
+where T: ProtocolsHandler
+{
+    type Handler = Self;
+
+    fn into_handler(self, _: &PeerId, _: &ConnectedPoint) -> Self {
+        self
+    }
+
+    fn inbound_protocol(&self) -> <Self::Handler as ProtocolsHandler>::InboundProtocol {
+        self.listen_protocol().into_upgrade()
+    }
+}
+
+/// How long the connection should be kept alive.
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum KeepAlive {
+    /// If nothing new happens, the connection should be closed at the given `Instant`.
+    Until(Instant),
+    /// Keep the connection alive.
+    Yes,
+    /// Close the connection as soon as possible.
+    No,
+}
+
+impl KeepAlive {
+    /// Returns true for `Yes`, false otherwise.
+    pub fn is_yes(&self) -> bool {
+        match *self {
+            KeepAlive::Yes => true,
+            _ => false,
+        }
+    }
+}
+
+impl PartialOrd for KeepAlive {
+    fn partial_cmp(&self, other: &KeepAlive) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for KeepAlive {
+    fn cmp(&self, other: &KeepAlive) -> Ordering {
+        use self::KeepAlive::*;
+
+        match (self, other) {
+            (No, No) | (Yes, Yes)  => Ordering::Equal,
+            (No,  _) | (_,   Yes)  => Ordering::Less,
+            (_,  No) | (Yes,   _)  => Ordering::Greater,
+            (Until(t1), Until(t2)) => t1.cmp(t2),
+        }
+    }
+}

swarm/src/protocols_handler/node_handler.rs

@@ -0,0 +1,325 @@
+// 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::protocols_handler::{
+    KeepAlive,
+    ProtocolsHandler,
+    IntoProtocolsHandler,
+    ProtocolsHandlerEvent,
+    ProtocolsHandlerUpgrErr
+};
+use futures::prelude::*;
+use libp2p_core::{
+    ConnectedPoint,
+    PeerId,
+    nodes::collection::ConnectionInfo,
+    nodes::handled_node::{NodeHandler, NodeHandlerEndpoint, NodeHandlerEvent},
+    nodes::handled_node_tasks::IntoNodeHandler,
+    upgrade::{self, InboundUpgradeApply, OutboundUpgradeApply}
+};
+use std::{error, fmt, time::Duration};
+use wasm_timer::{Delay, Timeout};
+
+/// Prototype for a `NodeHandlerWrapper`.
+pub struct NodeHandlerWrapperBuilder<TIntoProtoHandler> {
+    /// The underlying handler.
+    handler: TIntoProtoHandler,
+}
+
+impl<TIntoProtoHandler> NodeHandlerWrapperBuilder<TIntoProtoHandler>
+where
+    TIntoProtoHandler: IntoProtocolsHandler
+{
+    /// Builds a `NodeHandlerWrapperBuilder`.
+    #[inline]
+    pub(crate) fn new(handler: TIntoProtoHandler) -> Self {
+        NodeHandlerWrapperBuilder {
+            handler,
+        }
+    }
+
+    /// Builds the `NodeHandlerWrapper`.
+    #[deprecated(note = "Pass the NodeHandlerWrapperBuilder directly")]
+    #[inline]
+    pub fn build(self) -> NodeHandlerWrapper<TIntoProtoHandler>
+    where TIntoProtoHandler: ProtocolsHandler
+    {
+        NodeHandlerWrapper {
+            handler: self.handler,
+            negotiating_in: Vec::new(),
+            negotiating_out: Vec::new(),
+            queued_dial_upgrades: Vec::new(),
+            unique_dial_upgrade_id: 0,
+            shutdown: Shutdown::None,
+        }
+    }
+}
+
+impl<TIntoProtoHandler, TProtoHandler, TConnInfo> IntoNodeHandler<(TConnInfo, ConnectedPoint)>
+    for NodeHandlerWrapperBuilder<TIntoProtoHandler>
+where
+    TIntoProtoHandler: IntoProtocolsHandler<Handler = TProtoHandler>,
+    TProtoHandler: ProtocolsHandler,
+    TConnInfo: ConnectionInfo<PeerId = PeerId>,
+{
+    type Handler = NodeHandlerWrapper<TIntoProtoHandler::Handler>;
+
+    fn into_handler(self, remote_info: &(TConnInfo, ConnectedPoint)) -> Self::Handler {
+        NodeHandlerWrapper {
+            handler: self.handler.into_handler(&remote_info.0.peer_id(), &remote_info.1),
+            negotiating_in: Vec::new(),
+            negotiating_out: Vec::new(),
+            queued_dial_upgrades: Vec::new(),
+            unique_dial_upgrade_id: 0,
+            shutdown: Shutdown::None,
+        }
+    }
+}
+
+/// Wraps around an implementation of `ProtocolsHandler`, and implements `NodeHandler`.
+// TODO: add a caching system for protocols that are supported or not
+pub struct NodeHandlerWrapper<TProtoHandler>
+where
+    TProtoHandler: ProtocolsHandler,
+{
+    /// The underlying handler.
+    handler: TProtoHandler,
+    /// Futures that upgrade incoming substreams.
+    negotiating_in:
+        Vec<Timeout<InboundUpgradeApply<TProtoHandler::Substream, TProtoHandler::InboundProtocol>>>,
+    /// Futures that upgrade outgoing substreams. The first element of the tuple is the userdata
+    /// to pass back once successfully opened.
+    negotiating_out: Vec<(
+        TProtoHandler::OutboundOpenInfo,
+        Timeout<OutboundUpgradeApply<TProtoHandler::Substream, TProtoHandler::OutboundProtocol>>,
+    )>,
+    /// For each outbound substream request, how to upgrade it. The first element of the tuple
+    /// is the unique identifier (see `unique_dial_upgrade_id`).
+    queued_dial_upgrades: Vec<(u64, TProtoHandler::OutboundProtocol)>,
+    /// Unique identifier assigned to each queued dial upgrade.
+    unique_dial_upgrade_id: u64,
+    /// The currently planned connection & handler shutdown.
+    shutdown: Shutdown,
+}
+
+/// The options for a planned connection & handler shutdown.
+///
+/// A shutdown is planned anew based on the the return value of
+/// [`ProtocolsHandler::connection_keep_alive`] of the underlying handler
+/// after every invocation of [`ProtocolsHandler::poll`].
+///
+/// A planned shutdown is always postponed for as long as there are ingoing
+/// or outgoing substreams being negotiated, i.e. it is a graceful, "idle"
+/// shutdown.
+enum Shutdown {
+    /// No shutdown is planned.
+    None,
+    /// A shut down is planned as soon as possible.
+    Asap,
+    /// A shut down is planned for when a `Delay` has elapsed.
+    Later(Delay)
+}
+
+/// Error generated by the `NodeHandlerWrapper`.
+#[derive(Debug)]
+pub enum NodeHandlerWrapperError<TErr> {
+    /// Error generated by the handler.
+    Handler(TErr),
+    /// The connection has been deemed useless and has been closed.
+    UselessTimeout,
+}
+
+impl<TErr> From<TErr> for NodeHandlerWrapperError<TErr> {
+    fn from(err: TErr) -> NodeHandlerWrapperError<TErr> {
+        NodeHandlerWrapperError::Handler(err)
+    }
+}
+
+impl<TErr> fmt::Display for NodeHandlerWrapperError<TErr>
+where
+    TErr: fmt::Display
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            NodeHandlerWrapperError::Handler(err) => write!(f, "{}", err),
+            NodeHandlerWrapperError::UselessTimeout =>
+                write!(f, "Node has been closed due to inactivity"),
+        }
+    }
+}
+
+impl<TErr> error::Error for NodeHandlerWrapperError<TErr>
+where
+    TErr: error::Error + 'static
+{
+    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
+        match self {
+            NodeHandlerWrapperError::Handler(err) => Some(err),
+            NodeHandlerWrapperError::UselessTimeout => None,
+        }
+    }
+}
+
+impl<TProtoHandler> NodeHandler for NodeHandlerWrapper<TProtoHandler>
+where
+    TProtoHandler: ProtocolsHandler,
+{
+    type InEvent = TProtoHandler::InEvent;
+    type OutEvent = TProtoHandler::OutEvent;
+    type Error = NodeHandlerWrapperError<TProtoHandler::Error>;
+    type Substream = TProtoHandler::Substream;
+    // The first element of the tuple is the unique upgrade identifier
+    // (see `unique_dial_upgrade_id`).
+    type OutboundOpenInfo = (u64, TProtoHandler::OutboundOpenInfo, Duration);
+
+    fn inject_substream(
+        &mut self,
+        substream: Self::Substream,
+        endpoint: NodeHandlerEndpoint<Self::OutboundOpenInfo>,
+    ) {
+        match endpoint {
+            NodeHandlerEndpoint::Listener => {
+                let protocol = self.handler.listen_protocol();
+                let timeout = protocol.timeout().clone();
+                let upgrade = upgrade::apply_inbound(substream, protocol.into_upgrade());
+                let with_timeout = Timeout::new(upgrade, timeout);
+                self.negotiating_in.push(with_timeout);
+            }
+            NodeHandlerEndpoint::Dialer((upgrade_id, user_data, timeout)) => {
+                let pos = match self
+                    .queued_dial_upgrades
+                    .iter()
+                    .position(|(id, _)| id == &upgrade_id)
+                {
+                    Some(p) => p,
+                    None => {
+                        debug_assert!(false, "Received an upgrade with an invalid upgrade ID");
+                        return;
+                    }
+                };
+
+                let (_, proto_upgrade) = self.queued_dial_upgrades.remove(pos);
+                let upgrade = upgrade::apply_outbound(substream, proto_upgrade);
+                let with_timeout = Timeout::new(upgrade, timeout);
+                self.negotiating_out.push((user_data, with_timeout));
+            }
+        }
+    }
+
+    #[inline]
+    fn inject_event(&mut self, event: Self::InEvent) {
+        self.handler.inject_event(event);
+    }
+
+    fn poll(&mut self) -> Poll<NodeHandlerEvent<Self::OutboundOpenInfo, Self::OutEvent>, Self::Error> {
+        // 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 = self.negotiating_in.swap_remove(n);
+            match in_progress.poll() {
+                Ok(Async::Ready(upgrade)) =>
+                    self.handler.inject_fully_negotiated_inbound(upgrade),
+                Ok(Async::NotReady) => self.negotiating_in.push(in_progress),
+                // TODO: return a diagnostic event?
+                Err(_err) => {}
+            }
+        }
+
+        // Continue negotiation of newly-opened substreams.
+        // We remove each element from `negotiating_out` one by one and add them back if not ready.
+        for n in (0..self.negotiating_out.len()).rev() {
+            let (upgr_info, mut in_progress) = self.negotiating_out.swap_remove(n);
+            match in_progress.poll() {
+                Ok(Async::Ready(upgrade)) => {
+                    self.handler.inject_fully_negotiated_outbound(upgrade, upgr_info);
+                }
+                Ok(Async::NotReady) => {
+                    self.negotiating_out.push((upgr_info, in_progress));
+                }
+                Err(err) => {
+                    let err = if err.is_elapsed() {
+                        ProtocolsHandlerUpgrErr::Timeout
+                    } else if err.is_timer() {
+                        ProtocolsHandlerUpgrErr::Timer
+                    } else {
+                        debug_assert!(err.is_inner());
+                        let err = err.into_inner().expect("Timeout error is one of {elapsed, \
+                            timer, inner}; is_elapsed and is_timer are both false; error is \
+                            inner; QED");
+                        ProtocolsHandlerUpgrErr::Upgrade(err)
+                    };
+
+                    self.handler.inject_dial_upgrade_error(upgr_info, err);
+                }
+            }
+        }
+
+        // Poll the handler at the end so that we see the consequences of the method
+        // calls on `self.handler`.
+        let poll_result = self.handler.poll()?;
+
+        // Ask the handler whether it wants the connection (and the handler itself)
+        // to be kept alive, which determines the planned shutdown, if any.
+        match (&mut self.shutdown, self.handler.connection_keep_alive()) {
+            (Shutdown::Later(d), KeepAlive::Until(t)) =>
+                if d.deadline() != t {
+                    d.reset(t)
+                },
+            (_, KeepAlive::Until(t)) => self.shutdown = Shutdown::Later(Delay::new(t)),
+            (_, KeepAlive::No) => self.shutdown = Shutdown::Asap,
+            (_, KeepAlive::Yes) => self.shutdown = Shutdown::None
+        };
+
+        match poll_result {
+            Async::Ready(ProtocolsHandlerEvent::Custom(event)) => {
+                return Ok(Async::Ready(NodeHandlerEvent::Custom(event)));
+            }
+            Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                protocol,
+                info,
+            }) => {
+                let id = self.unique_dial_upgrade_id;
+                let timeout = protocol.timeout().clone();
+                self.unique_dial_upgrade_id += 1;
+                self.queued_dial_upgrades.push((id, protocol.into_upgrade()));
+                return Ok(Async::Ready(
+                    NodeHandlerEvent::OutboundSubstreamRequest((id, info, timeout)),
+                ));
+            }
+            Async::NotReady => (),
+        };
+
+        // Check if the connection (and handler) should be shut down.
+        // As long as we're still negotiating substreams, shutdown is always postponed.
+        if self.negotiating_in.is_empty() && self.negotiating_out.is_empty() {
+            match self.shutdown {
+                Shutdown::None => {},
+                Shutdown::Asap => return Err(NodeHandlerWrapperError::UselessTimeout),
+                Shutdown::Later(ref mut delay) => match delay.poll() {
+                    Ok(Async::Ready(_)) | Err(_) =>
+                        return Err(NodeHandlerWrapperError::UselessTimeout),
+                    Ok(Async::NotReady) => {}
+                }
+            }
+        }
+
+        Ok(Async::NotReady)
+    }
+}

swarm/src/protocols_handler/one_shot.rs

@@ -0,0 +1,243 @@
+// Copyright 2019 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::protocols_handler::{
+    KeepAlive,
+    ProtocolsHandler,
+    ProtocolsHandlerEvent,
+    ProtocolsHandlerUpgrErr,
+    SubstreamProtocol
+};
+use futures::prelude::*;
+use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade};
+use smallvec::SmallVec;
+use std::{error, marker::PhantomData, time::Duration};
+use tokio_io::{AsyncRead, AsyncWrite};
+use wasm_timer::Instant;
+
+/// Implementation of `ProtocolsHandler` that opens a new substream for each individual message.
+///
+/// This struct is meant to be a helper for other implementations to use.
+// TODO: Debug
+pub struct OneShotHandler<TSubstream, TInProto, TOutProto, TOutEvent>
+where
+    TOutProto: OutboundUpgrade<TSubstream>,
+{
+    /// The upgrade for inbound substreams.
+    listen_protocol: SubstreamProtocol<TInProto>,
+    /// If `Some`, something bad happened and we should shut down the handler with an error.
+    pending_error:
+        Option<ProtocolsHandlerUpgrErr<<TOutProto as OutboundUpgrade<TSubstream>>::Error>>,
+    /// Queue of events to produce in `poll()`.
+    events_out: SmallVec<[TOutEvent; 4]>,
+    /// Queue of outbound substreams to open.
+    dial_queue: SmallVec<[TOutProto; 4]>,
+    /// Current number of concurrent outbound substreams being opened.
+    dial_negotiated: u32,
+    /// Maximum number of concurrent outbound substreams being opened. Value is never modified.
+    max_dial_negotiated: u32,
+    /// Value to return from `connection_keep_alive`.
+    keep_alive: KeepAlive,
+    /// After the given duration has elapsed, an inactive connection will shutdown.
+    inactive_timeout: Duration,
+    /// Pin the `TSubstream` generic.
+    marker: PhantomData<TSubstream>,
+}
+
+impl<TSubstream, TInProto, TOutProto, TOutEvent>
+    OneShotHandler<TSubstream, TInProto, TOutProto, TOutEvent>
+where
+    TOutProto: OutboundUpgrade<TSubstream>,
+{
+    /// Creates a `OneShotHandler`.
+    #[inline]
+    pub fn new(
+        listen_protocol: SubstreamProtocol<TInProto>,
+        inactive_timeout: Duration
+    ) -> Self {
+        OneShotHandler {
+            listen_protocol,
+            pending_error: None,
+            events_out: SmallVec::new(),
+            dial_queue: SmallVec::new(),
+            dial_negotiated: 0,
+            max_dial_negotiated: 8,
+            keep_alive: KeepAlive::Yes,
+            inactive_timeout,
+            marker: PhantomData,
+        }
+    }
+
+    /// Returns the number of pending requests.
+    #[inline]
+    pub fn pending_requests(&self) -> u32 {
+        self.dial_negotiated + self.dial_queue.len() as u32
+    }
+
+    /// Returns a reference to the listen protocol configuration.
+    ///
+    /// > **Note**: If you modify the protocol, modifications will only applies to future inbound
+    /// >           substreams, not the ones already being negotiated.
+    #[inline]
+    pub fn listen_protocol_ref(&self) -> &SubstreamProtocol<TInProto> {
+        &self.listen_protocol
+    }
+
+    /// Returns a mutable reference to the listen protocol configuration.
+    ///
+    /// > **Note**: If you modify the protocol, modifications will only applies to future inbound
+    /// >           substreams, not the ones already being negotiated.
+    #[inline]
+    pub fn listen_protocol_mut(&mut self) -> &mut SubstreamProtocol<TInProto> {
+        &mut self.listen_protocol
+    }
+
+    /// Opens an outbound substream with `upgrade`.
+    #[inline]
+    pub fn send_request(&mut self, upgrade: TOutProto) {
+        self.keep_alive = KeepAlive::Yes;
+        self.dial_queue.push(upgrade);
+    }
+}
+
+impl<TSubstream, TInProto, TOutProto, TOutEvent> Default
+    for OneShotHandler<TSubstream, TInProto, TOutProto, TOutEvent>
+where
+    TOutProto: OutboundUpgrade<TSubstream>,
+    TInProto: InboundUpgrade<TSubstream> + Default,
+{
+    #[inline]
+    fn default() -> Self {
+        OneShotHandler::new(SubstreamProtocol::new(Default::default()), Duration::from_secs(10))
+    }
+}
+
+impl<TSubstream, TInProto, TOutProto, TOutEvent> ProtocolsHandler
+    for OneShotHandler<TSubstream, TInProto, TOutProto, TOutEvent>
+where
+    TSubstream: AsyncRead + AsyncWrite,
+    TInProto: InboundUpgrade<TSubstream>,
+    TOutProto: OutboundUpgrade<TSubstream>,
+    TInProto::Output: Into<TOutEvent>,
+    TOutProto::Output: Into<TOutEvent>,
+    TOutProto::Error: error::Error + 'static,
+    SubstreamProtocol<TInProto>: Clone,
+{
+    type InEvent = TOutProto;
+    type OutEvent = TOutEvent;
+    type Error = ProtocolsHandlerUpgrErr<
+        <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error,
+    >;
+    type Substream = TSubstream;
+    type InboundProtocol = TInProto;
+    type OutboundProtocol = TOutProto;
+    type OutboundOpenInfo = ();
+
+    #[inline]
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+        self.listen_protocol.clone()
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_inbound(
+        &mut self,
+        out: <Self::InboundProtocol as InboundUpgrade<Self::Substream>>::Output,
+    ) {
+        // If we're shutting down the connection for inactivity, reset the timeout.
+        if !self.keep_alive.is_yes() {
+            self.keep_alive = KeepAlive::Until(Instant::now() + self.inactive_timeout);
+        }
+
+        self.events_out.push(out.into());
+    }
+
+    #[inline]
+    fn inject_fully_negotiated_outbound(
+        &mut self,
+        out: <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Output,
+        _: Self::OutboundOpenInfo,
+    ) {
+        self.dial_negotiated -= 1;
+
+        if self.dial_negotiated == 0 && self.dial_queue.is_empty() {
+            self.keep_alive = KeepAlive::Until(Instant::now() + self.inactive_timeout);
+        }
+
+        self.events_out.push(out.into());
+    }
+
+    #[inline]
+    fn inject_event(&mut self, event: Self::InEvent) {
+        self.send_request(event);
+    }
+
+    #[inline]
+    fn inject_dial_upgrade_error(
+        &mut self,
+        _: Self::OutboundOpenInfo,
+        error: ProtocolsHandlerUpgrErr<
+            <Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error,
+        >,
+    ) {
+        if self.pending_error.is_none() {
+            self.pending_error = Some(error);
+        }
+    }
+
+    #[inline]
+    fn connection_keep_alive(&self) -> KeepAlive {
+        self.keep_alive
+    }
+
+    fn poll(
+        &mut self,
+    ) -> Poll<
+        ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent>,
+        Self::Error,
+    > {
+        if let Some(err) = self.pending_error.take() {
+            return Err(err);
+        }
+
+        if !self.events_out.is_empty() {
+            return Ok(Async::Ready(ProtocolsHandlerEvent::Custom(
+                self.events_out.remove(0),
+            )));
+        } else {
+            self.events_out.shrink_to_fit();
+        }
+
+        if !self.dial_queue.is_empty() {
+            if self.dial_negotiated < self.max_dial_negotiated {
+                self.dial_negotiated += 1;
+                return Ok(Async::Ready(
+                    ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                        protocol: SubstreamProtocol::new(self.dial_queue.remove(0)),
+                        info: (),
+                    },
+                ));
+            }
+        } else {
+            self.dial_queue.shrink_to_fit();
+        }
+
+        Ok(Async::NotReady)
+    }
+}

swarm/src/protocols_handler/select.rs

@@ -0,0 +1,240 @@
+// Copyright 2019 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::protocols_handler::{
+    KeepAlive,
+    SubstreamProtocol,
+    IntoProtocolsHandler,
+    ProtocolsHandler,
+    ProtocolsHandlerEvent,
+    ProtocolsHandlerUpgrErr,
+};
+use futures::prelude::*;
+use libp2p_core::{
+    ConnectedPoint,
+    PeerId,
+    either::{EitherError, EitherOutput},
+    upgrade::{InboundUpgrade, OutboundUpgrade, EitherUpgrade, SelectUpgrade, UpgradeError}
+};
+use std::cmp;
+use tokio_io::{AsyncRead, AsyncWrite};
+
+/// Implementation of `IntoProtocolsHandler` that combines two protocols into one.
+#[derive(Debug, Clone)]
+pub struct IntoProtocolsHandlerSelect<TProto1, TProto2> {
+    /// The first protocol.
+    proto1: TProto1,
+    /// The second protocol.
+    proto2: TProto2,
+}
+
+impl<TProto1, TProto2> IntoProtocolsHandlerSelect<TProto1, TProto2> {
+    /// Builds a `IntoProtocolsHandlerSelect`.
+    #[inline]
+    pub(crate) fn new(proto1: TProto1, proto2: TProto2) -> Self {
+        IntoProtocolsHandlerSelect {
+            proto1,
+            proto2,
+        }
+    }
+}
+
+impl<TProto1, TProto2, TSubstream> IntoProtocolsHandler for IntoProtocolsHandlerSelect<TProto1, TProto2>
+where
+    TProto1: IntoProtocolsHandler,
+    TProto2: IntoProtocolsHandler,
+    TProto1::Handler: ProtocolsHandler<Substream = TSubstream>,
+    TProto2::Handler: ProtocolsHandler<Substream = TSubstream>,
+    TSubstream: AsyncRead + AsyncWrite,
+    <TProto1::Handler as ProtocolsHandler>::InboundProtocol: InboundUpgrade<TSubstream>,
+    <TProto2::Handler as ProtocolsHandler>::InboundProtocol: InboundUpgrade<TSubstream>,
+    <TProto1::Handler as ProtocolsHandler>::OutboundProtocol: OutboundUpgrade<TSubstream>,
+    <TProto2::Handler as ProtocolsHandler>::OutboundProtocol: OutboundUpgrade<TSubstream>
+{
+    type Handler = ProtocolsHandlerSelect<TProto1::Handler, TProto2::Handler>;
+
+    fn into_handler(self, remote_peer_id: &PeerId, connected_point: &ConnectedPoint) -> Self::Handler {
+        ProtocolsHandlerSelect {
+            proto1: self.proto1.into_handler(remote_peer_id, connected_point),
+            proto2: self.proto2.into_handler(remote_peer_id, connected_point),
+        }
+    }
+
+    fn inbound_protocol(&self) -> <Self::Handler as ProtocolsHandler>::InboundProtocol {
+        SelectUpgrade::new(self.proto1.inbound_protocol(), self.proto2.inbound_protocol())
+    }
+}
+
+/// Implementation of `ProtocolsHandler` that combines two protocols into one.
+#[derive(Debug, Clone)]
+pub struct ProtocolsHandlerSelect<TProto1, TProto2> {
+    /// The first protocol.
+    proto1: TProto1,
+    /// The second protocol.
+    proto2: TProto2,
+}
+
+impl<TProto1, TProto2> ProtocolsHandlerSelect<TProto1, TProto2> {
+    /// Builds a `ProtocolsHandlerSelect`.
+    #[inline]
+    pub(crate) fn new(proto1: TProto1, proto2: TProto2) -> Self {
+        ProtocolsHandlerSelect {
+            proto1,
+            proto2,
+        }
+    }
+}
+
+impl<TSubstream, TProto1, TProto2>
+    ProtocolsHandler for ProtocolsHandlerSelect<TProto1, TProto2>
+where
+    TProto1: ProtocolsHandler<Substream = TSubstream>,
+    TProto2: ProtocolsHandler<Substream = TSubstream>,
+    TSubstream: AsyncRead + AsyncWrite,
+    TProto1::InboundProtocol: InboundUpgrade<TSubstream>,
+    TProto2::InboundProtocol: InboundUpgrade<TSubstream>,
+    TProto1::OutboundProtocol: OutboundUpgrade<TSubstream>,
+    TProto2::OutboundProtocol: OutboundUpgrade<TSubstream>
+{
+    type InEvent = EitherOutput<TProto1::InEvent, TProto2::InEvent>;
+    type OutEvent = EitherOutput<TProto1::OutEvent, TProto2::OutEvent>;
+    type Error = EitherError<TProto1::Error, TProto2::Error>;
+    type Substream = TSubstream;
+    type InboundProtocol = SelectUpgrade<<TProto1 as ProtocolsHandler>::InboundProtocol, <TProto2 as ProtocolsHandler>::InboundProtocol>;
+    type OutboundProtocol = EitherUpgrade<TProto1::OutboundProtocol, TProto2::OutboundProtocol>;
+    type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>;
+
+    #[inline]
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
+        let proto1 = self.proto1.listen_protocol();
+        let proto2 = self.proto2.listen_protocol();
+        let timeout = std::cmp::max(proto1.timeout(), proto2.timeout()).clone();
+        SubstreamProtocol::new(SelectUpgrade::new(proto1.into_upgrade(), proto2.into_upgrade()))
+            .with_timeout(timeout)
+    }
+
+    fn inject_fully_negotiated_outbound(&mut self, protocol: <Self::OutboundProtocol as OutboundUpgrade<TSubstream>>::Output, endpoint: Self::OutboundOpenInfo) {
+        match (protocol, endpoint) {
+            (EitherOutput::First(protocol), EitherOutput::First(info)) =>
+                self.proto1.inject_fully_negotiated_outbound(protocol, info),
+            (EitherOutput::Second(protocol), EitherOutput::Second(info)) =>
+                self.proto2.inject_fully_negotiated_outbound(protocol, info),
+            (EitherOutput::First(_), EitherOutput::Second(_)) =>
+                panic!("wrong API usage: the protocol doesn't match the upgrade info"),
+            (EitherOutput::Second(_), EitherOutput::First(_)) =>
+                panic!("wrong API usage: the protocol doesn't match the upgrade info")
+        }
+    }
+
+    fn inject_fully_negotiated_inbound(&mut self, protocol: <Self::InboundProtocol as InboundUpgrade<TSubstream>>::Output) {
+        match protocol {
+            EitherOutput::First(protocol) =>
+                self.proto1.inject_fully_negotiated_inbound(protocol),
+            EitherOutput::Second(protocol) =>
+                self.proto2.inject_fully_negotiated_inbound(protocol)
+        }
+    }
+
+    #[inline]
+    fn inject_event(&mut self, event: Self::InEvent) {
+        match event {
+            EitherOutput::First(event) => self.proto1.inject_event(event),
+            EitherOutput::Second(event) => self.proto2.inject_event(event),
+        }
+    }
+
+    #[inline]
+    fn inject_dial_upgrade_error(&mut self, info: Self::OutboundOpenInfo, error: ProtocolsHandlerUpgrErr<<Self::OutboundProtocol as OutboundUpgrade<Self::Substream>>::Error>) {
+        match (info, error) {
+            (EitherOutput::First(info), ProtocolsHandlerUpgrErr::Timer) => {
+                self.proto1.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Timer)
+            },
+            (EitherOutput::First(info), ProtocolsHandlerUpgrErr::Timeout) => {
+                self.proto1.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Timeout)
+            },
+            (EitherOutput::First(info), ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(err))) => {
+                self.proto1.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(err)))
+            },
+            (EitherOutput::First(info), ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(err)))) => {
+                self.proto1.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(err)))
+            },
+            (EitherOutput::First(_), ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(_)))) => {
+                panic!("Wrong API usage; the upgrade error doesn't match the outbound open info");
+            },
+            (EitherOutput::Second(info), ProtocolsHandlerUpgrErr::Timeout) => {
+                self.proto2.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Timeout)
+            },
+            (EitherOutput::Second(info), ProtocolsHandlerUpgrErr::Timer) => {
+                self.proto2.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Timer)
+            },
+            (EitherOutput::Second(info), ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(err))) => {
+                self.proto2.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Select(err)))
+            },
+            (EitherOutput::Second(info), ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::B(err)))) => {
+                self.proto2.inject_dial_upgrade_error(info, ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(err)))
+            },
+            (EitherOutput::Second(_), ProtocolsHandlerUpgrErr::Upgrade(UpgradeError::Apply(EitherError::A(_)))) => {
+                panic!("Wrong API usage; the upgrade error doesn't match the outbound open info");
+            },
+        }
+    }
+
+    #[inline]
+    fn connection_keep_alive(&self) -> KeepAlive {
+        cmp::max(self.proto1.connection_keep_alive(), self.proto2.connection_keep_alive())
+    }
+
+    fn poll(&mut self) -> Poll<ProtocolsHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::OutEvent>, Self::Error> {
+
+        match self.proto1.poll().map_err(EitherError::A)? {
+            Async::Ready(ProtocolsHandlerEvent::Custom(event)) => {
+                return Ok(Async::Ready(ProtocolsHandlerEvent::Custom(EitherOutput::First(event))));
+            },
+            Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                protocol,
+                info,
+            }) => {
+                return Ok(Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                    protocol: protocol.map_upgrade(EitherUpgrade::A),
+                    info: EitherOutput::First(info),
+                }));
+            },
+            Async::NotReady => ()
+        };
+
+        match self.proto2.poll().map_err(EitherError::B)? {
+            Async::Ready(ProtocolsHandlerEvent::Custom(event)) => {
+                return Ok(Async::Ready(ProtocolsHandlerEvent::Custom(EitherOutput::Second(event))));
+            },
+            Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                protocol,
+                info,
+            }) => {
+                return Ok(Async::Ready(ProtocolsHandlerEvent::OutboundSubstreamRequest {
+                    protocol: protocol.map_upgrade(EitherUpgrade::B),
+                    info: EitherOutput::Second(info),
+                }));
+            },
+            Async::NotReady => ()
+        };
+
+        Ok(Async::NotReady)
+    }
+}