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

New core (#568)

Browse Source 
* New core

* Fix lifetime requirements

* Remove identify transport

* Address &mut & ref ref mut

* Fix whitespaces
This commit is contained in:
Pierre Krieger
2018-10-17 10:17:40 +01:00
committed by GitHub
parent 724d0f5d82
commit 5d1c54cc10
52 changed files with 849 additions and 1492 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
core/src/either.rs
View File

@ -18,10 +18,11 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use futures::{prelude::*, future};
use futures::prelude::*;
use muxing::{Shutdown, StreamMuxer};
use std::io::{Error as IoError, Read, Write};
use tokio_io::{AsyncRead, AsyncWrite};
use Multiaddr;
/// Implements `AsyncRead` and `AsyncWrite` and dispatches all method calls to
/// either `First` or `Second`.
@ -39,8 +40,8 @@ where
#[inline]
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
match self {
&EitherOutput::First(ref a) => a.prepare_uninitialized_buffer(buf),
&EitherOutput::Second(ref b) => b.prepare_uninitialized_buffer(buf),
EitherOutput::First(a) => a.prepare_uninitialized_buffer(buf),
EitherOutput::Second(b) => b.prepare_uninitialized_buffer(buf),
}
}
}
@ -53,8 +54,8 @@ where
#[inline]
fn read(&mut self, buf: &mut [u8]) -> Result<usize, IoError> {
match self {
&mut EitherOutput::First(ref mut a) => a.read(buf),
&mut EitherOutput::Second(ref mut b) => b.read(buf),
EitherOutput::First(a) => a.read(buf),
EitherOutput::Second(b) => b.read(buf),
}
}
}
@ -67,8 +68,8 @@ where
#[inline]
fn shutdown(&mut self) -> Poll<(), IoError> {
match self {
&mut EitherOutput::First(ref mut a) => a.shutdown(),
&mut EitherOutput::Second(ref mut b) => b.shutdown(),
EitherOutput::First(a) => a.shutdown(),
EitherOutput::Second(b) => b.shutdown(),
}
}
}
@ -81,16 +82,16 @@ where
#[inline]
fn write(&mut self, buf: &[u8]) -> Result<usize, IoError> {
match self {
&mut EitherOutput::First(ref mut a) => a.write(buf),
&mut EitherOutput::Second(ref mut b) => b.write(buf),
EitherOutput::First(a) => a.write(buf),
EitherOutput::Second(b) => b.write(buf),
}
}
#[inline]
fn flush(&mut self) -> Result<(), IoError> {
match self {
&mut EitherOutput::First(ref mut a) => a.flush(),
&mut EitherOutput::Second(ref mut b) => b.flush(),
EitherOutput::First(a) => a.flush(),
EitherOutput::Second(b) => b.flush(),
}
}
}
@ -104,16 +105,16 @@ where
type OutboundSubstream = EitherOutbound<A, B>;
fn poll_inbound(&self) -> Poll<Option<Self::Substream>, IoError> {
match *self {
EitherOutput::First(ref inner) => inner.poll_inbound().map(|p| p.map(|o| o.map(EitherOutput::First))),
EitherOutput::Second(ref inner) => inner.poll_inbound().map(|p| p.map(|o| o.map(EitherOutput::Second))),
match self {
EitherOutput::First(inner) => inner.poll_inbound().map(|p| p.map(|o| o.map(EitherOutput::First))),
EitherOutput::Second(inner) => inner.poll_inbound().map(|p| p.map(|o| o.map(EitherOutput::Second))),
}
}
fn open_outbound(&self) -> Self::OutboundSubstream {
match *self {
EitherOutput::First(ref inner) => EitherOutbound::A(inner.open_outbound()),
EitherOutput::Second(ref inner) => EitherOutbound::B(inner.open_outbound()),
match self {
EitherOutput::First(inner) => EitherOutbound::A(inner.open_outbound()),
EitherOutput::Second(inner) => EitherOutbound::B(inner.open_outbound()),
}
}
@ -130,14 +131,14 @@ where
}
fn destroy_outbound(&self, substream: Self::OutboundSubstream) {
match *self {
EitherOutput::First(ref inner) => {
match self {
EitherOutput::First(inner) => {
match substream {
EitherOutbound::A(substream) => inner.destroy_outbound(substream),
_ => panic!("Wrong API usage")
}
},
EitherOutput::Second(ref inner) => {
EitherOutput::Second(inner) => {
match substream {
EitherOutbound::B(substream) => inner.destroy_outbound(substream),
_ => panic!("Wrong API usage")
@ -195,14 +196,14 @@ where
}
fn destroy_substream(&self, substream: Self::Substream) {
match *self {
EitherOutput::First(ref inner) => {
match self {
EitherOutput::First(inner) => {
match substream {
EitherOutput::First(substream) => inner.destroy_substream(substream),
_ => panic!("Wrong API usage")
}
},
EitherOutput::Second(ref inner) => {
EitherOutput::Second(inner) => {
match substream {
EitherOutput::Second(substream) => inner.destroy_substream(substream),
_ => panic!("Wrong API usage")
@ -212,16 +213,16 @@ where
}
fn shutdown(&self, kind: Shutdown) -> Poll<(), IoError> {
match *self {
EitherOutput::First(ref inner) => inner.shutdown(kind),
EitherOutput::Second(ref inner) => inner.shutdown(kind)
match self {
EitherOutput::First(inner) => inner.shutdown(kind),
EitherOutput::Second(inner) => inner.shutdown(kind)
}
}
fn flush_all(&self) -> Poll<(), IoError> {
match *self {
EitherOutput::First(ref inner) => inner.flush_all(),
EitherOutput::Second(ref inner) => inner.flush_all()
match self {
EitherOutput::First(inner) => inner.flush_all(),
EitherOutput::Second(inner) => inner.flush_all()
}
}
}
@ -243,52 +244,44 @@ pub enum EitherListenStream<A, B> {
impl<AStream, BStream, AInner, BInner> Stream for EitherListenStream<AStream, BStream>
where
AStream: Stream<Item = AInner, Error = IoError>,
BStream: Stream<Item = BInner, Error = IoError>,
AStream: Stream<Item = (AInner, Multiaddr), Error = IoError>,
BStream: Stream<Item = (BInner, Multiaddr), Error = IoError>,
{
type Item = EitherListenUpgrade<AInner, BInner>;
type Item = (EitherFuture<AInner, BInner>, Multiaddr);
type Error = IoError;
#[inline]
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
match self {
&mut EitherListenStream::First(ref mut a) => a.poll()
.map(|i| i.map(|v| v.map(EitherListenUpgrade::First))),
&mut EitherListenStream::Second(ref mut a) => a.poll()
.map(|i| i.map(|v| v.map(EitherListenUpgrade::Second))),
EitherListenStream::First(a) => a.poll()
.map(|i| (i.map(|v| (v.map(|(o, addr)| (EitherFuture::First(o), addr)))))),
EitherListenStream::Second(a) => a.poll()
.map(|i| (i.map(|v| (v.map(|(o, addr)| (EitherFuture::Second(o), addr)))))),
}
}
}
// TODO: This type is needed because of the lack of `impl Trait` in stable Rust.
// If Rust had impl Trait we could use the Either enum from the futures crate and add some
// modifiers to it. This custom enum is a combination of Either and these modifiers.
/// Implements `Future` and dispatches all method calls to either `First` or `Second`.
#[derive(Debug, Copy, Clone)]
#[must_use = "futures do nothing unless polled"]
pub enum EitherListenUpgrade<A, B> {
pub enum EitherFuture<A, B> {
First(A),
Second(B),
}
impl<A, B, Ao, Bo, Af, Bf> Future for EitherListenUpgrade<A, B>
impl<AFuture, BFuture, AInner, BInner> Future for EitherFuture<AFuture, BFuture>
where
A: Future<Item = (Ao, Af), Error = IoError>,
B: Future<Item = (Bo, Bf), Error = IoError>,
AFuture: Future<Item = AInner, Error = IoError>,
BFuture: Future<Item = BInner, Error = IoError>,
{
type Item = (EitherOutput<Ao, Bo>, future::Either<Af, Bf>);
type Item = EitherOutput<AInner, BInner>;
type Error = IoError;
#[inline]
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self {
&mut EitherListenUpgrade::First(ref mut a) => {
let (item, addr) = try_ready!(a.poll());
Ok(Async::Ready((EitherOutput::First(item), future::Either::A(addr))))
}
&mut EitherListenUpgrade::Second(ref mut b) => {
let (item, addr) = try_ready!(b.poll());
Ok(Async::Ready((EitherOutput::Second(item), future::Either::B(addr))))
}
EitherFuture::First(a) => a.poll().map(|v| v.map(EitherOutput::First)),
EitherFuture::Second(a) => a.poll().map(|v| v.map(EitherOutput::Second)),
}
}
}

4
core/src/lib.rs
View File

@ -142,7 +142,7 @@
//! // TODO: right now the only available protocol is ping, but we want to replace it with
//! // something that is more simple to use
//! .dial("127.0.0.1:12345".parse::<libp2p_core::Multiaddr>().unwrap()).unwrap_or_else(|_| panic!())
//! .and_then(|(out, _)| {
//! .and_then(|out| {
//! match out {
//! PingOutput::Ponger(processing) => Box::new(processing) as Box<Future<Item = _, Error = _>>,
//! PingOutput::Pinger(mut pinger) => {
@ -220,5 +220,5 @@ pub use self::multiaddr::Multiaddr;
pub use self::muxing::StreamMuxer;
pub use self::peer_id::PeerId;
pub use self::public_key::PublicKey;
pub use self::transport::{MuxedTransport, Transport};
pub use self::transport::Transport;
pub use self::upgrade::{ConnectionUpgrade, Endpoint};

95
core/src/nodes/collection.rs
View File

@ -25,16 +25,15 @@ use nodes::node::Substream;
use nodes::handled_node_tasks::{HandledNodesEvent, HandledNodesTasks};
use nodes::handled_node_tasks::{Task as HandledNodesTask, TaskId};
use nodes::handled_node::NodeHandler;
use std::{collections::hash_map::Entry, fmt, mem};
use std::io::{Error as IoError, ErrorKind as IoErrorKind};
use {Multiaddr, PeerId};
use std::{collections::hash_map::Entry, fmt, io, mem};
use PeerId;
// TODO: make generic over PeerId
/// Implementation of `Stream` that handles a collection of nodes.
pub struct CollectionStream<TInEvent, TOutEvent> {
pub struct CollectionStream<TInEvent, TOutEvent, THandler> {
/// Object that handles the tasks.
inner: HandledNodesTasks<TInEvent, TOutEvent>,
inner: HandledNodesTasks<TInEvent, TOutEvent, THandler>,
/// List of nodes, with the task id that handles this node. The corresponding entry in `tasks`
/// must always be in the `Connected` state.
nodes: FnvHashMap<PeerId, TaskId>,
@ -43,7 +42,7 @@ pub struct CollectionStream<TInEvent, TOutEvent> {
tasks: FnvHashMap<TaskId, TaskState>,
}
impl<TInEvent, TOutEvent> fmt::Debug for CollectionStream<TInEvent, TOutEvent> {
impl<TInEvent, TOutEvent, THandler> fmt::Debug for CollectionStream<TInEvent, TOutEvent, THandler> {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
let mut list = f.debug_list();
for (id, task) in &self.tasks {
@ -70,10 +69,10 @@ enum TaskState {
}
/// Event that can happen on the `CollectionStream`.
pub enum CollectionEvent<'a, TInEvent:'a , TOutEvent: 'a> {
pub enum CollectionEvent<'a, TInEvent:'a , TOutEvent: 'a, THandler: 'a> {
/// A connection to a node has succeeded. You must use the provided event in order to accept
/// the connection.
NodeReached(CollectionReachEvent<'a, TInEvent, TOutEvent>),
NodeReached(CollectionReachEvent<'a, TInEvent, TOutEvent, THandler>),
/// A connection to a node has been closed.
///
@ -85,11 +84,13 @@ pub enum CollectionEvent<'a, TInEvent:'a , TOutEvent: 'a> {
},
/// A connection to a node has errored.
///
/// Can only happen after a node has been successfully reached.
NodeError {
/// Identifier of the node.
peer_id: PeerId,
/// The error that happened.
error: IoError,
error: io::Error,
},
/// An error happened on the future that was trying to reach a node.
@ -97,7 +98,9 @@ pub enum CollectionEvent<'a, TInEvent:'a , TOutEvent: 'a> {
/// Identifier of the reach attempt that failed.
id: ReachAttemptId,
/// Error that happened on the future.
error: IoError,
error: io::Error,
/// The handler that was passed to `add_reach_attempt`.
handler: THandler,
},
/// A node has produced an event.
@ -109,7 +112,7 @@ pub enum CollectionEvent<'a, TInEvent:'a , TOutEvent: 'a> {
},
}
impl<'a, TInEvent, TOutEvent> fmt::Debug for CollectionEvent<'a, TInEvent, TOutEvent>
impl<'a, TInEvent, TOutEvent, THandler> fmt::Debug for CollectionEvent<'a, TInEvent, TOutEvent, THandler>
where TOutEvent: fmt::Debug
{
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
@ -130,7 +133,7 @@ where TOutEvent: fmt::Debug
.field("error", error)
.finish()
},
CollectionEvent::ReachError { ref id, ref error } => {
CollectionEvent::ReachError { ref id, ref error, .. } => {
f.debug_struct("CollectionEvent::ReachError")
.field("id", id)
.field("error", error)
@ -148,16 +151,16 @@ where TOutEvent: fmt::Debug
/// Event that happens when we reach a node.
#[must_use = "The node reached event is used to accept the newly-opened connection"]
pub struct CollectionReachEvent<'a, TInEvent: 'a, TOutEvent: 'a> {
pub struct CollectionReachEvent<'a, TInEvent: 'a, TOutEvent: 'a, THandler: 'a> {
/// Peer id we connected to.
peer_id: PeerId,
/// The task id that reached the node.
id: TaskId,
/// The `CollectionStream` we are referencing.
parent: &'a mut CollectionStream<TInEvent, TOutEvent>,
parent: &'a mut CollectionStream<TInEvent, TOutEvent, THandler>,
}
impl<'a, TInEvent, TOutEvent> CollectionReachEvent<'a, TInEvent, TOutEvent> {
impl<'a, TInEvent, TOutEvent, THandler> CollectionReachEvent<'a, TInEvent, TOutEvent, THandler> {
/// Returns the peer id the node that has been reached.
#[inline]
pub fn peer_id(&self) -> &PeerId {
@ -220,7 +223,7 @@ impl<'a, TInEvent, TOutEvent> CollectionReachEvent<'a, TInEvent, TOutEvent> {
}
}
impl<'a, TInEvent, TOutEvent> fmt::Debug for CollectionReachEvent<'a, TInEvent, TOutEvent> {
impl<'a, TInEvent, TOutEvent, THandler> fmt::Debug for CollectionReachEvent<'a, TInEvent, TOutEvent, THandler> {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
f.debug_struct("CollectionReachEvent")
.field("peer_id", &self.peer_id)
@ -229,7 +232,7 @@ impl<'a, TInEvent, TOutEvent> fmt::Debug for CollectionReachEvent<'a, TInEvent,
}
}
impl<'a, TInEvent, TOutEvent> Drop for CollectionReachEvent<'a, TInEvent, TOutEvent> {
impl<'a, TInEvent, TOutEvent, THandler> Drop for CollectionReachEvent<'a, TInEvent, TOutEvent, THandler> {
fn drop(&mut self) {
let task_state = self.parent.tasks.remove(&self.id);
debug_assert!(if let Some(TaskState::Pending) = task_state { true } else { false });
@ -255,7 +258,7 @@ pub enum CollectionNodeAccept {
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
pub struct ReachAttemptId(TaskId);
impl<TInEvent, TOutEvent> CollectionStream<TInEvent, TOutEvent> {
impl<TInEvent, TOutEvent, THandler> CollectionStream<TInEvent, TOutEvent, THandler> {
/// Creates a new empty collection.
#[inline]
pub fn new() -> Self {
@ -270,12 +273,11 @@ impl<TInEvent, TOutEvent> CollectionStream<TInEvent, TOutEvent> {
///
/// This method spawns a task dedicated to resolving this future and processing the node's
/// events.
pub fn add_reach_attempt<TFut, TMuxer, TAddrFut, THandler>(&mut self, future: TFut, handler: THandler)
pub fn add_reach_attempt<TFut, TMuxer>(&mut self, future: TFut, handler: THandler)
-> ReachAttemptId
where
TFut: Future<Item = ((PeerId, TMuxer), TAddrFut), Error = IoError> + Send + 'static,
TAddrFut: Future<Item = Multiaddr, Error = IoError> + Send + 'static,
THandler: NodeHandler<Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent> + Send + 'static,
TFut: Future<Item = (PeerId, TMuxer), Error = io::Error> + Send + 'static,
THandler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent> + Send + 'static,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
THandler::OutboundOpenInfo: Send + 'static, // TODO: shouldn't be required?
@ -362,44 +364,44 @@ impl<TInEvent, TOutEvent> CollectionStream<TInEvent, TOutEvent> {
/// > **Note**: we use a regular `poll` method instead of implementing `Stream` in order to
/// > remove the `Err` variant, but also because we want the `CollectionStream` to stay
/// > borrowed if necessary.
pub fn poll(&mut self) -> Async<Option<CollectionEvent<TInEvent, TOutEvent>>> {
pub fn poll(&mut self) -> Async<CollectionEvent<TInEvent, TOutEvent, THandler>> {
let item = match self.inner.poll() {
Async::Ready(item) => item,
Async::NotReady => return Async::NotReady,
};
match item {
Some(HandledNodesEvent::TaskClosed { id, result }) => {
match (self.tasks.remove(&id), result) {
(Some(TaskState::Pending), Err(err)) => {
Async::Ready(Some(CollectionEvent::ReachError {
HandledNodesEvent::TaskClosed { id, result, handler } => {
match (self.tasks.remove(&id), result, handler) {
(Some(TaskState::Pending), Err(err), Some(handler)) => {
Async::Ready(CollectionEvent::ReachError {
id: ReachAttemptId(id),
error: err,
}))
handler,
})
},
(Some(TaskState::Pending), Ok(())) => {
(Some(TaskState::Pending), _, _) => {
// TODO: this variant shouldn't happen ; prove this
Async::Ready(Some(CollectionEvent::ReachError {
id: ReachAttemptId(id),
error: IoError::new(IoErrorKind::Other, "couldn't reach the node"),
}))
panic!()
},
(Some(TaskState::Connected(peer_id)), Ok(())) => {
(Some(TaskState::Connected(peer_id)), Ok(()), _handler) => {
debug_assert!(_handler.is_none());
let _node_task_id = self.nodes.remove(&peer_id);
debug_assert_eq!(_node_task_id, Some(id));
Async::Ready(Some(CollectionEvent::NodeClosed {
Async::Ready(CollectionEvent::NodeClosed {
peer_id,
}))
})
},
(Some(TaskState::Connected(peer_id)), Err(err)) => {
(Some(TaskState::Connected(peer_id)), Err(err), _handler) => {
debug_assert!(_handler.is_none());
let _node_task_id = self.nodes.remove(&peer_id);
debug_assert_eq!(_node_task_id, Some(id));
Async::Ready(Some(CollectionEvent::NodeError {
Async::Ready(CollectionEvent::NodeError {
peer_id,
error: err,
}))
})
},
(None, _) => {
(None, _, _) => {
panic!("self.tasks is always kept in sync with the tasks in self.inner ; \
when we add a task in self.inner we add a corresponding entry in \
self.tasks, and remove the entry only when the task is closed ; \
@ -407,14 +409,14 @@ impl<TInEvent, TOutEvent> CollectionStream<TInEvent, TOutEvent> {
},
}
},
Some(HandledNodesEvent::NodeReached { id, peer_id }) => {
Async::Ready(Some(CollectionEvent::NodeReached(CollectionReachEvent {
HandledNodesEvent::NodeReached { id, peer_id } => {
Async::Ready(CollectionEvent::NodeReached(CollectionReachEvent {
parent: self,
id,
peer_id,
})))
}))
},
Some(HandledNodesEvent::NodeEvent { id, event }) => {
HandledNodesEvent::NodeEvent { id, event } => {
let peer_id = match self.tasks.get(&id) {
Some(TaskState::Connected(peer_id)) => peer_id.clone(),
_ => panic!("we can only receive NodeEvent events from a task after we \
@ -423,12 +425,11 @@ impl<TInEvent, TOutEvent> CollectionStream<TInEvent, TOutEvent> {
self.tasks is switched to the Connected state ; qed"),
};
Async::Ready(Some(CollectionEvent::NodeEvent {
Async::Ready(CollectionEvent::NodeEvent {
peer_id,
event,
}))
})
}
None => Async::Ready(None),
}
}
}

70
core/src/nodes/handled_node.rs
View File

@ -22,20 +22,18 @@ use muxing::StreamMuxer;
use nodes::node::{NodeEvent, NodeStream, Substream};
use futures::{prelude::*, stream::Fuse};
use std::io::Error as IoError;
use Multiaddr;
/// Handler for the substreams of a node.
///
/// > Note: When implementing the various methods, don't forget that you have to register the
/// > task that was the latest to poll and notify it.
// TODO: right now it is possible for a node handler to be built, then shut down right after if we
// realize we dialed the wrong peer for example ; this could be surprising and should either
// be documented or changed (favouring the "documented" right now)
pub trait NodeHandler<TSubstream> {
pub trait NodeHandler {
/// 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 by the swarm.
type OutEvent;
/// The type of the substream containing the data.
type Substream;
/// Information about a substream. Can be sent to the handler through a `NodeHandlerEndpoint`,
/// and will be passed back in `inject_substream` or `inject_outbound_closed`.
type OutboundOpenInfo;
@ -43,7 +41,7 @@ pub trait NodeHandler<TSubstream> {
/// Sends a new substream to the handler.
///
/// The handler is responsible for upgrading the substream to whatever protocol it wants.
fn inject_substream(&mut self, substream: TSubstream, endpoint: NodeHandlerEndpoint<Self::OutboundOpenInfo>);
fn inject_substream(&mut self, substream: Self::Substream, endpoint: NodeHandlerEndpoint<Self::OutboundOpenInfo>);
/// Indicates to the handler that the inbound part of the muxer has been closed, and that
/// therefore no more inbound substream will be produced.
@ -53,9 +51,6 @@ pub trait NodeHandler<TSubstream> {
/// part of the muxer has been closed.
fn inject_outbound_closed(&mut self, user_data: Self::OutboundOpenInfo);
/// Indicates to the handler that the multiaddr future has resolved.
fn inject_multiaddr(&mut self, multiaddr: Result<Multiaddr, IoError>);
/// Injects an event coming from the outside into the handler.
fn inject_event(&mut self, event: Self::InEvent);
@ -78,6 +73,26 @@ pub enum NodeHandlerEndpoint<TOutboundOpenInfo> {
Listener,
}
impl<TOutboundOpenInfo> NodeHandlerEndpoint<TOutboundOpenInfo> {
/// Returns true for `Dialer`.
#[inline]
pub fn is_dialer(&self) -> bool {
match self {
NodeHandlerEndpoint::Dialer(_) => true,
NodeHandlerEndpoint::Listener => false,
}
}
/// Returns true for `Listener`.
#[inline]
pub fn is_listener(&self) -> bool {
match self {
NodeHandlerEndpoint::Dialer(_) => false,
NodeHandlerEndpoint::Listener => true,
}
}
}
/// Event produced by a handler.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum NodeHandlerEvent<TOutboundOpenInfo, TCustom> {
@ -119,30 +134,29 @@ impl<TOutboundOpenInfo, TCustom> NodeHandlerEvent<TOutboundOpenInfo, TCustom> {
/// A node combined with an implementation of `NodeHandler`.
// TODO: impl Debug
pub struct HandledNode<TMuxer, TAddrFut, THandler>
pub struct HandledNode<TMuxer, THandler>
where
TMuxer: StreamMuxer,
THandler: NodeHandler<Substream<TMuxer>>,
THandler: NodeHandler<Substream = Substream<TMuxer>>,
{
/// Node that handles the muxing.
node: Fuse<NodeStream<TMuxer, TAddrFut, THandler::OutboundOpenInfo>>,
node: Fuse<NodeStream<TMuxer, THandler::OutboundOpenInfo>>,
/// Handler that processes substreams.
handler: THandler,
// True, if the node is shutting down.
is_shutting_down: bool
}
impl<TMuxer, TAddrFut, THandler> HandledNode<TMuxer, TAddrFut, THandler>
impl<TMuxer, THandler> HandledNode<TMuxer, THandler>
where
TMuxer: StreamMuxer,
THandler: NodeHandler<Substream<TMuxer>>,
TAddrFut: Future<Item = Multiaddr, Error = IoError>,
THandler: NodeHandler<Substream = Substream<TMuxer>>,
{
/// Builds a new `HandledNode`.
#[inline]
pub fn new(muxer: TMuxer, multiaddr_future: TAddrFut, handler: THandler) -> Self {
pub fn new(muxer: TMuxer, handler: THandler) -> Self {
HandledNode {
node: NodeStream::new(muxer, multiaddr_future).fuse(),
node: NodeStream::new(muxer).fuse(),
handler,
is_shutting_down: false
}
@ -192,11 +206,10 @@ where
}
}
impl<TMuxer, TAddrFut, THandler> Stream for HandledNode<TMuxer, TAddrFut, THandler>
impl<TMuxer, THandler> Stream for HandledNode<TMuxer, THandler>
where
TMuxer: StreamMuxer,
THandler: NodeHandler<Substream<TMuxer>>,
TAddrFut: Future<Item = Multiaddr, Error = IoError>,
THandler: NodeHandler<Substream = Substream<TMuxer>>,
{
type Item = THandler::OutEvent;
type Error = IoError;
@ -220,9 +233,6 @@ where
self.handler.shutdown()
}
}
Async::Ready(Some(NodeEvent::Multiaddr(result))) => {
self.handler.inject_multiaddr(result)
}
Async::Ready(Some(NodeEvent::OutboundClosed { user_data })) => {
self.handler.inject_outbound_closed(user_data)
}
@ -263,8 +273,8 @@ where
#[cfg(test)]
mod tests {
use super::*;
use futures::future;
use muxing::{StreamMuxer, Shutdown};
use std::marker::PhantomData;
use tokio::runtime::current_thread;
// TODO: move somewhere? this could be useful as a dummy
@ -288,15 +298,17 @@ mod tests {
#[test]
fn proper_shutdown() {
// Test that `shutdown()` is properly called on the handler once a node stops.
struct Handler {
struct Handler<T> {
did_substream_attempt: bool,
inbound_closed: bool,
substream_attempt_cancelled: bool,
shutdown_called: bool,
marker: PhantomData<T>,
};
impl<T> NodeHandler<T> for Handler {
impl<T> NodeHandler for Handler<T> {
type InEvent = ();
type OutEvent = ();
type Substream = T;
type OutboundOpenInfo = ();
fn inject_substream(&mut self, _: T, _: NodeHandlerEndpoint<()>) { panic!() }
fn inject_inbound_closed(&mut self) {
@ -307,7 +319,6 @@ mod tests {
assert!(!self.substream_attempt_cancelled);
self.substream_attempt_cancelled = true;
}
fn inject_multiaddr(&mut self, _: Result<Multiaddr, IoError>) {}
fn inject_event(&mut self, _: Self::InEvent) { panic!() }
fn shutdown(&mut self) {
assert!(self.inbound_closed);
@ -325,17 +336,18 @@ mod tests {
}
}
}
impl Drop for Handler {
impl<T> Drop for Handler<T> {
fn drop(&mut self) {
assert!(self.shutdown_called);
}
}
let handled = HandledNode::new(InstaCloseMuxer, future::empty(), Handler {
let handled = HandledNode::new(InstaCloseMuxer, Handler {
did_substream_attempt: false,
inbound_closed: false,
substream_attempt_cancelled: false,
shutdown_called: false,
marker: PhantomData,
});
current_thread::Runtime::new().unwrap().block_on(handled.for_each(|_| Ok(()))).unwrap();

87
core/src/nodes/handled_node_tasks.rs
View File

@ -29,7 +29,7 @@ use std::io::Error as IoError;
use std::{fmt, mem};
use tokio_executor;
use void::Void;
use {Multiaddr, PeerId};
use PeerId;
// TODO: make generic over PeerId
@ -51,7 +51,7 @@ use {Multiaddr, PeerId};
/// Implementation of `Stream` that handles a collection of nodes.
// TODO: implement Debug
pub struct HandledNodesTasks<TInEvent, TOutEvent> {
pub struct HandledNodesTasks<TInEvent, TOutEvent, THandler> {
/// For each active task, a sender allowing to transmit messages. Closing the sender interrupts
/// the task. It is possible that we receive messages from tasks that used to be in this list
/// but no longer are, in which case we should ignore them.
@ -64,12 +64,12 @@ pub struct HandledNodesTasks<TInEvent, TOutEvent> {
to_spawn: SmallVec<[Box<Future<Item = (), Error = ()> + Send>; 8]>,
/// Sender to emit events to the outside. Meant to be cloned and sent to tasks.
events_tx: mpsc::UnboundedSender<(InToExtMessage<TOutEvent>, TaskId)>,
events_tx: mpsc::UnboundedSender<(InToExtMessage<TOutEvent, THandler>, TaskId)>,
/// Receiver side for the events.
events_rx: mpsc::UnboundedReceiver<(InToExtMessage<TOutEvent>, TaskId)>,
events_rx: mpsc::UnboundedReceiver<(InToExtMessage<TOutEvent, THandler>, TaskId)>,
}
impl<TInEvent, TOutEvent> fmt::Debug for HandledNodesTasks<TInEvent, TOutEvent> {
impl<TInEvent, TOutEvent, THandler> fmt::Debug for HandledNodesTasks<TInEvent, TOutEvent, THandler> {
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
f.debug_list()
.entries(self.tasks.keys().cloned())
@ -79,15 +79,19 @@ impl<TInEvent, TOutEvent> fmt::Debug for HandledNodesTasks<TInEvent, TOutEvent>
/// Event that can happen on the `HandledNodesTasks`.
#[derive(Debug)]
pub enum HandledNodesEvent<TOutEvent> {
pub enum HandledNodesEvent<TOutEvent, THandler> {
/// A task has been closed.
///
/// This happens once the node handler closes or an error happens.
// TODO: send back undelivered events?
TaskClosed {
/// Identifier of the task that closed.
id: TaskId,
/// What happened.
result: Result<(), IoError>,
/// If the task closed before reaching the node, this contains the handler that was passed
/// to `add_reach_attempt`.
handler: Option<THandler>,
},
/// A task has succeesfully connected to a node.
@ -111,7 +115,7 @@ pub enum HandledNodesEvent<TOutEvent> {
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
pub struct TaskId(usize);
impl<TInEvent, TOutEvent> HandledNodesTasks<TInEvent, TOutEvent> {
impl<TInEvent, TOutEvent, THandler> HandledNodesTasks<TInEvent, TOutEvent, THandler> {
/// Creates a new empty collection.
#[inline]
pub fn new() -> Self {
@ -130,12 +134,11 @@ impl<TInEvent, TOutEvent> HandledNodesTasks<TInEvent, TOutEvent> {
///
/// This method spawns a task dedicated to resolving this future and processing the node's
/// events.
pub fn add_reach_attempt<TFut, TMuxer, TAddrFut, THandler>(&mut self, future: TFut, handler: THandler)
pub fn add_reach_attempt<TFut, TMuxer>(&mut self, future: TFut, handler: THandler)
-> TaskId
where
TFut: Future<Item = ((PeerId, TMuxer), TAddrFut), Error = IoError> + Send + 'static,
TAddrFut: Future<Item = Multiaddr, Error = IoError> + Send + 'static,
THandler: NodeHandler<Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent> + Send + 'static,
TFut: Future<Item = (PeerId, TMuxer), Error = IoError> + Send + 'static,
THandler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent> + Send + 'static,
TInEvent: Send + 'static,
TOutEvent: Send + 'static,
THandler::OutboundOpenInfo: Send + 'static, // TODO: shouldn't be required?
@ -193,7 +196,7 @@ impl<TInEvent, TOutEvent> HandledNodesTasks<TInEvent, TOutEvent> {
}
/// Provides an API similar to `Stream`, except that it cannot error.
pub fn poll(&mut self) -> Async<Option<HandledNodesEvent<TOutEvent>>> {
pub fn poll(&mut self) -> Async<HandledNodesEvent<TOutEvent, THandler>> {
for to_spawn in self.to_spawn.drain() {
tokio_executor::spawn(to_spawn);
}
@ -210,22 +213,22 @@ impl<TInEvent, TOutEvent> HandledNodesTasks<TInEvent, TOutEvent> {
match message {
InToExtMessage::NodeEvent(event) => {
break Async::Ready(Some(HandledNodesEvent::NodeEvent {
break Async::Ready(HandledNodesEvent::NodeEvent {
id: task_id,
event,
}));
});
},
InToExtMessage::NodeReached(peer_id) => {
break Async::Ready(Some(HandledNodesEvent::NodeReached {
break Async::Ready(HandledNodesEvent::NodeReached {
id: task_id,
peer_id,
}));
});
},
InToExtMessage::TaskClosed(result) => {
InToExtMessage::TaskClosed(result, handler) => {
let _ = self.tasks.remove(&task_id);
break Async::Ready(Some(HandledNodesEvent::TaskClosed {
id: task_id, result
}));
break Async::Ready(HandledNodesEvent::TaskClosed {
id: task_id, result, handler
});
},
}
}
@ -249,6 +252,7 @@ pub struct Task<'a, TInEvent: 'a> {
impl<'a, TInEvent> Task<'a, TInEvent> {
/// Sends an event to the given node.
// TODO: report back on delivery
#[inline]
pub fn send_event(&mut self, event: TInEvent) {
// It is possible that the sender is closed if the background task has already finished
@ -279,48 +283,48 @@ impl<'a, TInEvent> fmt::Debug for Task<'a, TInEvent> {
}
}
impl<TInEvent, TOutEvent> Stream for HandledNodesTasks<TInEvent, TOutEvent> {
type Item = HandledNodesEvent<TOutEvent>;
impl<TInEvent, TOutEvent, THandler> Stream for HandledNodesTasks<TInEvent, TOutEvent, THandler> {
type Item = HandledNodesEvent<TOutEvent, THandler>;
type Error = Void; // TODO: use ! once stable
#[inline]
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
Ok(self.poll())
Ok(self.poll().map(Option::Some))
}
}
/// Message to transmit from a task to the public API.
#[derive(Debug)]
enum InToExtMessage<TOutEvent> {
enum InToExtMessage<TOutEvent, THandler> {
/// A connection to a node has succeeded.
NodeReached(PeerId),
/// The task closed.
TaskClosed(Result<(), IoError>),
TaskClosed(Result<(), IoError>, Option<THandler>),
/// An event from the node.
NodeEvent(TOutEvent),
}
/// Implementation of `Future` that handles a single node, and all the communications between
/// the various components of the `HandledNodesTasks`.
struct NodeTask<TFut, TMuxer, TAddrFut, THandler, TInEvent, TOutEvent>
struct NodeTask<TFut, TMuxer, THandler, TInEvent, TOutEvent>
where
TMuxer: StreamMuxer,
THandler: NodeHandler<Substream<TMuxer>>,
THandler: NodeHandler<Substream = Substream<TMuxer>>,
{
/// Sender to transmit events to the outside.
events_tx: mpsc::UnboundedSender<(InToExtMessage<TOutEvent>, TaskId)>,
events_tx: mpsc::UnboundedSender<(InToExtMessage<TOutEvent, THandler>, TaskId)>,
/// Receiving end for events sent from the main `HandledNodesTasks`.
in_events_rx: stream::Fuse<mpsc::UnboundedReceiver<TInEvent>>,
/// Inner state of the `NodeTask`.
inner: NodeTaskInner<TFut, TMuxer, TAddrFut, THandler, TInEvent>,
inner: NodeTaskInner<TFut, TMuxer, THandler, TInEvent>,
/// Identifier of the attempt.
id: TaskId,
}
enum NodeTaskInner<TFut, TMuxer, TAddrFut, THandler, TInEvent>
enum NodeTaskInner<TFut, TMuxer, THandler, TInEvent>
where
TMuxer: StreamMuxer,
THandler: NodeHandler<Substream<TMuxer>>,
THandler: NodeHandler<Substream = Substream<TMuxer>>,
{
/// Future to resolve to connect to the node.
Future {
@ -335,19 +339,18 @@ where
},
/// Fully functional node.
Node(HandledNode<TMuxer, TAddrFut, THandler>),
Node(HandledNode<TMuxer, THandler>),
/// A panic happened while polling.
Poisoned,
}
impl<TFut, TMuxer, TAddrFut, THandler, TInEvent, TOutEvent> Future for
NodeTask<TFut, TMuxer, TAddrFut, THandler, TInEvent, TOutEvent>
impl<TFut, TMuxer, THandler, TInEvent, TOutEvent> Future for
NodeTask<TFut, TMuxer, THandler, TInEvent, TOutEvent>
where
TMuxer: StreamMuxer,
TFut: Future<Item = ((PeerId, TMuxer), TAddrFut), Error = IoError>,
TAddrFut: Future<Item = Multiaddr, Error = IoError>,
THandler: NodeHandler<Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent>,
TFut: Future<Item = (PeerId, TMuxer), Error = IoError>,
THandler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent>,
{
type Item = ();
type Error = ();
@ -369,9 +372,9 @@ where
// Check whether dialing succeeded.
match future.poll() {
Ok(Async::Ready(((peer_id, muxer), addr_fut))) => {
Ok(Async::Ready((peer_id, muxer))) => {
let event = InToExtMessage::NodeReached(peer_id);
let mut node = HandledNode::new(muxer, addr_fut, handler);
let mut node = HandledNode::new(muxer, handler);
for event in events_buffer {
node.inject_event(event);
}
@ -386,7 +389,7 @@ where
},
Err(err) => {
// End the task
let event = InToExtMessage::TaskClosed(Err(err));
let event = InToExtMessage::TaskClosed(Err(err), Some(handler));
let _ = self.events_tx.unbounded_send((event, self.id));
return Ok(Async::Ready(()));
}
@ -427,12 +430,12 @@ where
}
}
Ok(Async::Ready(None)) => {
let event = InToExtMessage::TaskClosed(Ok(()));
let event = InToExtMessage::TaskClosed(Ok(()), None);
let _ = self.events_tx.unbounded_send((event, self.id));
return Ok(Async::Ready(())); // End the task.
}
Err(err) => {
let event = InToExtMessage::TaskClosed(Err(err));
let event = InToExtMessage::TaskClosed(Err(err), None);
let _ = self.events_tx.unbounded_send((event, self.id));
return Ok(Async::Ready(())); // End the task.
}

50
core/src/nodes/listeners.rs
View File

@ -60,7 +60,7 @@ use {Multiaddr, Transport};
/// ListenersEvent::Closed { listen_addr, listener, result } => {
/// println!("Listener {} has been closed: {:?}", listen_addr, result);
/// },
/// ListenersEvent::Incoming { upgrade, listen_addr } => {
/// ListenersEvent::Incoming { upgrade, listen_addr, .. } => {
/// println!("A connection has arrived on {}", listen_addr);
/// // We don't do anything with the newly-opened connection, but in a real-life
/// // program you probably want to use it!
@ -107,6 +107,8 @@ where
upgrade: TTrans::ListenerUpgrade,
/// Address of the listener which received the connection.
listen_addr: Multiaddr,
/// Address used to send back data to the incoming client.
send_back_addr: Multiaddr,
},
/// A listener has closed, either gracefully or with an error.
@ -177,7 +179,7 @@ where
}
/// Provides an API similar to `Stream`, except that it cannot error.
pub fn poll(&mut self) -> Async<Option<ListenersEvent<TTrans>>> {
pub fn poll(&mut self) -> Async<ListenersEvent<TTrans>> {
// We remove each element from `listeners` one by one and add them back.
for n in (0..self.listeners.len()).rev() {
let mut listener = self.listeners.swap_remove(n);
@ -185,27 +187,28 @@ where
Ok(Async::NotReady) => {
self.listeners.push(listener);
}
Ok(Async::Ready(Some(upgrade))) => {
Ok(Async::Ready(Some((upgrade, send_back_addr)))) => {
let listen_addr = listener.address.clone();
self.listeners.push(listener);
return Async::Ready(Some(ListenersEvent::Incoming {
return Async::Ready(ListenersEvent::Incoming {
upgrade,
listen_addr,
}));
send_back_addr,
});
}
Ok(Async::Ready(None)) => {
return Async::Ready(Some(ListenersEvent::Closed {
return Async::Ready(ListenersEvent::Closed {
listen_addr: listener.address,
listener: listener.listener,
result: Ok(()),
}));
});
}
Err(err) => {
return Async::Ready(Some(ListenersEvent::Closed {
return Async::Ready(ListenersEvent::Closed {
listen_addr: listener.address,
listener: listener.listener,
result: Err(err),
}));
});
}
}
}
@ -224,7 +227,7 @@ where
#[inline]
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
Ok(self.poll())
Ok(self.poll().map(Option::Some))
}
}
@ -294,7 +297,7 @@ mod tests {
Async::Ready(Some(n)) => {
let addr = l.address.clone();
let stream = stream::iter_ok(n..)
.map(move |stream| future::ok( (stream, future::ok(addr.clone())) ));
.map(move |stream| (future::ok(stream), addr.clone()));
Box::new(stream)
}
Async::Ready(None) => {
@ -320,8 +323,9 @@ mod tests {
.map_err(|(err, _)| err)
.and_then(|(event, _)| {
match event {
Some(ListenersEvent::Incoming { listen_addr, upgrade }) => {
Some(ListenersEvent::Incoming { listen_addr, upgrade, send_back_addr }) => {
assert_eq!(listen_addr, "/memory".parse().unwrap());
assert_eq!(send_back_addr, "/memory".parse().unwrap());
upgrade.map(|_| ()).map_err(|_| panic!())
},
_ => panic!()
@ -384,11 +388,11 @@ mod tests {
ls.listen_on(addr1).expect("listen_on failed");
ls.listen_on(addr2).expect("listen_on failed");
assert_matches!(ls.poll(), Async::Ready(Some(listeners_event)) => {
assert_matches!(listeners_event, ListenersEvent::Incoming{mut upgrade, listen_addr} => {
assert_matches!(ls.poll(), Async::Ready(listeners_event) => {
assert_matches!(listeners_event, ListenersEvent::Incoming{mut upgrade, listen_addr, ..} => {
assert_eq!(listen_addr.to_string(), "/ip4/127.0.0.2/tcp/4321");
assert_matches!(upgrade.poll().unwrap(), Async::Ready(tup) => {
assert_matches!(tup, (1, _))
assert_eq!(tup, 1)
});
})
});
@ -407,11 +411,11 @@ mod tests {
// Make the second listener return NotReady so we get the first listener next poll()
set_listener_state(&mut ls, 1, ListenerState::Ok(Async::NotReady));
assert_matches!(ls.poll(), Async::Ready(Some(listeners_event)) => {
assert_matches!(listeners_event, ListenersEvent::Incoming{mut upgrade, listen_addr} => {
assert_matches!(ls.poll(), Async::Ready(listeners_event) => {
assert_matches!(listeners_event, ListenersEvent::Incoming{mut upgrade, listen_addr, ..} => {
assert_eq!(listen_addr.to_string(), "/ip4/127.0.0.1/tcp/1234");
assert_matches!(upgrade.poll().unwrap(), Async::Ready(tup) => {
assert_matches!(tup, (1, _))
assert_eq!(tup, 1)
});
})
});
@ -425,7 +429,7 @@ mod tests {
let mut ls = ListenersStream::new(t);
ls.listen_on(addr).expect("listen_on failed");
set_listener_state(&mut ls, 0, ListenerState::Ok(Async::Ready(None)));
assert_matches!(ls.poll(), Async::Ready(Some(listeners_event)) => {
assert_matches!(ls.poll(), Async::Ready(listeners_event) => {
assert_matches!(listeners_event, ListenersEvent::Closed{..})
});
assert_eq!(ls.listeners.len(), 0); // it's gone
@ -439,7 +443,7 @@ mod tests {
let mut ls = ListenersStream::new(t);
ls.listen_on(addr).expect("listen_on failed");
set_listener_state(&mut ls, 0, ListenerState::Error); // simulate an error on the socket
assert_matches!(ls.poll(), Async::Ready(Some(listeners_event)) => {
assert_matches!(ls.poll(), Async::Ready(listeners_event) => {
assert_matches!(listeners_event, ListenersEvent::Closed{..})
});
assert_eq!(ls.listeners.len(), 0); // it's gone
@ -458,14 +462,14 @@ mod tests {
// polling processes listeners in reverse order
// Only the last listener ever gets processed
for _n in 0..10 {
assert_matches!(ls.poll(), Async::Ready(Some(ListenersEvent::Incoming{listen_addr, ..})) => {
assert_matches!(ls.poll(), Async::Ready(ListenersEvent::Incoming{listen_addr, ..}) => {
assert_eq!(listen_addr.to_string(), "/ip4/127.0.0.3/tcp/1233")
})
}
// Make last listener NotReady so now only the third listener is processed
set_listener_state(&mut ls, 3, ListenerState::Ok(Async::NotReady));
for _n in 0..10 {
assert_matches!(ls.poll(), Async::Ready(Some(ListenersEvent::Incoming{listen_addr, ..})) => {
assert_matches!(ls.poll(), Async::Ready(ListenersEvent::Incoming{listen_addr, ..}) => {
assert_eq!(listen_addr.to_string(), "/ip4/127.0.0.2/tcp/1232")
})
}
@ -483,7 +487,7 @@ mod tests {
// If the listeners do not yield items continuously (the normal case) we
// process them in the expected, reverse, order.
for n in (0..4).rev() {
assert_matches!(ls.poll(), Async::Ready(Some(ListenersEvent::Incoming{listen_addr, ..})) => {
assert_matches!(ls.poll(), Async::Ready(ListenersEvent::Incoming{listen_addr, ..}) => {
assert_eq!(listen_addr.to_string(), format!("/ip4/127.0.0.{}/tcp/123{}", n, n));
});
// kick the last listener (current) to NotReady state

9
core/src/nodes/mod.rs
View File

@ -18,10 +18,13 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
mod handled_node_tasks;
pub mod collection;
pub mod handled_node;
pub mod handled_node_tasks;
pub mod listeners;
pub mod node;
pub mod swarm;
pub mod raw_swarm;
pub use self::node::Substream;
pub use self::handled_node::{NodeHandlerEvent, NodeHandlerEndpoint};
pub use self::raw_swarm::{ConnectedPoint, Peer, RawSwarm, RawSwarmEvent};

136
core/src/nodes/node.rs
View File

@ -24,16 +24,12 @@ use smallvec::SmallVec;
use std::fmt;
use std::io::Error as IoError;
use std::sync::Arc;
use Multiaddr;
// Implementor notes
// =================
//
// In order to minimize the risk of bugs in higher-level code, we want to avoid as much as
// possible having a racy API. The behaviour of methods should be well-defined and predictable.
// As an example, calling the `multiaddr()` method should return `Some` only after a
// `MultiaddrResolved` event has been emitted and never before, even if we technically already
// know the address.
//
// In order to respect this coding practice, we should theoretically provide events such as "data
// incoming on a substream", or "a substream is ready to be written". This would however make the
@ -53,7 +49,7 @@ use Multiaddr;
///
/// The stream will close once both the inbound and outbound channels are closed, and no more
/// outbound substream attempt is pending.
pub struct NodeStream<TMuxer, TAddrFut, TUserData>
pub struct NodeStream<TMuxer, TUserData>
where
TMuxer: muxing::StreamMuxer,
{
@ -63,23 +59,10 @@ where
inbound_state: StreamState,
/// Tracks the state of the muxers outbound direction.
outbound_state: StreamState,
/// Address of the node ; can be empty if the address hasn't been resolved yet.
address: Addr<TAddrFut>,
/// List of substreams we are currently opening.
outbound_substreams: SmallVec<[(TUserData, TMuxer::OutboundSubstream); 8]>,
}
/// Address of the node.
#[derive(Debug, Clone)]
enum Addr<TAddrFut> {
/// Future that will resolve the address.
Future(TAddrFut),
/// The address is now known.
Resolved(Multiaddr),
/// An error happened while resolving the future.
Errored,
}
/// A successfully opened substream.
pub type Substream<TMuxer> = muxing::SubstreamRef<Arc<TMuxer>>;
@ -102,12 +85,6 @@ pub enum NodeEvent<TMuxer, TUserData>
where
TMuxer: muxing::StreamMuxer,
{
/// The multiaddress future of the node has been resolved.
///
/// If this succeeded, after this event has been emitted calling `multiaddr()` will return
/// `Some`.
Multiaddr(Result<Multiaddr, IoError>),
/// A new inbound substream arrived.
InboundSubstream {
/// The newly-opened substream.
@ -137,35 +114,21 @@ where
#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
pub struct OutboundSubstreamId(usize);
impl<TMuxer, TAddrFut, TUserData> NodeStream<TMuxer, TAddrFut, TUserData>
impl<TMuxer, TUserData> NodeStream<TMuxer, TUserData>
where
TMuxer: muxing::StreamMuxer,
TAddrFut: Future<Item = Multiaddr, Error = IoError>,
{
/// Creates a new node events stream.
#[inline]
pub fn new(muxer: TMuxer, multiaddr_future: TAddrFut) -> Self {
pub fn new(muxer: TMuxer) -> Self {
NodeStream {
muxer: Arc::new(muxer),
inbound_state: StreamState::Open,
outbound_state: StreamState::Open,
address: Addr::Future(multiaddr_future),
outbound_substreams: SmallVec::new(),
}
}
/// Returns the multiaddress of the node, if already known.
///
/// This method will always return `None` before a successful `Multiaddr` event has been
/// returned by `poll()`, and will always return `Some` afterwards.
#[inline]
pub fn multiaddr(&self) -> Option<&Multiaddr> {
match self.address {
Addr::Resolved(ref addr) => Some(addr),
Addr::Future(_) | Addr::Errored => None,
}
}
/// Starts the process of opening a new outbound substream.
///
/// Returns an error if the outbound side of the muxer is closed.
@ -286,10 +249,9 @@ where
}
}
impl<TMuxer, TAddrFut, TUserData> Stream for NodeStream<TMuxer, TAddrFut, TUserData>
impl<TMuxer, TUserData> Stream for NodeStream<TMuxer, TUserData>
where
TMuxer: muxing::StreamMuxer,
TAddrFut: Future<Item = Multiaddr, Error = IoError>,
{
type Item = NodeEvent<TMuxer, TUserData>;
type Error = IoError;
@ -345,26 +307,6 @@ where
}
}
// Check whether the multiaddress is resolved.
{
let poll = match self.address {
Addr::Future(ref mut fut) => Some(fut.poll()),
Addr::Resolved(_) | Addr::Errored => None,
};
match poll {
Some(Ok(Async::NotReady)) | None => {}
Some(Ok(Async::Ready(addr))) => {
self.address = Addr::Resolved(addr.clone());
return Ok(Async::Ready(Some(NodeEvent::Multiaddr(Ok(addr)))));
}
Some(Err(err)) => {
self.address = Addr::Errored;
return Ok(Async::Ready(Some(NodeEvent::Multiaddr(Err(err)))));
}
}
}
// Closing the node if there's no way we can do anything more.
if self.inbound_state == StreamState::Closed
&& self.outbound_state == StreamState::Closed
@ -378,14 +320,12 @@ where
}
}
impl<TMuxer, TAddrFut, TUserData> fmt::Debug for NodeStream<TMuxer, TAddrFut, TUserData>
impl<TMuxer, TUserData> fmt::Debug for NodeStream<TMuxer, TUserData>
where
TMuxer: muxing::StreamMuxer,
TAddrFut: Future<Item = Multiaddr, Error = IoError>,
{
fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
f.debug_struct("NodeStream")
.field("address", &self.multiaddr())
.field("inbound_state", &self.inbound_state)
.field("outbound_state", &self.outbound_state)
.field("outbound_substreams", &self.outbound_substreams.len())
@ -393,7 +333,7 @@ where
}
}
impl<TMuxer, TAddrFut, TUserData> Drop for NodeStream<TMuxer, TAddrFut, TUserData>
impl<TMuxer, TUserData> Drop for NodeStream<TMuxer, TUserData>
where
TMuxer: muxing::StreamMuxer,
{
@ -436,40 +376,22 @@ where TTrans: Transport,
#[cfg(test)]
mod node_stream {
use multiaddr::Multiaddr;
use super::NodeStream;
use futures::{future::self, prelude::*, Future};
use futures::prelude::*;
use tokio_mock_task::MockTask;
use super::NodeEvent;
use tests::dummy_muxer::{DummyMuxer, DummyConnectionState};
use std::io::Error as IoError;
fn build_node_stream() -> NodeStream<DummyMuxer, impl Future<Item=Multiaddr, Error=IoError>, Vec<u8>> {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad maddr"));
fn build_node_stream() -> NodeStream<DummyMuxer, Vec<u8>> {
let muxer = DummyMuxer::new();
NodeStream::<_, _, Vec<u8>>::new(muxer, addr)
}
#[test]
fn multiaddr_is_available_once_polled() {
let mut node_stream = build_node_stream();
assert!(node_stream.multiaddr().is_none());
match node_stream.poll() {
Ok(Async::Ready(Some(NodeEvent::Multiaddr(Ok(addr))))) => {
assert_eq!(addr.to_string(), "/ip4/127.0.0.1/tcp/1234")
}
_ => panic!("unexpected poll return value" )
}
assert!(node_stream.multiaddr().is_some());
NodeStream::<_, Vec<u8>>::new(muxer)
}
#[test]
fn can_open_outbound_substreams_until_an_outbound_channel_is_closed() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad maddr"));
let mut muxer = DummyMuxer::new();
muxer.set_outbound_connection_state(DummyConnectionState::Closed);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
// open first substream works
assert!(ns.open_substream(vec![1,2,3]).is_ok());
@ -498,10 +420,9 @@ mod node_stream {
#[test]
fn query_inbound_state() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad maddr"));
let mut muxer = DummyMuxer::new();
muxer.set_inbound_connection_state(DummyConnectionState::Closed);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
assert_matches!(ns.poll(), Ok(Async::Ready(Some(node_event))) => {
assert_matches!(node_event, NodeEvent::InboundClosed)
@ -512,10 +433,9 @@ mod node_stream {
#[test]
fn query_outbound_state() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr"));
let mut muxer = DummyMuxer::new();
muxer.set_outbound_connection_state(DummyConnectionState::Closed);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
assert!(ns.is_outbound_open());
@ -548,13 +468,12 @@ mod node_stream {
let mut task = MockTask::new();
task.enter(|| {
// ensure the address never resolves
let addr = future::empty();
let mut muxer = DummyMuxer::new();
// ensure muxer.poll_inbound() returns Async::NotReady
muxer.set_inbound_connection_state(DummyConnectionState::Pending);
// ensure muxer.poll_outbound() returns Async::NotReady
muxer.set_outbound_connection_state(DummyConnectionState::Pending);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
assert_matches!(ns.poll(), Ok(Async::NotReady));
});
@ -562,13 +481,12 @@ mod node_stream {
#[test]
fn poll_closes_the_node_stream_when_no_more_work_can_be_done() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr"));
let mut muxer = DummyMuxer::new();
// ensure muxer.poll_inbound() returns Async::Ready(None)
muxer.set_inbound_connection_state(DummyConnectionState::Closed);
// ensure muxer.poll_outbound() returns Async::Ready(None)
muxer.set_outbound_connection_state(DummyConnectionState::Closed);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
ns.open_substream(vec![]).unwrap();
ns.poll().unwrap(); // poll_inbound()
ns.poll().unwrap(); // poll_outbound()
@ -577,32 +495,14 @@ mod node_stream {
assert_matches!(ns.poll(), Ok(Async::Ready(None)));
}
#[test]
fn poll_resolves_the_address() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr"));
let mut muxer = DummyMuxer::new();
// ensure muxer.poll_inbound() returns Async::Ready(None)
muxer.set_inbound_connection_state(DummyConnectionState::Closed);
// ensure muxer.poll_outbound() returns Async::Ready(None)
muxer.set_outbound_connection_state(DummyConnectionState::Closed);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
ns.open_substream(vec![]).unwrap();
ns.poll().unwrap(); // poll_inbound()
ns.poll().unwrap(); // poll_outbound()
assert_matches!(ns.poll(), Ok(Async::Ready(Some(node_event))) => {
assert_matches!(node_event, NodeEvent::Multiaddr(Ok(_)))
});
}
#[test]
fn poll_sets_up_substreams_yielding_them_in_reverse_order() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr"));
let mut muxer = DummyMuxer::new();
// ensure muxer.poll_inbound() returns Async::Ready(None)
muxer.set_inbound_connection_state(DummyConnectionState::Closed);
// ensure muxer.poll_outbound() returns Async::Ready(Some(substream))
muxer.set_outbound_connection_state(DummyConnectionState::Opened);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
ns.open_substream(vec![1]).unwrap();
ns.open_substream(vec![2]).unwrap();
ns.poll().unwrap(); // poll_inbound()
@ -623,13 +523,12 @@ mod node_stream {
#[test]
fn poll_keeps_outbound_substreams_when_the_outgoing_connection_is_not_ready() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr"));
let mut muxer = DummyMuxer::new();
// ensure muxer.poll_inbound() returns Async::Ready(None)
muxer.set_inbound_connection_state(DummyConnectionState::Closed);
// ensure muxer.poll_outbound() returns Async::NotReady
muxer.set_outbound_connection_state(DummyConnectionState::Pending);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
ns.open_substream(vec![1]).unwrap();
ns.poll().unwrap(); // poll past inbound
ns.poll().unwrap(); // poll outbound
@ -639,11 +538,10 @@ mod node_stream {
#[test]
fn poll_returns_incoming_substream() {
let addr = future::ok("/ip4/127.0.0.1/tcp/1234".parse::<Multiaddr>().expect("bad multiaddr"));
let mut muxer = DummyMuxer::new();
// ensure muxer.poll_inbound() returns Async::Ready(Some(subs))
muxer.set_inbound_connection_state(DummyConnectionState::Opened);
let mut ns = NodeStream::<_, _, Vec<u8>>::new(muxer, addr);
let mut ns = NodeStream::<_, Vec<u8>>::new(muxer);
assert_matches!(ns.poll(), Ok(Async::Ready(Some(node_event))) => {
assert_matches!(node_event, NodeEvent::InboundSubstream{ substream: _ });
});

574
core/src/nodes/swarm.rs → core/src/nodes/raw_swarm.rs
View File

File diff suppressed because it is too large Load Diff

9
core/src/tests/dummy_transport.rs
View File

@ -47,10 +47,9 @@ impl DummyTransport {
}
impl Transport for DummyTransport {
type Output = usize;
type Listener = Box<Stream<Item=Self::ListenerUpgrade, Error=io::Error> + Send>;
type ListenerUpgrade = FutureResult<(Self::Output, Self::MultiaddrFuture), io::Error>;
type MultiaddrFuture = FutureResult<Multiaddr, io::Error>;
type Dial = Box<Future<Item=(Self::Output, Self::MultiaddrFuture), Error=io::Error> + Send>;
type Listener = Box<Stream<Item=(Self::ListenerUpgrade, Multiaddr), Error=io::Error> + Send>;
type ListenerUpgrade = FutureResult<Self::Output, io::Error>;
type Dial = Box<Future<Item=Self::Output, Error=io::Error> + Send>;
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)>
where
@ -59,7 +58,7 @@ impl Transport for DummyTransport {
let addr2 = addr.clone();
match self.listener_state {
ListenerState::Ok(async) => {
let tupelize = move |stream| future::ok( (stream, future::ok(addr.clone())) );
let tupelize = move |stream| (future::ok(stream), addr.clone());
Ok(match async {
Async::NotReady => {
let stream = stream::poll_fn(|| Ok(Async::NotReady)).map(tupelize);

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

@ -20,9 +20,9 @@
use futures::prelude::*;
use multiaddr::Multiaddr;
use nodes::raw_swarm::ConnectedPoint;
use std::io::Error as IoError;
use transport::{MuxedTransport, Transport};
use upgrade::Endpoint;
use transport::Transport;
/// See the `Transport::and_then` method.
#[inline]
@ -37,21 +37,19 @@ pub struct AndThen<T, C> {
upgrade: C,
}
impl<T, C, F, O, Maf> Transport for AndThen<T, C>
impl<T, C, F, O> Transport for AndThen<T, C>
where
T: Transport + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
C: FnOnce(T::Output, Endpoint, T::MultiaddrFuture) -> F + Clone + Send + 'static,
F: Future<Item = (O, Maf), Error = IoError> + Send + 'static,
Maf: Future<Item = Multiaddr, Error = IoError> + 'static,
C: FnOnce(T::Output, ConnectedPoint) -> F + Clone + Send + 'static,
F: Future<Item = O, Error = IoError> + Send + 'static,
{
type Output = O;
type MultiaddrFuture = Maf;
type Listener = Box<Stream<Item = Self::ListenerUpgrade, Error = IoError> + Send>;
type ListenerUpgrade = Box<Future<Item = (O, Self::MultiaddrFuture), Error = IoError> + Send>;
type Dial = Box<Future<Item = (O, Self::MultiaddrFuture), Error = IoError> + Send>;
type Listener = Box<Stream<Item = (Self::ListenerUpgrade, Multiaddr), Error = IoError> + Send>;
type ListenerUpgrade = Box<Future<Item = O, Error = IoError> + Send>;
type Dial = Box<Future<Item = O, Error = IoError> + Send>;
#[inline]
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
@ -69,17 +67,24 @@ where
}
};
let listen_addr = new_addr.clone();
// Try to negotiate the protocol.
// Note that failing to negotiate a protocol will never produce a future with an error.
// Instead the `stream` will produce `Ok(Err(...))`.
// `stream` can only produce an `Err` if `listening_stream` produces an `Err`.
let stream = listening_stream.map(move |connection| {
let stream = listening_stream.map(move |(connection, client_addr)| {
let upgrade = upgrade.clone();
let future = connection.and_then(move |(stream, client_addr)| {
upgrade(stream, Endpoint::Listener, client_addr)
let connected_point = ConnectedPoint::Listener {
listen_addr: listen_addr.clone(),
send_back_addr: client_addr.clone(),
};
let future = connection.and_then(move |stream| {
upgrade(stream, connected_point)
});
Box::new(future) as Box<_>
(Box::new(future) as Box<_>, client_addr)
});
Ok((Box::new(stream), new_addr))
@ -101,10 +106,14 @@ where
}
};
let connected_point = ConnectedPoint::Dialer {
address: addr,
};
let future = dialed_fut
// Try to negotiate the protocol.
.and_then(move |(connection, client_addr)| {
upgrade(connection, Endpoint::Dialer, client_addr)
.and_then(move |connection| {
upgrade(connection, connected_point)
});
Ok(Box::new(future))
@ -115,36 +124,3 @@ where
self.transport.nat_traversal(server, observed)
}
}
impl<T, C, F, O, Maf> MuxedTransport for AndThen<T, C>
where
T: MuxedTransport + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
C: FnOnce(T::Output, Endpoint, T::MultiaddrFuture) -> F + Clone + Send + 'static,
F: Future<Item = (O, Maf), Error = IoError> + Send + 'static,
Maf: Future<Item = Multiaddr, Error = IoError> + 'static,
{
type Incoming = Box<Future<Item = Self::IncomingUpgrade, Error = IoError> + Send>;
type IncomingUpgrade = Box<Future<Item = (O, Self::MultiaddrFuture), Error = IoError> + Send>;
#[inline]
fn next_incoming(self) -> Self::Incoming {
let upgrade = self.upgrade;
let future = self.transport.next_incoming().map(|future| {
// Try to negotiate the protocol.
let future = future.and_then(move |(connection, client_addr)| {
let upgrade = upgrade.clone();
upgrade(connection, Endpoint::Listener, client_addr)
});
Box::new(future) as Box<Future<Item = _, Error = _> + Send>
});
Box::new(future) as Box<_>
}
}

122
core/src/transport/boxed.rs
View File

@ -23,7 +23,7 @@ use multiaddr::Multiaddr;
use std::fmt;
use std::io::Error as IoError;
use std::sync::Arc;
use transport::{MuxedTransport, Transport};
use transport::Transport;
/// See the `Transport::boxed` method.
#[inline]
@ -33,35 +33,17 @@ where
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
T::MultiaddrFuture: Send + 'static,
{
Boxed {
inner: Arc::new(transport) as Arc<_>,
}
}
/// See the `Transport::boxed_muxed` method.
#[inline]
pub fn boxed_muxed<T>(transport: T) -> BoxedMuxed<T::Output>
where
T: MuxedTransport + Clone + Send + Sync + 'static,
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Incoming: Send + 'static,
T::IncomingUpgrade: Send + 'static,
{
BoxedMuxed {
inner: Arc::new(transport) as Arc<_>,
}
}
pub type MultiaddrFuture = Box<Future<Item = Multiaddr, Error = IoError> + Send>;
pub type Dial<O> = Box<Future<Item = (O, MultiaddrFuture), Error = IoError> + Send>;
pub type Listener<O> = Box<Stream<Item = ListenerUpgrade<O>, Error = IoError> + Send>;
pub type ListenerUpgrade<O> = Box<Future<Item = (O, MultiaddrFuture), Error = IoError> + Send>;
pub type Incoming<O> = Box<Future<Item = IncomingUpgrade<O>, Error = IoError> + Send>;
pub type IncomingUpgrade<O> = Box<Future<Item = (O, MultiaddrFuture), Error = IoError> + Send>;
pub type Dial<O> = Box<Future<Item = O, Error = IoError> + Send>;
pub type Listener<O> = Box<Stream<Item = (ListenerUpgrade<O>, Multiaddr), Error = IoError> + Send>;
pub type ListenerUpgrade<O> = Box<Future<Item = O, Error = IoError> + Send>;
pub type Incoming<O> = Box<Future<Item = (IncomingUpgrade<O>, Multiaddr), Error = IoError> + Send>;
pub type IncomingUpgrade<O> = Box<Future<Item = O, Error = IoError> + Send>;
trait Abstract<O> {
fn listen_on(&self, addr: Multiaddr) -> Result<(Listener<O>, Multiaddr), Multiaddr>;
@ -75,22 +57,19 @@ where
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
T::MultiaddrFuture: Send + 'static,
{
fn listen_on(&self, addr: Multiaddr) -> Result<(Listener<O>, Multiaddr), Multiaddr> {
let (listener, new_addr) =
Transport::listen_on(self.clone(), addr).map_err(|(_, addr)| addr)?;
let fut = listener.map(|upgrade| {
let fut = upgrade.map(|(out, addr)| (out, Box::new(addr) as MultiaddrFuture));
Box::new(fut) as ListenerUpgrade<O>
let fut = listener.map(|(upgrade, addr)| {
(Box::new(upgrade) as ListenerUpgrade<O>, addr)
});
Ok((Box::new(fut) as Box<_>, new_addr))
}
fn dial(&self, addr: Multiaddr) -> Result<Dial<O>, Multiaddr> {
let fut = Transport::dial(self.clone(), addr)
.map_err(|(_, addr)| addr)?
.map(|(out, addr)| (out, Box::new(addr) as MultiaddrFuture));
.map_err(|(_, addr)| addr)?;
Ok(Box::new(fut) as Box<_>)
}
@ -100,29 +79,6 @@ where
}
}
trait AbstractMuxed<O>: Abstract<O> {
fn next_incoming(&self) -> Incoming<O>;
}
impl<T, O> AbstractMuxed<O> for T
where
T: MuxedTransport<Output = O> + Clone + 'static,
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Incoming: Send + 'static,
T::IncomingUpgrade: Send + 'static,
{
fn next_incoming(&self) -> Incoming<O> {
let fut = MuxedTransport::next_incoming(self.clone()).map(|upgrade| {
let fut = upgrade.map(|(out, addr)| (out, Box::new(addr) as MultiaddrFuture));
Box::new(fut) as IncomingUpgrade<O>
});
Box::new(fut) as Box<_>
}
}
/// See the `Transport::boxed` method.
pub struct Boxed<O> {
inner: Arc<Abstract<O> + Send + Sync>,
@ -145,7 +101,6 @@ impl<O> Clone for Boxed<O> {
impl<O> Transport for Boxed<O> {
type Output = O;
type MultiaddrFuture = MultiaddrFuture;
type Listener = Listener<O>;
type ListenerUpgrade = ListenerUpgrade<O>;
type Dial = Dial<O>;
@ -171,62 +126,3 @@ impl<O> Transport for Boxed<O> {
self.inner.nat_traversal(server, observed)
}
}
/// See the `Transport::boxed_muxed` method.
pub struct BoxedMuxed<O> {
inner: Arc<AbstractMuxed<O> + Send + Sync>,
}
impl<O> fmt::Debug for BoxedMuxed<O> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "BoxedMuxedTransport")
}
}
impl<O> Clone for BoxedMuxed<O> {
#[inline]
fn clone(&self) -> Self {
BoxedMuxed {
inner: self.inner.clone(),
}
}
}
impl<O> Transport for BoxedMuxed<O> {
type Output = O;
type MultiaddrFuture = MultiaddrFuture;
type Listener = Listener<O>;
type ListenerUpgrade = ListenerUpgrade<O>;
type Dial = Dial<O>;
#[inline]
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
match self.inner.listen_on(addr) {
Ok(listen) => Ok(listen),
Err(addr) => Err((self, addr)),
}
}
#[inline]
fn dial(self, addr: Multiaddr) -> Result<Self::Dial, (Self, Multiaddr)> {
match self.inner.dial(addr) {
Ok(dial) => Ok(dial),
Err(addr) => Err((self, addr)),
}
}
#[inline]
fn nat_traversal(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.inner.nat_traversal(server, observed)
}
}
impl<O> MuxedTransport for BoxedMuxed<O> {
type Incoming = Incoming<O>;
type IncomingUpgrade = IncomingUpgrade<O>;
#[inline]
fn next_incoming(self) -> Self::Incoming {
self.inner.next_incoming()
}
}

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

@ -18,11 +18,9 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use either::{EitherListenStream, EitherListenUpgrade, EitherOutput};
use futures::{prelude::*, future};
use either::{EitherListenStream, EitherOutput, EitherFuture};
use multiaddr::Multiaddr;
use std::io::Error as IoError;
use transport::{MuxedTransport, Transport};
use transport::Transport;
/// Struct returned by `or_transport()`.
#[derive(Debug, Copy, Clone)]
@ -41,10 +39,8 @@ where
{
type Output = EitherOutput<A::Output, B::Output>;
type Listener = EitherListenStream<A::Listener, B::Listener>;
type ListenerUpgrade = EitherListenUpgrade<A::ListenerUpgrade, B::ListenerUpgrade>;
type MultiaddrFuture = future::Either<A::MultiaddrFuture, B::MultiaddrFuture>;
type Dial =
EitherListenUpgrade<<A::Dial as IntoFuture>::Future, <B::Dial as IntoFuture>::Future>;
type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
type Dial = EitherFuture<A::Dial, B::Dial>;
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
let (first, addr) = match self.0.listen_on(addr) {
@ -60,12 +56,12 @@ where
fn dial(self, addr: Multiaddr) -> Result<Self::Dial, (Self, Multiaddr)> {
let (first, addr) = match self.0.dial(addr) {
Ok(connec) => return Ok(EitherListenUpgrade::First(connec)),
Ok(connec) => return Ok(EitherFuture::First(connec)),
Err(err) => err,
};
match self.1.dial(addr) {
Ok(connec) => Ok(EitherListenUpgrade::Second(connec)),
Ok(connec) => Ok(EitherFuture::Second(connec)),
Err((second, addr)) => Err((OrTransport(first, second), addr)),
}
}
@ -80,33 +76,3 @@ where
self.1.nat_traversal(server, observed)
}
}
impl<A, B> MuxedTransport for OrTransport<A, B>
where
A: MuxedTransport,
B: MuxedTransport,
A::Incoming: Send + 'static, // TODO: meh :-/
B::Incoming: Send + 'static, // TODO: meh :-/
A::IncomingUpgrade: Send + 'static, // TODO: meh :-/
B::IncomingUpgrade: Send + 'static, // TODO: meh :-/
A::Output: 'static, // TODO: meh :-/
B::Output: 'static, // TODO: meh :-/
{
type Incoming = Box<Future<Item = Self::IncomingUpgrade, Error = IoError> + Send>;
type IncomingUpgrade =
Box<Future<Item = (EitherOutput<A::Output, B::Output>, Self::MultiaddrFuture), Error = IoError> + Send>;
#[inline]
fn next_incoming(self) -> Self::Incoming {
let first = self.0.next_incoming().map(|out| {
let fut = out.map(move |(v, addr)| (EitherOutput::First(v), future::Either::A(addr)));
Box::new(fut) as Box<Future<Item = _, Error = _> + Send>
});
let second = self.1.next_incoming().map(|out| {
let fut = out.map(move |(v, addr)| (EitherOutput::Second(v), future::Either::B(addr)));
Box::new(fut) as Box<Future<Item = _, Error = _> + Send>
});
let future = first.select(second).map(|(i, _)| i).map_err(|(e, _)| e);
Box::new(future) as Box<_>
}
}

19
core/src/transport/denied.rs
View File

@ -18,11 +18,9 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use futures::future;
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::io::{self, Cursor};
use transport::MuxedTransport;
use transport::Transport;
/// Dummy implementation of `Transport` that just denies every single attempt.
@ -32,10 +30,9 @@ pub struct DeniedTransport;
impl Transport for DeniedTransport {
// TODO: could use `!` for associated types once stable
type Output = Cursor<Vec<u8>>;
type MultiaddrFuture = Box<Future<Item = Multiaddr, Error = io::Error> + Send + Sync>;
type Listener = Box<Stream<Item = Self::ListenerUpgrade, Error = io::Error> + Send + Sync>;
type ListenerUpgrade = Box<Future<Item = (Self::Output, Self::MultiaddrFuture), Error = io::Error> + Send + Sync>;
type Dial = Box<Future<Item = (Self::Output, Self::MultiaddrFuture), Error = io::Error> + Send + Sync>;
type Listener = Box<Stream<Item = (Self::ListenerUpgrade, Multiaddr), Error = io::Error> + Send + Sync>;
type ListenerUpgrade = Box<Future<Item = Self::Output, Error = io::Error> + Send + Sync>;
type Dial = Box<Future<Item = Self::Output, Error = io::Error> + Send + Sync>;
#[inline]
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
@ -52,13 +49,3 @@ impl Transport for DeniedTransport {
None
}
}
impl MuxedTransport for DeniedTransport {
type Incoming = future::Empty<Self::IncomingUpgrade, io::Error>;
type IncomingUpgrade = future::Empty<(Self::Output, Self::MultiaddrFuture), io::Error>;
#[inline]
fn next_incoming(self) -> Self::Incoming {
future::empty()
}
}

16
core/src/transport/interruptible.rs
View File

@ -20,7 +20,7 @@
use futures::{future, prelude::*, sync::oneshot};
use std::io::{Error as IoError, ErrorKind as IoErrorKind};
use transport::{MuxedTransport, Transport};
use transport::Transport;
use Multiaddr;
/// See `Transport::interruptible`.
@ -46,7 +46,6 @@ where
T: Transport,
{
type Output = T::Output;
type MultiaddrFuture = T::MultiaddrFuture;
type Listener = T::Listener;
type ListenerUpgrade = T::ListenerUpgrade;
type Dial = InterruptibleDial<T::Dial>;
@ -78,19 +77,6 @@ where
}
}
impl<T> MuxedTransport for Interruptible<T>
where
T: MuxedTransport,
{
type Incoming = T::Incoming;
type IncomingUpgrade = T::IncomingUpgrade;
#[inline]
fn next_incoming(self) -> Self::Incoming {
self.transport.next_incoming()
}
}
/// Dropping this object interrupts the dialing of the corresponding `Interruptible`.
pub struct Interrupt {
_tx: oneshot::Sender<()>,

42
core/src/transport/map.rs
View File

@ -21,7 +21,7 @@
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::io::Error as IoError;
use transport::{MuxedTransport, Transport};
use transport::Transport;
use Endpoint;
/// See `Transport::map`.
@ -48,22 +48,21 @@ where
F: FnOnce(T::Output, Endpoint) -> D + Clone + Send + 'static, // TODO: 'static :-/
{
type Output = D;
type MultiaddrFuture = T::MultiaddrFuture;
type Listener = Box<Stream<Item = Self::ListenerUpgrade, Error = IoError> + Send>;
type ListenerUpgrade = Box<Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
type Dial = Box<Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
type Listener = Box<Stream<Item = (Self::ListenerUpgrade, Multiaddr), Error = IoError> + Send>;
type ListenerUpgrade = Box<Future<Item = Self::Output, Error = IoError> + Send>;
type Dial = Box<Future<Item = Self::Output, Error = IoError> + Send>;
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
let map = self.map;
match self.transport.listen_on(addr) {
Ok((stream, listen_addr)) => {
let stream = stream.map(move |future| {
let stream = stream.map(move |(future, addr)| {
let map = map.clone();
let future = future
.into_future()
.map(move |(output, addr)| (map(output, Endpoint::Listener), addr));
Box::new(future) as Box<_>
.map(move |output| map(output, Endpoint::Listener));
(Box::new(future) as Box<_>, addr)
});
Ok((Box::new(stream), listen_addr))
}
@ -78,7 +77,7 @@ where
Ok(future) => {
let future = future
.into_future()
.map(move |(output, addr)| (map(output, Endpoint::Dialer), addr));
.map(move |output| map(output, Endpoint::Dialer));
Ok(Box::new(future))
}
Err((transport, addr)) => Err((Map { transport, map }, addr)),
@ -90,28 +89,3 @@ where
self.transport.nat_traversal(server, observed)
}
}
impl<T, F, D> MuxedTransport for Map<T, F>
where
T: MuxedTransport + 'static, // TODO: 'static :-/
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
F: FnOnce(T::Output, Endpoint) -> D + Clone + Send + 'static, // TODO: 'static :-/
{
type Incoming = Box<Future<Item = Self::IncomingUpgrade, Error = IoError> + Send>;
type IncomingUpgrade = Box<Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
fn next_incoming(self) -> Self::Incoming {
let map = self.map;
let future = self.transport.next_incoming().map(move |upgrade| {
let future = upgrade.map(move |(output, addr)| {
(map(output, Endpoint::Listener), addr)
});
Box::new(future) as Box<_>
});
Box::new(future)
}
}

95
core/src/transport/map_err.rs
View File

@ -21,7 +21,7 @@
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::io::Error as IoError;
use transport::{MuxedTransport, Transport};
use transport::Transport;
/// See `Transport::map_err`.
#[derive(Debug, Copy, Clone)]
@ -44,7 +44,6 @@ where
F: FnOnce(IoError) -> IoError + Clone,
{
type Output = T::Output;
type MultiaddrFuture = T::MultiaddrFuture;
type Listener = MapErrListener<T, F>;
type ListenerUpgrade = MapErrListenerUpgrade<T, F>;
type Dial = MapErrDial<T, F>;
@ -76,23 +75,6 @@ where
}
}
impl<T, F> MuxedTransport for MapErr<T, F>
where
T: MuxedTransport,
F: FnOnce(IoError) -> IoError + Clone,
{
type Incoming = MapErrIncoming<T, F>;
type IncomingUpgrade = MapErrIncomingUpgrade<T, F>;
#[inline]
fn next_incoming(self) -> Self::Incoming {
MapErrIncoming {
inner: self.transport.next_incoming(),
map: Some(self.map),
}
}
}
/// Listening stream for `MapErr`.
pub struct MapErrListener<T, F>
where T: Transport {
@ -104,14 +86,14 @@ impl<T, F> Stream for MapErrListener<T, F>
where T: Transport,
F: FnOnce(IoError) -> IoError + Clone,
{
type Item = MapErrListenerUpgrade<T, F>;
type Item = (MapErrListenerUpgrade<T, F>, Multiaddr);
type Error = IoError;
#[inline]
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
match try_ready!(self.inner.poll()) {
Some(value) => Ok(Async::Ready(
Some(MapErrListenerUpgrade { inner: value, map: Some(self.map.clone()) }))),
Some((value, addr)) => Ok(Async::Ready(
Some((MapErrListenerUpgrade { inner: value, map: Some(self.map.clone()) }, addr)))),
None => Ok(Async::Ready(None))
}
}
@ -128,7 +110,7 @@ impl<T, F> Future for MapErrListenerUpgrade<T, F>
where T: Transport,
F: FnOnce(IoError) -> IoError,
{
type Item = (T::Output, T::MultiaddrFuture);
type Item = T::Output;
type Error = IoError;
#[inline]
@ -159,72 +141,7 @@ impl<T, F> Future for MapErrDial<T, F>
where T: Transport,
F: FnOnce(IoError) -> IoError,
{
type Item = (T::Output, T::MultiaddrFuture);
type Error = IoError;
#[inline]
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.inner.poll() {
Ok(Async::Ready(value)) => {
Ok(Async::Ready(value))
},
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => {
let map = self.map.take().expect("poll() called again after error");
Err(map(err))
}
}
}
}
/// Incoming future for `MapErr`.
pub struct MapErrIncoming<T, F>
where T: MuxedTransport
{
inner: T::Incoming,
map: Option<F>,
}
impl<T, F> Future for MapErrIncoming<T, F>
where T: MuxedTransport,
F: FnOnce(IoError) -> IoError,
{
type Item = MapErrIncomingUpgrade<T, F>;
type Error = IoError;
#[inline]
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.inner.poll() {
Ok(Async::Ready(value)) => {
let map = self.map.take().expect("poll() called again after error");
let value = MapErrIncomingUpgrade {
inner: value,
map: Some(map),
};
Ok(Async::Ready(value))
},
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => {
let map = self.map.take().expect("poll() called again after error");
Err(map(err))
}
}
}
}
/// Incoming upgrade future for `MapErr`.
pub struct MapErrIncomingUpgrade<T, F>
where T: MuxedTransport
{
inner: T::IncomingUpgrade,
map: Option<F>,
}
impl<T, F> Future for MapErrIncomingUpgrade<T, F>
where T: MuxedTransport,
F: FnOnce(IoError) -> IoError,
{
type Item = (T::Output, T::MultiaddrFuture);
type Item = T::Output;
type Error = IoError;
#[inline]

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

@ -21,7 +21,7 @@
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::io::Error as IoError;
use transport::{MuxedTransport, Transport};
use transport::Transport;
/// See `Transport::map_err_dial`.
#[derive(Debug, Copy, Clone)]
@ -45,10 +45,9 @@ where
F: FnOnce(IoError, Multiaddr) -> IoError + Clone + Send + 'static, // TODO: 'static :-/
{
type Output = T::Output;
type MultiaddrFuture = T::MultiaddrFuture;
type Listener = T::Listener;
type ListenerUpgrade = T::ListenerUpgrade;
type Dial = Box<Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
type Dial = Box<Future<Item = Self::Output, Error = IoError> + Send>;
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
match self.transport.listen_on(addr) {
@ -74,18 +73,3 @@ where
self.transport.nat_traversal(server, observed)
}
}
impl<T, F> MuxedTransport for MapErrDial<T, F>
where
T: MuxedTransport + 'static, // TODO: 'static :-/
T::Dial: Send,
F: FnOnce(IoError, Multiaddr) -> IoError + Clone + Send + 'static, // TODO: 'static :-/
{
type Incoming = T::Incoming;
type IncomingUpgrade = T::IncomingUpgrade;
#[inline]
fn next_incoming(self) -> Self::Incoming {
self.transport.next_incoming()
}
}

18
core/src/transport/memory.rs
View File

@ -52,10 +52,9 @@ impl<T> Clone for Dialer<T> {
impl<T: IntoBuf + Send + 'static> Transport for Dialer<T> {
type Output = Channel<T>;
type Listener = Box<Stream<Item=Self::ListenerUpgrade, Error=io::Error> + Send>;
type ListenerUpgrade = FutureResult<(Self::Output, Self::MultiaddrFuture), io::Error>;
type MultiaddrFuture = FutureResult<Multiaddr, io::Error>;
type Dial = Box<Future<Item=(Self::Output, Self::MultiaddrFuture), Error=io::Error> + Send>;
type Listener = Box<Stream<Item=(Self::ListenerUpgrade, Multiaddr), Error=io::Error> + Send>;
type ListenerUpgrade = FutureResult<Self::Output, io::Error>;
type Dial = Box<Future<Item=Self::Output, Error=io::Error> + Send>;
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
Err((self, addr))
@ -70,7 +69,7 @@ impl<T: IntoBuf + Send + 'static> Transport for Dialer<T> {
let a = Chan { incoming: a_rx, outgoing: b_tx };
let b = Chan { incoming: b_rx, outgoing: a_tx };
let future = self.0.send(b)
.map(move |_| (a.into(), future::ok(addr)))
.map(move |_| a.into())
.map_err(|_| io::ErrorKind::ConnectionRefused.into());
Ok(Box::new(future))
}
@ -95,10 +94,9 @@ impl<T> Clone for Listener<T> {
impl<T: IntoBuf + Send + 'static> Transport for Listener<T> {
type Output = Channel<T>;
type Listener = Box<Stream<Item=Self::ListenerUpgrade, Error=io::Error> + Send>;
type ListenerUpgrade = FutureResult<(Self::Output, Self::MultiaddrFuture), io::Error>;
type MultiaddrFuture = FutureResult<Multiaddr, io::Error>;
type Dial = Box<Future<Item=(Self::Output, Self::MultiaddrFuture), Error=io::Error> + Send>;
type Listener = Box<Stream<Item=(Self::ListenerUpgrade, Multiaddr), Error=io::Error> + Send>;
type ListenerUpgrade = FutureResult<Self::Output, io::Error>;
type Dial = Box<Future<Item=Self::Output, Error=io::Error> + Send>;
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
if !is_memory_addr(&addr) {
@ -108,7 +106,7 @@ impl<T: IntoBuf + Send + 'static> Transport for Listener<T> {
let receiver = self.0.clone();
let stream = stream::poll_fn(move || receiver.lock().poll())
.map(move |channel| {
future::ok((channel.into(), future::ok(addr.clone())))
(future::ok(channel.into()), addr.clone())
})
.map_err(|()| unreachable!());
Ok((Box::new(stream), addr2))

55
core/src/transport/mod.rs
View File

@ -31,6 +31,7 @@
use futures::prelude::*;
use multiaddr::Multiaddr;
use nodes::raw_swarm::ConnectedPoint;
use std::io::Error as IoError;
use tokio_io::{AsyncRead, AsyncWrite};
use upgrade::{ConnectionUpgrade, Endpoint};
@ -39,21 +40,16 @@ pub mod and_then;
pub mod boxed;
pub mod choice;
pub mod denied;
pub mod dummy;
pub mod interruptible;
pub mod map;
pub mod map_err;
pub mod map_err_dial;
pub mod memory;
pub mod muxed;
pub mod upgrade;
pub use self::boxed::BoxedMuxed;
pub use self::choice::OrTransport;
pub use self::denied::DeniedTransport;
pub use self::dummy::DummyMuxing;
pub use self::memory::connector;
pub use self::muxed::MuxedTransport;
pub use self::upgrade::UpgradedNode;
/// A transport is an object that can be used to produce connections by listening or dialing a
@ -75,19 +71,15 @@ pub trait Transport {
/// An item should be produced whenever a connection is received at the lowest level of the
/// transport stack. The item is a `Future` that is signalled once some pre-processing has
/// taken place, and that connection has been upgraded to the wanted protocols.
type Listener: Stream<Item = Self::ListenerUpgrade, Error = IoError>;
/// Future that produces the multiaddress of the remote.
type MultiaddrFuture: Future<Item = Multiaddr, Error = IoError>;
type Listener: Stream<Item = (Self::ListenerUpgrade, Multiaddr), Error = IoError>;
/// After a connection has been received, we may need to do some asynchronous pre-processing
/// on it (eg. an intermediary protocol negotiation). While this pre-processing takes place, we
/// want to be able to continue polling on the listener.
// TODO: we could move the `MultiaddrFuture` to the `Listener` trait
type ListenerUpgrade: Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError>;
type ListenerUpgrade: Future<Item = Self::Output, Error = IoError>;
/// A future which indicates that we are currently dialing to a peer.
type Dial: Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError>;
type Dial: Future<Item = Self::Output, Error = IoError>;
/// Listen on the given multiaddr. Returns a stream of incoming connections, plus a modified
/// version of the `Multiaddr`. This new `Multiaddr` is the one that that should be advertised
@ -132,27 +124,10 @@ pub trait Transport {
Self::Dial: Send + 'static,
Self::Listener: Send + 'static,
Self::ListenerUpgrade: Send + 'static,
Self::MultiaddrFuture: Send + 'static,
{
boxed::boxed(self)
}
/// Turns this `Transport` into an abstract boxed transport.
///
/// This is the version if the transport supports muxing.
#[inline]
fn boxed_muxed(self) -> boxed::BoxedMuxed<Self::Output>
where Self: Sized + MuxedTransport + Clone + Send + Sync + 'static,
Self::Dial: Send + 'static,
Self::Listener: Send + 'static,
Self::ListenerUpgrade: Send + 'static,
Self::MultiaddrFuture: Send + 'static,
Self::Incoming: Send + 'static,
Self::IncomingUpgrade: Send + 'static,
{
boxed::boxed_muxed(self)
}
/// Applies a function on the output of the `Transport`.
#[inline]
fn map<F, O>(self, map: F) -> map::Map<Self, F>
@ -207,7 +182,7 @@ pub trait Transport {
where
Self: Sized,
Self::Output: AsyncRead + AsyncWrite,
U: ConnectionUpgrade<Self::Output, Self::MultiaddrFuture>,
U: ConnectionUpgrade<Self::Output>,
{
UpgradedNode::new(self, upgrade)
}
@ -218,29 +193,15 @@ pub trait Transport {
/// > **Note**: The concept of an *upgrade* for example includes middlewares such *secio*
/// > (communication encryption), *multiplex*, but also a protocol handler.
#[inline]
fn and_then<C, F, O, Maf>(self, upgrade: C) -> and_then::AndThen<Self, C>
fn and_then<C, F, O>(self, upgrade: C) -> and_then::AndThen<Self, C>
where
Self: Sized,
C: FnOnce(Self::Output, Endpoint, Self::MultiaddrFuture) -> F + Clone + 'static,
F: Future<Item = (O, Maf), Error = IoError> + 'static,
Maf: Future<Item = Multiaddr, Error = IoError> + 'static,
C: FnOnce(Self::Output, ConnectedPoint) -> F + Clone + 'static,
F: Future<Item = O, Error = IoError> + 'static,
{
and_then::and_then(self, upgrade)
}
/// Builds a dummy implementation of `MuxedTransport` that uses this transport.
///
/// The resulting object will not actually use muxing. This means that dialing the same node
/// twice will result in two different connections instead of two substreams on the same
/// connection.
#[inline]
fn with_dummy_muxing(self) -> DummyMuxing<Self>
where
Self: Sized,
{
DummyMuxing::new(self)
}
/// Wraps around the `Transport` and makes it interruptible.
#[inline]
fn interruptible(self) -> (interruptible::Interruptible<Self>, interruptible::Interrupt)

92
core/src/transport/upgrade.rs
View File

@ -22,7 +22,7 @@ use futures::prelude::*;
use multiaddr::Multiaddr;
use std::io::Error as IoError;
use tokio_io::{AsyncRead, AsyncWrite};
use transport::{MuxedTransport, Transport};
use transport::Transport;
use upgrade::{apply, ConnectionUpgrade, Endpoint};
/// Implements the `Transport` trait. Dials or listens, then upgrades any dialed or received
@ -50,9 +50,8 @@ where
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::MultiaddrFuture: Send,
T::Output: Send + AsyncRead + AsyncWrite,
C: ConnectionUpgrade<T::Output, T::MultiaddrFuture> + Send + 'a,
C: ConnectionUpgrade<T::Output> + Send + 'a,
C::NamesIter: Send,
C::Future: Send,
C::UpgradeIdentifier: Send,
@ -72,7 +71,7 @@ where
pub fn dial(
self,
addr: Multiaddr,
) -> Result<Box<Future<Item = (C::Output, C::MultiaddrFuture), Error = IoError> + Send + 'a>, (Self, Multiaddr)>
) -> Result<Box<Future<Item = C::Output, Error = IoError> + Send + 'a>, (Self, Multiaddr)>
where
C::NamesIter: Clone, // TODO: not elegant
{
@ -92,48 +91,13 @@ where
let future = dialed_fut
// Try to negotiate the protocol.
.and_then(move |(connection, client_addr)| {
apply(connection, upgrade, Endpoint::Dialer, client_addr)
.and_then(move |connection| {
apply(connection, upgrade, Endpoint::Dialer)
});
Ok(Box::new(future))
}
/// If the underlying transport is a `MuxedTransport`, then after calling `dial` we may receive
/// substreams opened by the dialed nodes.
///
/// This function returns the next incoming substream. You are strongly encouraged to call it
/// if you have a muxed transport.
pub fn next_incoming(
self,
) -> Box<
Future<
Item = Box<Future<Item = (C::Output, C::MultiaddrFuture), Error = IoError> + Send + 'a>,
Error = IoError,
>
+ Send + 'a,
>
where
T: MuxedTransport,
T::Incoming: Send,
T::IncomingUpgrade: Send,
C::NamesIter: Clone, // TODO: not elegant
C: Clone,
{
let upgrade = self.upgrade;
let future = self.transports.next_incoming().map(|future| {
// Try to negotiate the protocol.
let future = future.and_then(move |(connection, client_addr)| {
apply(connection, upgrade, Endpoint::Listener, client_addr)
});
Box::new(future) as Box<Future<Item = _, Error = _> + Send>
});
Box::new(future) as Box<_>
}
/// Start listening on the multiaddr using the transport that was passed to `new`.
/// Then whenever a connection is opened, it is upgraded.
///
@ -147,7 +111,7 @@ where
(
Box<
Stream<
Item = Box<Future<Item = (C::Output, C::MultiaddrFuture), Error = IoError> + Send + 'a>,
Item = (Box<Future<Item = C::Output, Error = IoError> + Send + 'a>, Multiaddr),
Error = IoError,
>
+ Send
@ -179,15 +143,15 @@ where
// Note that failing to negotiate a protocol will never produce a future with an error.
// Instead the `stream` will produce `Ok(Err(...))`.
// `stream` can only produce an `Err` if `listening_stream` produces an `Err`.
let stream = listening_stream.map(move |connection| {
let stream = listening_stream.map(move |(connection, client_addr)| {
let upgrade = upgrade.clone();
let connection = connection
// Try to negotiate the protocol.
.and_then(move |(connection, client_addr)| {
apply(connection, upgrade, Endpoint::Listener, client_addr)
.and_then(move |connection| {
apply(connection, upgrade, Endpoint::Listener)
});
Box::new(connection) as Box<_>
(Box::new(connection) as Box<_>, client_addr)
});
Ok((Box::new(stream), new_addr))
@ -200,19 +164,16 @@ where
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::MultiaddrFuture: Send,
T::Output: Send + AsyncRead + AsyncWrite,
C: ConnectionUpgrade<T::Output, T::MultiaddrFuture> + Clone + Send + 'static,
C::MultiaddrFuture: Future<Item = Multiaddr, Error = IoError>,
C: ConnectionUpgrade<T::Output> + Clone + Send + 'static,
C::NamesIter: Clone + Send,
C::Future: Send,
C::UpgradeIdentifier: Send,
{
type Output = C::Output;
type MultiaddrFuture = C::MultiaddrFuture;
type Listener = Box<Stream<Item = Self::ListenerUpgrade, Error = IoError> + Send>;
type ListenerUpgrade = Box<Future<Item = (C::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
type Dial = Box<Future<Item = (C::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
type Listener = Box<Stream<Item = (Self::ListenerUpgrade, Multiaddr), Error = IoError> + Send>;
type ListenerUpgrade = Box<Future<Item = C::Output, Error = IoError> + Send>;
type Dial = Box<Future<Item = C::Output, Error = IoError> + Send>;
#[inline]
fn listen_on(self, addr: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
@ -229,28 +190,3 @@ where
self.transports.nat_traversal(server, observed)
}
}
impl<T, C> MuxedTransport for UpgradedNode<T, C>
where
T: MuxedTransport + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::MultiaddrFuture: Send,
T::Output: Send + AsyncRead + AsyncWrite,
T::Incoming: Send,
T::IncomingUpgrade: Send,
C: ConnectionUpgrade<T::Output, T::MultiaddrFuture> + Clone + Send + 'static,
C::MultiaddrFuture: Future<Item = Multiaddr, Error = IoError>,
C::NamesIter: Clone + Send,
C::Future: Send,
C::UpgradeIdentifier: Send,
{
type Incoming = Box<Future<Item = Self::IncomingUpgrade, Error = IoError> + Send>;
type IncomingUpgrade = Box<Future<Item = (C::Output, Self::MultiaddrFuture), Error = IoError> + Send>;
#[inline]
fn next_incoming(self) -> Self::Incoming {
self.next_incoming()
}
}

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

@ -29,9 +29,9 @@ use upgrade::{ConnectionUpgrade, Endpoint};
///
/// Returns a `Future` that returns the outcome of the connection upgrade.
#[inline]
pub fn apply<C, U, Maf>(conn: C, upgrade: U, e: Endpoint, remote: Maf) -> UpgradeApplyFuture<C, U, Maf>
pub fn apply<C, U>(conn: C, upgrade: U, e: Endpoint) -> UpgradeApplyFuture<C, U>
where
U: ConnectionUpgrade<C, Maf>,
U: ConnectionUpgrade<C>,
U::NamesIter: Clone, // TODO: not elegant
C: AsyncRead + AsyncWrite,
{
@ -40,31 +40,28 @@ where
future: negotiate(conn, &upgrade, e),
upgrade,
endpoint: e,
remote
}
}
}
/// Future, returned from `apply` which performs a connection upgrade.
pub struct UpgradeApplyFuture<C, U, Maf>
pub struct UpgradeApplyFuture<C, U>
where
U: ConnectionUpgrade<C, Maf>,
U: ConnectionUpgrade<C>,
C: AsyncRead + AsyncWrite
{
inner: UpgradeApplyState<C, U, Maf>
inner: UpgradeApplyState<C, U>
}
enum UpgradeApplyState<C, U, Maf>
enum UpgradeApplyState<C, U>
where
U: ConnectionUpgrade<C, Maf>,
U: ConnectionUpgrade<C>,
C: AsyncRead + AsyncWrite
{
Init {
future: NegotiationFuture<C, ProtocolNames<U::NamesIter>, U::UpgradeIdentifier>,
upgrade: U,
endpoint: Endpoint,
remote: Maf
endpoint: Endpoint
},
Upgrade {
future: U::Future
@ -72,28 +69,28 @@ where
Undefined
}
impl<C, U, Maf> Future for UpgradeApplyFuture<C, U, Maf>
impl<C, U> Future for UpgradeApplyFuture<C, U>
where
U: ConnectionUpgrade<C, Maf>,
U: ConnectionUpgrade<C>,
U::NamesIter: Clone,
C: AsyncRead + AsyncWrite
{
type Item = (U::Output, U::MultiaddrFuture);
type Item = U::Output;
type Error = IoError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
loop {
match mem::replace(&mut self.inner, UpgradeApplyState::Undefined) {
UpgradeApplyState::Init { mut future, upgrade, endpoint, remote } => {
UpgradeApplyState::Init { mut future, upgrade, endpoint } => {
let (upgrade_id, connection) = match future.poll()? {
Async::Ready(x) => x,
Async::NotReady => {
self.inner = UpgradeApplyState::Init { future, upgrade, endpoint, remote };
self.inner = UpgradeApplyState::Init { future, upgrade, endpoint };
return Ok(Async::NotReady)
}
};
self.inner = UpgradeApplyState::Upgrade {
future: upgrade.upgrade(connection, upgrade_id, endpoint, remote)
future: upgrade.upgrade(connection, upgrade_id, endpoint)
};
}
UpgradeApplyState::Upgrade { mut future } => {
@ -124,13 +121,13 @@ where
///
/// Returns a `Future` that returns the negotiated protocol and the stream.
#[inline]
pub fn negotiate<C, I, U, Maf>(
pub fn negotiate<C, I, U>(
connection: C,
upgrade: &U,
endpoint: Endpoint,
) -> NegotiationFuture<C, ProtocolNames<U::NamesIter>, U::UpgradeIdentifier>
where
U: ConnectionUpgrade<I, Maf>,
U: ConnectionUpgrade<I>,
U::NamesIter: Clone, // TODO: not elegant
C: AsyncRead + AsyncWrite,
{
@ -144,7 +141,6 @@ where
}
}
/// Future, returned by `negotiate`, which negotiates a protocol and stream.
pub struct NegotiationFuture<R: AsyncRead + AsyncWrite, I, P> {
inner: Either<ListenerSelectFuture<R, I, P>, DialerSelectFuture<R, I, P>>
@ -175,7 +171,6 @@ where
}
}
/// Iterator adapter which adds equality matching predicates to items.
/// Used in `NegotiationFuture`.
#[derive(Clone)]

54
core/src/upgrade/choice.rs
View File

@ -19,8 +19,7 @@
// DEALINGS IN THE SOFTWARE.
use bytes::Bytes;
use futures::{future, prelude::*};
use std::io::Error as IoError;
use futures::future;
use tokio_io::{AsyncRead, AsyncWrite};
use upgrade::{ConnectionUpgrade, Endpoint};
@ -37,11 +36,11 @@ pub fn or<A, B>(me: A, other: B) -> OrUpgrade<A, B> {
#[derive(Debug, Copy, Clone)]
pub struct OrUpgrade<A, B>(A, B);
impl<C, A, B, O, Maf> ConnectionUpgrade<C, Maf> for OrUpgrade<A, B>
impl<C, A, B, O> ConnectionUpgrade<C> for OrUpgrade<A, B>
where
C: AsyncRead + AsyncWrite,
A: ConnectionUpgrade<C, Maf, Output = O>,
B: ConnectionUpgrade<C, Maf, Output = O>,
A: ConnectionUpgrade<C, Output = O>,
B: ConnectionUpgrade<C, Output = O>,
{
type NamesIter = NamesIterChain<A::NamesIter, B::NamesIter>;
type UpgradeIdentifier = EitherUpgradeIdentifier<A::UpgradeIdentifier, B::UpgradeIdentifier>;
@ -55,8 +54,7 @@ where
}
type Output = O;
type MultiaddrFuture = future::Either<A::MultiaddrFuture, B::MultiaddrFuture>;
type Future = EitherConnUpgrFuture<A::Future, B::Future>;
type Future = future::Either<A::Future, B::Future>;
#[inline]
fn upgrade(
@ -64,14 +62,13 @@ where
socket: C,
id: Self::UpgradeIdentifier,
ty: Endpoint,
remote_addr: Maf,
) -> Self::Future {
match id {
EitherUpgradeIdentifier::First(id) => {
EitherConnUpgrFuture::First(self.0.upgrade(socket, id, ty, remote_addr))
future::Either::A(self.0.upgrade(socket, id, ty))
}
EitherUpgradeIdentifier::Second(id) => {
EitherConnUpgrFuture::Second(self.1.upgrade(socket, id, ty, remote_addr))
future::Either::B(self.1.upgrade(socket, id, ty))
}
}
}
@ -84,43 +81,6 @@ pub enum EitherUpgradeIdentifier<A, B> {
Second(B),
}
/// Implements `Future` and redirects calls to either `First` or `Second`.
///
/// Additionally, the output will be wrapped inside a `EitherOutput`.
///
// TODO: This type is needed because of the lack of `impl Trait` in stable Rust.
// If Rust had impl Trait we could use the Either enum from the futures crate and add some
// modifiers to it. This custom enum is a combination of Either and these modifiers.
#[derive(Debug, Copy, Clone)]
#[must_use = "futures do nothing unless polled"]
pub enum EitherConnUpgrFuture<A, B> {
First(A),
Second(B),
}
impl<A, B, O, Ma, Mb> Future for EitherConnUpgrFuture<A, B>
where
A: Future<Error = IoError, Item = (O, Ma)>,
B: Future<Error = IoError, Item = (O, Mb)>,
{
type Item = (O, future::Either<Ma, Mb>);
type Error = IoError;
#[inline]
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self {
&mut EitherConnUpgrFuture::First(ref mut a) => {
let (item, fut) = try_ready!(a.poll());
Ok(Async::Ready((item, future::Either::A(fut))))
}
&mut EitherConnUpgrFuture::Second(ref mut b) => {
let (item, fut) = try_ready!(b.poll());
Ok(Async::Ready((item, future::Either::B(fut))))
}
}
}
}
/// Internal type used by the `OrUpgrade` struct.
///
/// > **Note**: This type is needed because of the lack of `-> impl Trait` in Rust. It can be

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

@ -20,7 +20,6 @@
use bytes::Bytes;
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::{io, iter};
use tokio_io::{AsyncRead, AsyncWrite};
use upgrade::{ConnectionUpgrade, Endpoint};
@ -29,15 +28,14 @@ use upgrade::{ConnectionUpgrade, Endpoint};
#[derive(Debug, Copy, Clone)]
pub struct DeniedConnectionUpgrade;
impl<C, Maf> ConnectionUpgrade<C, Maf> for DeniedConnectionUpgrade
impl<C> ConnectionUpgrade<C> for DeniedConnectionUpgrade
where
C: AsyncRead + AsyncWrite,
{
type NamesIter = iter::Empty<(Bytes, ())>;
type UpgradeIdentifier = (); // TODO: could use `!`
type Output = (); // TODO: could use `!`
type MultiaddrFuture = Box<Future<Item = Multiaddr, Error = io::Error> + Send + Sync>; // TODO: could use `!`
type Future = Box<Future<Item = ((), Self::MultiaddrFuture), Error = io::Error> + Send + Sync>; // TODO: could use `!`
type Future = Box<Future<Item = (), Error = io::Error> + Send + Sync>; // TODO: could use `!`
#[inline]
fn protocol_names(&self) -> Self::NamesIter {
@ -45,7 +43,7 @@ where
}
#[inline]
fn upgrade(self, _: C, _: Self::UpgradeIdentifier, _: Endpoint, _: Maf) -> Self::Future {
fn upgrade(self, _: C, _: Self::UpgradeIdentifier, _: Endpoint) -> Self::Future {
unreachable!("the denied connection upgrade always fails to negotiate")
}
}

20
core/src/upgrade/loop_upg.rs
View File

@ -61,23 +61,20 @@ pub struct LoopUpg<Inner> {
}
// TODO: 'static :-/
impl<State, Socket, Inner, Out, AddrFut> ConnectionUpgrade<(State, Socket), AddrFut>
impl<State, Socket, Inner, Out> ConnectionUpgrade<(State, Socket)>
for LoopUpg<Inner>
where
State: Send + 'static,
Socket: AsyncRead + AsyncWrite + Send + 'static,
Inner: ConnectionUpgrade<
(State, Socket),
AddrFut,
Output = Loop<State, Socket, Out>,
MultiaddrFuture = AddrFut,
> + Clone
+ Send
+ 'static,
Inner::NamesIter: Clone + Send + 'static,
Inner::UpgradeIdentifier: Send,
Inner::Future: Send,
AddrFut: Send + 'static,
Out: Send + 'static,
{
type NamesIter = Inner::NamesIter;
@ -88,29 +85,27 @@ where
}
type Output = Out;
type MultiaddrFuture = AddrFut;
type Future = Box<Future<Item = (Out, Self::MultiaddrFuture), Error = IoError> + Send>;
type Future = Box<Future<Item = Out, Error = IoError> + Send>;
fn upgrade(
self,
(state, socket): (State, Socket),
id: Self::UpgradeIdentifier,
endpoint: Endpoint,
remote_addr: AddrFut,
) -> Self::Future {
let inner = self.inner;
let fut = future::loop_fn(
(state, socket, id, remote_addr, MAX_LOOPS),
move |(state, socket, id, remote_addr, loops_remaining)| {
(state, socket, id, MAX_LOOPS),
move |(state, socket, id, loops_remaining)| {
// When we enter a recursion of the `loop_fn`, a protocol has already been
// negotiated. So what we have to do is upgrade then negotiate the next protocol
// (if necessary), and then only continue iteration in the `future::loop_fn`.
let inner = inner.clone();
inner
.clone()
.upgrade((state, socket), id, endpoint, remote_addr)
.and_then(move |(loop_out, remote_addr)| match loop_out {
.upgrade((state, socket), id, endpoint)
.and_then(move |loop_out| match loop_out {
Loop::Continue(state, socket) => {
// Produce an error if we reached the recursion limit.
if loops_remaining == 0 {
@ -126,14 +121,13 @@ where
state,
socket,
id,
remote_addr,
loops_remaining - 1,
))
});
future::Either::A(fut)
}
Loop::Break(fin) => {
future::Either::B(future::ok(FutLoop::Break((fin, remote_addr))))
future::Either::B(future::ok(FutLoop::Break(fin)))
}
})
},

12
core/src/upgrade/map.rs
View File

@ -36,9 +36,9 @@ pub struct Map<U, F> {
map: F,
}
impl<C, U, F, O, Maf> ConnectionUpgrade<C, Maf> for Map<U, F>
impl<C, U, F, O> ConnectionUpgrade<C> for Map<U, F>
where
U: ConnectionUpgrade<C, Maf>,
U: ConnectionUpgrade<C>,
U::Future: Send + 'static, // TODO: 'static :(
C: AsyncRead + AsyncWrite,
F: FnOnce(U::Output) -> O + Send + 'static, // TODO: 'static :(
@ -51,20 +51,18 @@ where
}
type Output = O;
type MultiaddrFuture = U::MultiaddrFuture;
type Future = Box<Future<Item = (O, Self::MultiaddrFuture), Error = IoError> + Send>;
type Future = Box<Future<Item = O, Error = IoError> + Send>;
fn upgrade(
self,
socket: C,
id: Self::UpgradeIdentifier,
ty: Endpoint,
remote_addr: Maf,
) -> Self::Future {
let map = self.map;
let fut = self.upgrade
.upgrade(socket, id, ty, remote_addr)
.map(move |(out, maf)| (map(out), maf));
.upgrade(socket, id, ty)
.map(map);
Box::new(fut) as Box<_>
}
}

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

@ -23,7 +23,6 @@ pub mod choice;
pub mod denied;
pub mod loop_upg;
pub mod map;
pub mod map_addr;
pub mod plaintext;
pub mod toggleable;
pub mod traits;
@ -33,7 +32,6 @@ pub use self::choice::{or, OrUpgrade};
pub use self::denied::DeniedConnectionUpgrade;
pub use self::loop_upg::{loop_upg, Loop};
pub use self::map::map;
pub use self::map_addr::map_with_addr;
pub use self::plaintext::PlainTextConfig;
pub use self::toggleable::toggleable;
pub use self::traits::{ConnectionUpgrade, Endpoint};

9
core/src/upgrade/plaintext.rs
View File

@ -32,19 +32,18 @@ use upgrade::{ConnectionUpgrade, Endpoint};
#[derive(Debug, Copy, Clone)]
pub struct PlainTextConfig;
impl<C, F> ConnectionUpgrade<C, F> for PlainTextConfig
impl<C> ConnectionUpgrade<C> for PlainTextConfig
where
C: AsyncRead + AsyncWrite,
{
type Output = C;
type Future = FutureResult<(C, F), IoError>;
type Future = FutureResult<C, IoError>;
type UpgradeIdentifier = ();
type MultiaddrFuture = F;
type NamesIter = iter::Once<(Bytes, ())>;
#[inline]
fn upgrade(self, i: C, _: (), _: Endpoint, remote_addr: F) -> Self::Future {
future::ok((i, remote_addr))
fn upgrade(self, i: C, _: (), _: Endpoint) -> Self::Future {
future::ok(i)
}
#[inline]

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

@ -65,10 +65,10 @@ impl<U> Toggleable<U> {
}
}
impl<C, U, Maf> ConnectionUpgrade<C, Maf> for Toggleable<U>
impl<C, U> ConnectionUpgrade<C> for Toggleable<U>
where
C: AsyncRead + AsyncWrite,
U: ConnectionUpgrade<C, Maf>,
U: ConnectionUpgrade<C>,
{
type NamesIter = ToggleableIter<U::NamesIter>;
type UpgradeIdentifier = U::UpgradeIdentifier;
@ -82,8 +82,7 @@ where
}
type Output = U::Output;
type MultiaddrFuture = U::MultiaddrFuture;
type Future = future::Either<future::Empty<(U::Output, U::MultiaddrFuture), IoError>, U::Future>;
type Future = future::Either<future::Empty<U::Output, IoError>, U::Future>;
#[inline]
fn upgrade(
@ -91,10 +90,9 @@ where
socket: C,
id: Self::UpgradeIdentifier,
ty: Endpoint,
remote_addr: Maf,
) -> Self::Future {
if self.enabled {
future::Either::B(self.inner.upgrade(socket, id, ty, remote_addr))
future::Either::B(self.inner.upgrade(socket, id, ty))
} else {
future::Either::A(future::empty())
}

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

@ -50,7 +50,7 @@ impl Not for Endpoint {
/// > **Note**: The `upgrade` method of this trait uses `self` and not `&self` or `&mut self`.
/// > This has been designed so that you would implement this trait on `&Foo` or
/// > `&mut Foo` instead of directly on `Foo`.
pub trait ConnectionUpgrade<C, TAddrFut> {
pub trait ConnectionUpgrade<C> {
/// Iterator returned by `protocol_names`.
type NamesIter: Iterator<Item = (Bytes, Self::UpgradeIdentifier)>;
/// Type that serves as an identifier for the protocol. This type only exists to be returned
@ -68,10 +68,8 @@ pub trait ConnectionUpgrade<C, TAddrFut> {
/// > **Note**: For upgrades that add an intermediary layer (such as `secio` or `multiplex`),
/// > this associated type must implement `AsyncRead + AsyncWrite`.
type Output;
/// Type of the future that will resolve to the remote's multiaddr.
type MultiaddrFuture;
/// Type of the future that will resolve to `Self::Output`.
type Future: Future<Item = (Self::Output, Self::MultiaddrFuture), Error = IoError>;
type Future: Future<Item = Self::Output, Error = IoError>;
/// This method is called after protocol negotiation has been performed.
///
@ -82,6 +80,5 @@ pub trait ConnectionUpgrade<C, TAddrFut> {
socket: C,
id: Self::UpgradeIdentifier,
ty: Endpoint,
remote_addr: TAddrFut,
) -> Self::Future;
}