fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-28 01:01:34 +00:00
Code Issues Projects Releases Wiki Activity

core/src/transport: Poll Transport directly, remove Transport::Listener (#2652)

Browse Source 
Remove the concept of individual `Transport::Listener` streams from `Transport`.
Instead the `Transport` is polled directly via `Transport::poll`. The
`Transport` is now responsible for driving its listeners.
This commit is contained in:
Elena Frank
2022-07-04 04:16:57 +02:00
committed by GitHub
parent b28cdb31f9
commit 62622a1bad
59 changed files with 2094 additions and 2044 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
core/src/connection.rs
View File

@ -43,25 +43,6 @@ impl std::ops::Add<usize> for ConnectionId {
}
}
/// The ID of a single listener.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct ListenerId(u64);
impl ListenerId {
/// Creates a `ListenerId` from a non-negative integer.
pub fn new(id: u64) -> Self {
Self(id)
}
}
impl std::ops::Add<u64> for ListenerId {
type Output = Self;
fn add(self, other: u64) -> Self {
Self(self.0 + other)
}
}
/// The endpoint roles associated with a peer-to-peer communication channel.
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum Endpoint {

108
core/src/either.rs
View File

@ -20,7 +20,7 @@
use crate::{
muxing::{StreamMuxer, StreamMuxerEvent},
transport::{ListenerEvent, Transport, TransportError},
transport::{ListenerId, Transport, TransportError, TransportEvent},
Multiaddr, ProtocolName,
};
use futures::{
@ -274,48 +274,6 @@ pub enum EitherOutbound<A: StreamMuxer, B: StreamMuxer> {
B(B::OutboundSubstream),
}
/// Implements `Stream` and dispatches all method calls to either `First` or `Second`.
#[pin_project(project = EitherListenStreamProj)]
#[derive(Debug, Copy, Clone)]
#[must_use = "futures do nothing unless polled"]
pub enum EitherListenStream<A, B> {
First(#[pin] A),
Second(#[pin] B),
}
impl<AStream, BStream, AInner, BInner, AError, BError> Stream
for EitherListenStream<AStream, BStream>
where
AStream: TryStream<Ok = ListenerEvent<AInner, AError>, Error = AError>,
BStream: TryStream<Ok = ListenerEvent<BInner, BError>, Error = BError>,
{
type Item = Result<
ListenerEvent<EitherFuture<AInner, BInner>, EitherError<AError, BError>>,
EitherError<AError, BError>,
>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
match self.project() {
EitherListenStreamProj::First(a) => match TryStream::try_poll_next(a, cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(None) => Poll::Ready(None),
Poll::Ready(Some(Ok(le))) => Poll::Ready(Some(Ok(le
.map(EitherFuture::First)
.map_err(EitherError::A)))),
Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(EitherError::A(err)))),
},
EitherListenStreamProj::Second(a) => match TryStream::try_poll_next(a, cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(None) => Poll::Ready(None),
Poll::Ready(Some(Ok(le))) => Poll::Ready(Some(Ok(le
.map(EitherFuture::Second)
.map_err(EitherError::B)))),
Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(EitherError::B(err)))),
},
}
}
}
/// Implements `Future` and dispatches all method calls to either `First` or `Second`.
#[pin_project(project = EitherFutureProj)]
#[derive(Debug, Copy, Clone)]
@ -385,11 +343,12 @@ impl<A: ProtocolName, B: ProtocolName> ProtocolName for EitherName<A, B> {
}
}
}
#[derive(Debug, Copy, Clone)]
#[pin_project(project = EitherTransportProj)]
#[derive(Debug)]
#[must_use = "transports do nothing unless polled"]
pub enum EitherTransport<A, B> {
Left(A),
Right(B),
Left(#[pin] A),
Right(#[pin] B),
}
impl<A, B> Transport for EitherTransport<A, B>
@ -399,26 +358,51 @@ where
{
type Output = EitherOutput<A::Output, B::Output>;
type Error = EitherError<A::Error, B::Error>;
type Listener = EitherListenStream<A::Listener, B::Listener>;
type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
type Dial = EitherFuture<A::Dial, B::Dial>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
match self.project() {
EitherTransportProj::Left(a) => match a.poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(event) => Poll::Ready(
event
.map_upgrade(EitherFuture::First)
.map_err(EitherError::A),
),
},
EitherTransportProj::Right(b) => match b.poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(event) => Poll::Ready(
event
.map_upgrade(EitherFuture::Second)
.map_err(EitherError::B),
),
},
}
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
match self {
EitherTransport::Left(t) => t.remove_listener(id),
EitherTransport::Right(t) => t.remove_listener(id),
}
}
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
use TransportError::*;
match self {
EitherTransport::Left(a) => match a.listen_on(addr) {
Ok(listener) => Ok(EitherListenStream::First(listener)),
Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
Err(Other(err)) => Err(Other(EitherError::A(err))),
},
EitherTransport::Right(b) => match b.listen_on(addr) {
Ok(listener) => Ok(EitherListenStream::Second(listener)),
Err(MultiaddrNotSupported(addr)) => Err(MultiaddrNotSupported(addr)),
Err(Other(err)) => Err(Other(EitherError::B(err))),
},
EitherTransport::Left(a) => a.listen_on(addr).map_err(|e| match e {
MultiaddrNotSupported(addr) => MultiaddrNotSupported(addr),
Other(err) => Other(EitherError::A(err)),
}),
EitherTransport::Right(b) => b.listen_on(addr).map_err(|e| match e {
MultiaddrNotSupported(addr) => MultiaddrNotSupported(addr),
Other(err) => Other(EitherError::B(err)),
}),
}
}

386
core/src/transport.rs
View File

@ -25,10 +25,14 @@
//! any desired protocols. The rest of the module defines combinators for
//! modifying a transport through composition with other transports or protocol upgrades.
use crate::connection::ConnectedPoint;
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::{error::Error, fmt};
use std::{
error::Error,
fmt,
pin::Pin,
task::{Context, Poll},
};
pub mod and_then;
pub mod choice;
@ -42,6 +46,8 @@ pub mod upgrade;
mod boxed;
mod optional;
use crate::ConnectedPoint;
pub use self::boxed::Boxed;
pub use self::choice::OrTransport;
pub use self::memory::MemoryTransport;
@ -87,21 +93,8 @@ pub trait Transport {
/// An error that occurred during connection setup.
type Error: Error;
/// A stream of [`Output`](Transport::Output)s for inbound connections.
///
/// An item should be produced whenever a connection is received at the lowest level of the
/// transport stack. The item must be a [`ListenerUpgrade`](Transport::ListenerUpgrade) future
/// that resolves to an [`Output`](Transport::Output) value once all protocol upgrades
/// have been applied.
///
/// If this stream produces an error, it is considered fatal and the listener is killed. It
/// is possible to report non-fatal errors by producing a [`ListenerEvent::Error`].
type Listener: Stream<
Item = Result<ListenerEvent<Self::ListenerUpgrade, Self::Error>, Self::Error>,
>;
/// A pending [`Output`](Transport::Output) for an inbound connection,
/// obtained from the [`Listener`](Transport::Listener) stream.
/// obtained from the [`Transport`] stream.
///
/// After a connection has been accepted by the transport, it may need to go through
/// asynchronous post-processing (i.e. protocol upgrade negotiations). Such
@ -115,22 +108,20 @@ pub trait Transport {
/// obtained from [dialing](Transport::dial).
type Dial: Future<Output = Result<Self::Output, Self::Error>>;
/// Listens on the given [`Multiaddr`], producing a stream of pending, inbound connections
/// and addresses this transport is listening on (cf. [`ListenerEvent`]).
/// Listens on the given [`Multiaddr`] for inbound connections.
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>>;
/// Remove a listener.
///
/// Returning an error from the stream is considered fatal. The listener can also report
/// non-fatal errors by producing a [`ListenerEvent::Error`].
fn listen_on(&mut self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>>
where
Self: Sized;
/// Return `true` if there was a listener with this Id, `false`
/// otherwise.
fn remove_listener(&mut self, id: ListenerId) -> bool;
/// Dials the given [`Multiaddr`], returning a future for a pending outbound connection.
///
/// If [`TransportError::MultiaddrNotSupported`] is returned, it may be desirable to
/// try an alternative [`Transport`], if available.
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>>
where
Self: Sized;
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>>;
/// As [`Transport::dial`] but has the local node act as a listener on the outgoing connection.
///
@ -140,9 +131,23 @@ pub trait Transport {
fn dial_as_listener(
&mut self,
addr: Multiaddr,
) -> Result<Self::Dial, TransportError<Self::Error>>
where
Self: Sized;
) -> Result<Self::Dial, TransportError<Self::Error>>;
/// Poll for [`TransportEvent`]s.
///
/// A [`TransportEvent::Incoming`] should be produced whenever a connection is received at the lowest
/// level of the transport stack. The item must be a [`ListenerUpgrade`](Transport::ListenerUpgrade)
/// future that resolves to an [`Output`](Transport::Output) value once all protocol upgrades have
/// been applied.
///
/// Transports are expected to produce [`TransportEvent::Incoming`] events only for
/// listen addresses which have previously been announced via
/// a [`TransportEvent::NewAddress`] event and which have not been invalidated by
/// an [`TransportEvent::AddressExpired`] event yet.
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>>;
/// Performs a transport-specific mapping of an address `observed` by
/// a remote onto a local `listen` address to yield an address for
@ -152,9 +157,8 @@ pub trait Transport {
/// Boxes the transport, including custom transport errors.
fn boxed(self) -> boxed::Boxed<Self::Output>
where
Self: Transport + Sized + Send + 'static,
Self: Sized + Send + Unpin + 'static,
Self::Dial: Send + 'static,
Self::Listener: Send + 'static,
Self::ListenerUpgrade: Send + 'static,
Self::Error: Send + Sync,
{
@ -221,149 +225,277 @@ pub trait Transport {
}
}
/// Event produced by [`Transport::Listener`]s.
///
/// Transports are expected to produce `Upgrade` events only for
/// listen addresses which have previously been announced via
/// a `NewAddress` event and which have not been invalidated by
/// an `AddressExpired` event yet.
#[derive(Clone, Debug, PartialEq)]
pub enum ListenerEvent<TUpgr, TErr> {
/// The transport is listening on a new additional [`Multiaddr`].
NewAddress(Multiaddr),
/// An upgrade, consisting of the upgrade future, the listener address and the remote address.
Upgrade {
/// The upgrade.
upgrade: TUpgr,
/// The local address which produced this upgrade.
local_addr: Multiaddr,
/// The remote address which produced this upgrade.
remote_addr: Multiaddr,
},
/// A [`Multiaddr`] is no longer used for listening.
AddressExpired(Multiaddr),
/// A non-fatal error has happened on the listener.
///
/// This event should be generated in order to notify the user that something wrong has
/// happened. The listener, however, continues to run.
Error(TErr),
/// The ID of a single listener.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct ListenerId(u64);
impl ListenerId {
/// Creates a new `ListenerId`.
pub fn new() -> Self {
ListenerId(rand::random())
}
}
impl<TUpgr, TErr> ListenerEvent<TUpgr, TErr> {
/// In case this [`ListenerEvent`] is an upgrade, apply the given function
/// to the upgrade and multiaddress and produce another listener event
/// based the the function's result.
pub fn map<U>(self, f: impl FnOnce(TUpgr) -> U) -> ListenerEvent<U, TErr> {
match self {
ListenerEvent::Upgrade {
upgrade,
local_addr,
remote_addr,
} => ListenerEvent::Upgrade {
upgrade: f(upgrade),
local_addr,
remote_addr,
},
ListenerEvent::NewAddress(a) => ListenerEvent::NewAddress(a),
ListenerEvent::AddressExpired(a) => ListenerEvent::AddressExpired(a),
ListenerEvent::Error(e) => ListenerEvent::Error(e),
}
impl Default for ListenerId {
fn default() -> Self {
Self::new()
}
}
/// In case this [`ListenerEvent`] is an [`Error`](ListenerEvent::Error),
/// apply the given function to the error and produce another listener event based on the
/// function's result.
pub fn map_err<U>(self, f: impl FnOnce(TErr) -> U) -> ListenerEvent<TUpgr, U> {
match self {
ListenerEvent::Upgrade {
upgrade,
local_addr,
remote_addr,
} => ListenerEvent::Upgrade {
upgrade,
local_addr,
remote_addr,
},
ListenerEvent::NewAddress(a) => ListenerEvent::NewAddress(a),
ListenerEvent::AddressExpired(a) => ListenerEvent::AddressExpired(a),
ListenerEvent::Error(e) => ListenerEvent::Error(f(e)),
}
}
/// Returns `true` if this is an `Upgrade` listener event.
pub fn is_upgrade(&self) -> bool {
matches!(self, ListenerEvent::Upgrade { .. })
}
/// Try to turn this listener event into upgrade parts.
/// Event produced by [`Transport`]s.
pub enum TransportEvent<TUpgr, TErr> {
/// A new address is being listened on.
NewAddress {
/// The listener that is listening on the new address.
listener_id: ListenerId,
/// The new address that is being listened on.
listen_addr: Multiaddr,
},
/// An address is no longer being listened on.
AddressExpired {
/// The listener that is no longer listening on the address.
listener_id: ListenerId,
/// The new address that is being listened on.
listen_addr: Multiaddr,
},
/// A connection is incoming on one of the listeners.
Incoming {
/// The listener that produced the upgrade.
listener_id: ListenerId,
/// The produced upgrade.
upgrade: TUpgr,
/// Local connection address.
local_addr: Multiaddr,
/// Address used to send back data to the incoming client.
send_back_addr: Multiaddr,
},
/// A listener closed.
ListenerClosed {
/// The ID of the listener that closed.
listener_id: ListenerId,
/// Reason for the closure. Contains `Ok(())` if the stream produced `None`, or `Err`
/// if the stream produced an error.
reason: Result<(), TErr>,
},
/// A listener errored.
///
/// Returns `None` if the event is not actually an upgrade,
/// The listener will continue to be polled for new events and the event
/// is for informational purposes only.
ListenerError {
/// The ID of the listener that errored.
listener_id: ListenerId,
/// The error value.
error: TErr,
},
}
impl<TUpgr, TErr> TransportEvent<TUpgr, TErr> {
/// In case this [`TransportEvent`] is an upgrade, apply the given function
/// to the upgrade and produce another transport event based the the function's result.
pub fn map_upgrade<U>(self, map: impl FnOnce(TUpgr) -> U) -> TransportEvent<U, TErr> {
match self {
TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
} => TransportEvent::Incoming {
listener_id,
upgrade: map(upgrade),
local_addr,
send_back_addr,
},
TransportEvent::NewAddress {
listen_addr,
listener_id,
} => TransportEvent::NewAddress {
listen_addr,
listener_id,
},
TransportEvent::AddressExpired {
listen_addr,
listener_id,
} => TransportEvent::AddressExpired {
listen_addr,
listener_id,
},
TransportEvent::ListenerError { listener_id, error } => {
TransportEvent::ListenerError { listener_id, error }
}
TransportEvent::ListenerClosed {
listener_id,
reason,
} => TransportEvent::ListenerClosed {
listener_id,
reason,
},
}
}
/// In case this [`TransportEvent`] is an [`ListenerError`](TransportEvent::ListenerError),
/// or [`ListenerClosed`](TransportEvent::ListenerClosed) apply the given function to the
/// error and produce another transport event based on the function's result.
pub fn map_err<E>(self, map_err: impl FnOnce(TErr) -> E) -> TransportEvent<TUpgr, E> {
match self {
TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
} => TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
},
TransportEvent::NewAddress {
listen_addr,
listener_id,
} => TransportEvent::NewAddress {
listen_addr,
listener_id,
},
TransportEvent::AddressExpired {
listen_addr,
listener_id,
} => TransportEvent::AddressExpired {
listen_addr,
listener_id,
},
TransportEvent::ListenerError { listener_id, error } => TransportEvent::ListenerError {
listener_id,
error: map_err(error),
},
TransportEvent::ListenerClosed {
listener_id,
reason,
} => TransportEvent::ListenerClosed {
listener_id,
reason: reason.map_err(map_err),
},
}
}
/// Returns `true` if this is an [`Incoming`](TransportEvent::Incoming) transport event.
pub fn is_upgrade(&self) -> bool {
matches!(self, TransportEvent::Incoming { .. })
}
/// Try to turn this transport event into the upgrade parts of the
/// incoming connection.
///
/// Returns `None` if the event is not actually an incoming connection,
/// otherwise the upgrade and the remote address.
pub fn into_upgrade(self) -> Option<(TUpgr, Multiaddr)> {
if let ListenerEvent::Upgrade {
pub fn into_incoming(self) -> Option<(TUpgr, Multiaddr)> {
if let TransportEvent::Incoming {
upgrade,
remote_addr,
send_back_addr,
..
} = self
{
Some((upgrade, remote_addr))
Some((upgrade, send_back_addr))
} else {
None
}
}
/// Returns `true` if this is a `NewAddress` listener event.
/// Returns `true` if this is a [`TransportEvent::NewAddress`].
pub fn is_new_address(&self) -> bool {
matches!(self, ListenerEvent::NewAddress(_))
matches!(self, TransportEvent::NewAddress { .. })
}
/// Try to turn this listener event into the `NewAddress` part.
/// Try to turn this transport event into the new `Multiaddr`.
///
/// Returns `None` if the event is not actually a `NewAddress`,
/// Returns `None` if the event is not actually a [`TransportEvent::NewAddress`],
/// otherwise the address.
pub fn into_new_address(self) -> Option<Multiaddr> {
if let ListenerEvent::NewAddress(a) = self {
Some(a)
if let TransportEvent::NewAddress { listen_addr, .. } = self {
Some(listen_addr)
} else {
None
}
}
/// Returns `true` if this is an `AddressExpired` listener event.
/// Returns `true` if this is an [`TransportEvent::AddressExpired`].
pub fn is_address_expired(&self) -> bool {
matches!(self, ListenerEvent::AddressExpired(_))
matches!(self, TransportEvent::AddressExpired { .. })
}
/// Try to turn this listener event into the `AddressExpired` part.
/// Try to turn this transport event into the expire `Multiaddr`.
///
/// Returns `None` if the event is not actually a `AddressExpired`,
/// Returns `None` if the event is not actually a [`TransportEvent::AddressExpired`],
/// otherwise the address.
pub fn into_address_expired(self) -> Option<Multiaddr> {
if let ListenerEvent::AddressExpired(a) = self {
Some(a)
if let TransportEvent::AddressExpired { listen_addr, .. } = self {
Some(listen_addr)
} else {
None
}
}
/// Returns `true` if this is an `Error` listener event.
pub fn is_error(&self) -> bool {
matches!(self, ListenerEvent::Error(_))
/// Returns `true` if this is an [`TransportEvent::ListenerError`] transport event.
pub fn is_listener_error(&self) -> bool {
matches!(self, TransportEvent::ListenerError { .. })
}
/// Try to turn this listener event into the `Error` part.
/// Try to turn this transport event into the listener error.
///
/// Returns `None` if the event is not actually a `Error`,
/// Returns `None` if the event is not actually a [`TransportEvent::ListenerError`]`,
/// otherwise the error.
pub fn into_error(self) -> Option<TErr> {
if let ListenerEvent::Error(err) = self {
Some(err)
pub fn into_listener_error(self) -> Option<TErr> {
if let TransportEvent::ListenerError { error, .. } = self {
Some(error)
} else {
None
}
}
}
impl<TUpgr, TErr: fmt::Debug> fmt::Debug for TransportEvent<TUpgr, TErr> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
match self {
TransportEvent::NewAddress {
listener_id,
listen_addr,
} => f
.debug_struct("TransportEvent::NewAddress")
.field("listener_id", listener_id)
.field("listen_addr", listen_addr)
.finish(),
TransportEvent::AddressExpired {
listener_id,
listen_addr,
} => f
.debug_struct("TransportEvent::AddressExpired")
.field("listener_id", listener_id)
.field("listen_addr", listen_addr)
.finish(),
TransportEvent::Incoming {
listener_id,
local_addr,
..
} => f
.debug_struct("TransportEvent::Incoming")
.field("listener_id", listener_id)
.field("local_addr", local_addr)
.finish(),
TransportEvent::ListenerClosed {
listener_id,
reason,
} => f
.debug_struct("TransportEvent::Closed")
.field("listener_id", listener_id)
.field("reason", reason)
.finish(),
TransportEvent::ListenerError { listener_id, error } => f
.debug_struct("TransportEvent::ListenerError")
.field("listener_id", listener_id)
.field("error", error)
.finish(),
}
}
}
/// An error during [dialing][Transport::dial] or [listening][Transport::listen_on]
/// on a [`Transport`].
#[derive(Debug, Clone)]

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

@ -21,15 +21,17 @@
use crate::{
connection::{ConnectedPoint, Endpoint},
either::EitherError,
transport::{ListenerEvent, Transport, TransportError},
transport::{ListenerId, Transport, TransportError, TransportEvent},
};
use futures::{future::Either, prelude::*};
use multiaddr::Multiaddr;
use std::{error, marker::PhantomPinned, pin::Pin, task::Context, task::Poll};
/// See the `Transport::and_then` method.
/// See the [`Transport::and_then`] method.
#[pin_project::pin_project]
#[derive(Debug, Clone)]
pub struct AndThen<T, C> {
#[pin]
transport: T,
fun: C,
}
@ -49,27 +51,17 @@ where
{
type Output = O;
type Error = EitherError<T::Error, F::Error>;
type Listener = AndThenStream<T::Listener, C>;
type ListenerUpgrade = AndThenFuture<T::ListenerUpgrade, C, F>;
type Dial = AndThenFuture<T::Dial, C, F>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
let listener = self
.transport
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.transport
.listen_on(addr)
.map_err(|err| err.map(EitherError::A))?;
// 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 = AndThenStream {
stream: listener,
fun: self.fun.clone(),
};
Ok(stream)
.map_err(|err| err.map(EitherError::A))
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.transport.remove_listener(id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
@ -116,68 +108,40 @@ where
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.transport.address_translation(server, observed)
}
}
/// Custom `Stream` to avoid boxing.
///
/// Applies a function to every stream item.
#[pin_project::pin_project]
#[derive(Debug, Clone)]
pub struct AndThenStream<TListener, TMap> {
#[pin]
stream: TListener,
fun: TMap,
}
impl<TListener, TMap, TTransOut, TMapOut, TListUpgr, TTransErr> Stream
for AndThenStream<TListener, TMap>
where
TListener: TryStream<Ok = ListenerEvent<TListUpgr, TTransErr>, Error = TTransErr>,
TListUpgr: TryFuture<Ok = TTransOut, Error = TTransErr>,
TMap: FnOnce(TTransOut, ConnectedPoint) -> TMapOut + Clone,
TMapOut: TryFuture,
{
type Item = Result<
ListenerEvent<
AndThenFuture<TListUpgr, TMap, TMapOut>,
EitherError<TTransErr, TMapOut::Error>,
>,
EitherError<TTransErr, TMapOut::Error>,
>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
let this = self.project();
match TryStream::try_poll_next(this.stream, cx) {
Poll::Ready(Some(Ok(event))) => {
let event = match event {
ListenerEvent::Upgrade {
upgrade,
local_addr,
remote_addr,
} => {
let point = ConnectedPoint::Listener {
local_addr: local_addr.clone(),
send_back_addr: remote_addr.clone(),
};
ListenerEvent::Upgrade {
upgrade: AndThenFuture {
inner: Either::Left(Box::pin(upgrade)),
args: Some((this.fun.clone(), point)),
_marker: PhantomPinned,
},
local_addr,
remote_addr,
}
}
ListenerEvent::NewAddress(a) => ListenerEvent::NewAddress(a),
ListenerEvent::AddressExpired(a) => ListenerEvent::AddressExpired(a),
ListenerEvent::Error(e) => ListenerEvent::Error(EitherError::A(e)),
match this.transport.poll(cx) {
Poll::Ready(TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
}) => {
let point = ConnectedPoint::Listener {
local_addr: local_addr.clone(),
send_back_addr: send_back_addr.clone(),
};
Poll::Ready(Some(Ok(event)))
Poll::Ready(TransportEvent::Incoming {
listener_id,
upgrade: AndThenFuture {
inner: Either::Left(Box::pin(upgrade)),
args: Some((this.fun.clone(), point)),
_marker: PhantomPinned,
},
local_addr,
send_back_addr,
})
}
Poll::Ready(other) => {
let mapped = other
.map_upgrade(|_upgrade| unreachable!("case already matched"))
.map_err(EitherError::A);
Poll::Ready(mapped)
}
Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(EitherError::A(err)))),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
}
}

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

@ -18,18 +18,22 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::transport::{ListenerEvent, Transport, TransportError};
use futures::prelude::*;
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use futures::{prelude::*, stream::FusedStream};
use multiaddr::Multiaddr;
use std::{error::Error, fmt, io, pin::Pin};
use std::{
error::Error,
fmt, io,
pin::Pin,
task::{Context, Poll},
};
/// Creates a new [`Boxed`] transport from the given transport.
pub fn boxed<T>(transport: T) -> Boxed<T::Output>
where
T: Transport + Send + 'static,
T: Transport + Send + Unpin + 'static,
T::Error: Send + Sync,
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
{
Boxed {
@ -41,19 +45,22 @@ where
/// and `ListenerUpgrade` futures are `Box`ed and only the `Output`
/// and `Error` types are captured in type variables.
pub struct Boxed<O> {
inner: Box<dyn Abstract<O> + Send>,
inner: Box<dyn Abstract<O> + Send + Unpin>,
}
type Dial<O> = Pin<Box<dyn Future<Output = io::Result<O>> + Send>>;
type Listener<O> =
Pin<Box<dyn Stream<Item = io::Result<ListenerEvent<ListenerUpgrade<O>, io::Error>>> + Send>>;
type ListenerUpgrade<O> = Pin<Box<dyn Future<Output = io::Result<O>> + Send>>;
trait Abstract<O> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<Listener<O>, TransportError<io::Error>>;
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<io::Error>>;
fn remove_listener(&mut self, id: ListenerId) -> bool;
fn dial(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>>;
fn dial_as_listener(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>>;
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr>;
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<ListenerUpgrade<O>, io::Error>>;
}
impl<T, O> Abstract<O> for T
@ -61,22 +68,14 @@ where
T: Transport<Output = O> + 'static,
T::Error: Send + Sync,
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
{
fn listen_on(&mut self, addr: Multiaddr) -> Result<Listener<O>, TransportError<io::Error>> {
let listener = Transport::listen_on(self, addr).map_err(|e| e.map(box_err))?;
let fut = listener
.map_ok(|event| {
event
.map(|upgrade| {
let up = upgrade.map_err(box_err);
Box::pin(up) as ListenerUpgrade<O>
})
.map_err(box_err)
})
.map_err(box_err);
Ok(Box::pin(fut))
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<io::Error>> {
Transport::listen_on(self, addr).map_err(|e| e.map(box_err))
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
Transport::remove_listener(self, id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Dial<O>, TransportError<io::Error>> {
@ -96,6 +95,20 @@ where
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
Transport::address_translation(self, server, observed)
}
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<ListenerUpgrade<O>, io::Error>> {
self.poll(cx).map(|event| {
event
.map_upgrade(|upgrade| {
let up = upgrade.map_err(box_err);
Box::pin(up) as ListenerUpgrade<O>
})
.map_err(box_err)
})
}
}
impl<O> fmt::Debug for Boxed<O> {
@ -107,17 +120,17 @@ impl<O> fmt::Debug for Boxed<O> {
impl<O> Transport for Boxed<O> {
type Output = O;
type Error = io::Error;
type Listener = Listener<O>;
type ListenerUpgrade = ListenerUpgrade<O>;
type Dial = Dial<O>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.inner.listen_on(addr)
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.inner.remove_listener(id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
self.inner.dial(addr)
}
@ -132,6 +145,27 @@ impl<O> Transport for Boxed<O> {
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.inner.address_translation(server, observed)
}
fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
Pin::new(self.inner.as_mut()).poll(cx)
}
}
impl<O> Stream for Boxed<O> {
type Item = TransportEvent<ListenerUpgrade<O>, io::Error>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Transport::poll(self, cx).map(Some)
}
}
impl<O> FusedStream for Boxed<O> {
fn is_terminated(&self) -> bool {
false
}
}
fn box_err<E: Error + Send + Sync + 'static>(e: E) -> io::Error {

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

@ -18,13 +18,15 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::either::{EitherError, EitherFuture, EitherListenStream, EitherOutput};
use crate::transport::{Transport, TransportError};
use crate::either::{EitherError, EitherFuture, EitherOutput};
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use multiaddr::Multiaddr;
use std::{pin::Pin, task::Context, task::Poll};
/// Struct returned by `or_transport()`.
#[derive(Debug, Copy, Clone)]
pub struct OrTransport<A, B>(A, B);
#[pin_project::pin_project]
pub struct OrTransport<A, B>(#[pin] A, #[pin] B);
impl<A, B> OrTransport<A, B> {
pub fn new(a: A, b: B) -> OrTransport<A, B> {
@ -39,33 +41,27 @@ where
{
type Output = EitherOutput<A::Output, B::Output>;
type Error = EitherError<A::Error, B::Error>;
type Listener = EitherListenStream<A::Listener, B::Listener>;
type ListenerUpgrade = EitherFuture<A::ListenerUpgrade, B::ListenerUpgrade>;
type Dial = EitherFuture<A::Dial, B::Dial>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
let addr = match self.0.listen_on(addr) {
Ok(listener) => return Ok(EitherListenStream::First(listener)),
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
Err(TransportError::Other(err)) => {
return Err(TransportError::Other(EitherError::A(err)))
}
res => return res.map_err(|err| err.map(EitherError::A)),
};
let addr = match self.1.listen_on(addr) {
Ok(listener) => return Ok(EitherListenStream::Second(listener)),
Err(TransportError::MultiaddrNotSupported(addr)) => addr,
Err(TransportError::Other(err)) => {
return Err(TransportError::Other(EitherError::B(err)))
}
res => return res.map_err(|err| err.map(EitherError::B)),
};
Err(TransportError::MultiaddrNotSupported(addr))
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.0.remove_listener(id) || self.1.remove_listener(id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
let addr = match self.0.dial(addr) {
Ok(connec) => return Ok(EitherFuture::First(connec)),
@ -116,4 +112,24 @@ where
self.1.address_translation(server, observed)
}
}
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
let this = self.project();
match this.0.poll(cx) {
Poll::Ready(ev) => {
return Poll::Ready(ev.map_upgrade(EitherFuture::First).map_err(EitherError::A))
}
Poll::Pending => {}
}
match this.1.poll(cx) {
Poll::Ready(ev) => {
return Poll::Ready(ev.map_upgrade(EitherFuture::Second).map_err(EitherError::B))
}
Poll::Pending => {}
}
Poll::Pending
}
}

21
core/src/transport/dummy.rs
View File

@ -18,7 +18,7 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::transport::{ListenerEvent, Transport, TransportError};
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use crate::Multiaddr;
use futures::{prelude::*, task::Context, task::Poll};
use std::{fmt, io, marker::PhantomData, pin::Pin};
@ -56,19 +56,17 @@ impl<TOut> Clone for DummyTransport<TOut> {
impl<TOut> Transport for DummyTransport<TOut> {
type Output = TOut;
type Error = io::Error;
type Listener = futures::stream::Pending<
Result<ListenerEvent<Self::ListenerUpgrade, Self::Error>, Self::Error>,
>;
type ListenerUpgrade = futures::future::Pending<Result<Self::Output, io::Error>>;
type Dial = futures::future::Pending<Result<Self::Output, io::Error>>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
Err(TransportError::MultiaddrNotSupported(addr))
}
fn remove_listener(&mut self, _id: ListenerId) -> bool {
false
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
Err(TransportError::MultiaddrNotSupported(addr))
}
@ -83,6 +81,13 @@ impl<TOut> Transport for DummyTransport<TOut> {
fn address_translation(&self, _server: &Multiaddr, _observed: &Multiaddr) -> Option<Multiaddr> {
None
}
fn poll(
self: Pin<&mut Self>,
_: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
Poll::Pending
}
}
/// Implementation of `AsyncRead` and `AsyncWrite`. Not meant to be instanciated.

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

@ -20,15 +20,19 @@
use crate::{
connection::{ConnectedPoint, Endpoint},
transport::{ListenerEvent, Transport, TransportError},
transport::{Transport, TransportError, TransportEvent},
};
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::{pin::Pin, task::Context, task::Poll};
use super::ListenerId;
/// See `Transport::map`.
#[derive(Debug, Copy, Clone)]
#[pin_project::pin_project]
pub struct Map<T, F> {
#[pin]
transport: T,
fun: F,
}
@ -54,19 +58,15 @@ where
{
type Output = D;
type Error = T::Error;
type Listener = MapStream<T::Listener, F>;
type ListenerUpgrade = MapFuture<T::ListenerUpgrade, F>;
type Dial = MapFuture<T::Dial, F>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
let stream = self.transport.listen_on(addr)?;
Ok(MapStream {
stream,
fun: self.fun.clone(),
})
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.transport.listen_on(addr)
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.transport.remove_listener(id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
@ -99,58 +99,37 @@ where
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.transport.address_translation(server, observed)
}
}
/// Custom `Stream` implementation to avoid boxing.
///
/// Maps a function over every stream item.
#[pin_project::pin_project]
#[derive(Clone, Debug)]
pub struct MapStream<T, F> {
#[pin]
stream: T,
fun: F,
}
impl<T, F, A, B, X, E> Stream for MapStream<T, F>
where
T: TryStream<Ok = ListenerEvent<X, E>, Error = E>,
X: TryFuture<Ok = A>,
F: FnOnce(A, ConnectedPoint) -> B + Clone,
{
type Item = Result<ListenerEvent<MapFuture<X, F>, E>, E>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
let this = self.project();
match TryStream::try_poll_next(this.stream, cx) {
Poll::Ready(Some(Ok(event))) => {
let event = match event {
ListenerEvent::Upgrade {
upgrade,
local_addr,
remote_addr,
} => {
let point = ConnectedPoint::Listener {
local_addr: local_addr.clone(),
send_back_addr: remote_addr.clone(),
};
ListenerEvent::Upgrade {
upgrade: MapFuture {
inner: upgrade,
args: Some((this.fun.clone(), point)),
},
local_addr,
remote_addr,
}
}
ListenerEvent::NewAddress(a) => ListenerEvent::NewAddress(a),
ListenerEvent::AddressExpired(a) => ListenerEvent::AddressExpired(a),
ListenerEvent::Error(e) => ListenerEvent::Error(e),
match this.transport.poll(cx) {
Poll::Ready(TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
}) => {
let point = ConnectedPoint::Listener {
local_addr: local_addr.clone(),
send_back_addr: send_back_addr.clone(),
};
Poll::Ready(Some(Ok(event)))
Poll::Ready(TransportEvent::Incoming {
listener_id,
upgrade: MapFuture {
inner: upgrade,
args: Some((this.fun.clone(), point)),
},
local_addr,
send_back_addr,
})
}
Poll::Ready(other) => {
let mapped = other.map_upgrade(|_upgrade| unreachable!("case already matched"));
Poll::Ready(mapped)
}
Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
}
}

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

@ -18,14 +18,16 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::transport::{ListenerEvent, Transport, TransportError};
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use futures::prelude::*;
use multiaddr::Multiaddr;
use std::{error, pin::Pin, task::Context, task::Poll};
/// See `Transport::map_err`.
#[derive(Debug, Copy, Clone)]
#[pin_project::pin_project]
pub struct MapErr<T, F> {
#[pin]
transport: T,
map: F,
}
@ -45,19 +47,16 @@ where
{
type Output = T::Output;
type Error = TErr;
type Listener = MapErrListener<T, F>;
type ListenerUpgrade = MapErrListenerUpgrade<T, F>;
type Dial = MapErrDial<T, F>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
let map = self.map.clone();
match self.transport.listen_on(addr) {
Ok(stream) => Ok(MapErrListener { inner: stream, map }),
Err(err) => Err(err.map(map)),
}
self.transport.listen_on(addr).map_err(|err| err.map(map))
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.transport.remove_listener(id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
@ -88,41 +87,20 @@ where
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.transport.address_translation(server, observed)
}
}
/// Listening stream for `MapErr`.
#[pin_project::pin_project]
pub struct MapErrListener<T: Transport, F> {
#[pin]
inner: T::Listener,
map: F,
}
impl<T, F, TErr> Stream for MapErrListener<T, F>
where
T: Transport,
F: FnOnce(T::Error) -> TErr + Clone,
TErr: error::Error,
{
type Item = Result<ListenerEvent<MapErrListenerUpgrade<T, F>, TErr>, TErr>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
let this = self.project();
match TryStream::try_poll_next(this.inner, cx) {
Poll::Ready(Some(Ok(event))) => {
let map = &*this.map;
let event = event
.map(move |value| MapErrListenerUpgrade {
inner: value,
map: Some(map.clone()),
})
.map_err(|err| (map.clone())(err));
Poll::Ready(Some(Ok(event)))
}
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err((this.map.clone())(err)))),
}
let map = &*this.map;
this.transport.poll(cx).map(|ev| {
ev.map_upgrade(move |value| MapErrListenerUpgrade {
inner: value,
map: Some(map.clone()),
})
.map_err(map.clone())
})
}
}

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

@ -18,10 +18,7 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::{
transport::{ListenerEvent, TransportError},
Transport,
};
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use fnv::FnvHashMap;
use futures::{
channel::mpsc,
@ -34,7 +31,12 @@ use lazy_static::lazy_static;
use multiaddr::{Multiaddr, Protocol};
use parking_lot::Mutex;
use rw_stream_sink::RwStreamSink;
use std::{collections::hash_map::Entry, error, fmt, io, num::NonZeroU64, pin::Pin};
use std::{
collections::{hash_map::Entry, VecDeque},
error, fmt, io,
num::NonZeroU64,
pin::Pin,
};
lazy_static! {
static ref HUB: Hub = Hub(Mutex::new(FnvHashMap::default()));
@ -91,8 +93,16 @@ impl Hub {
}
/// Transport that supports `/memory/N` multiaddresses.
#[derive(Debug, Copy, Clone, Default)]
pub struct MemoryTransport;
#[derive(Default)]
pub struct MemoryTransport {
listeners: VecDeque<Pin<Box<Listener>>>,
}
impl MemoryTransport {
pub fn new() -> Self {
Self::default()
}
}
/// Connection to a `MemoryTransport` currently being opened.
pub struct DialFuture {
@ -168,14 +178,10 @@ impl Future for DialFuture {
impl Transport for MemoryTransport {
type Output = Channel<Vec<u8>>;
type Error = MemoryTransportError;
type Listener = Listener;
type ListenerUpgrade = Ready<Result<Self::Output, Self::Error>>;
type Dial = DialFuture;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
let port = if let Ok(port) = parse_memory_addr(&addr) {
port
} else {
@ -187,14 +193,29 @@ impl Transport for MemoryTransport {
None => return Err(TransportError::Other(MemoryTransportError::Unreachable)),
};
let id = ListenerId::new();
let listener = Listener {
id,
port,
addr: Protocol::Memory(port.get()).into(),
receiver: rx,
tell_listen_addr: true,
};
self.listeners.push_back(Box::pin(listener));
Ok(listener)
Ok(id)
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
if let Some(index) = self.listeners.iter().position(|listener| listener.id == id) {
let listener = self.listeners.get_mut(index).unwrap();
let val_in = HUB.unregister_port(&listener.port);
debug_assert!(val_in.is_some());
listener.receiver.close();
true
} else {
false
}
}
fn dial(&mut self, addr: Multiaddr) -> Result<DialFuture, TransportError<Self::Error>> {
@ -221,6 +242,56 @@ impl Transport for MemoryTransport {
fn address_translation(&self, _server: &Multiaddr, _observed: &Multiaddr) -> Option<Multiaddr> {
None
}
fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>>
where
Self: Sized,
{
let mut remaining = self.listeners.len();
while let Some(mut listener) = self.listeners.pop_back() {
if listener.tell_listen_addr {
listener.tell_listen_addr = false;
let listen_addr = listener.addr.clone();
let listener_id = listener.id;
self.listeners.push_front(listener);
return Poll::Ready(TransportEvent::NewAddress {
listen_addr,
listener_id,
});
}
let event = match Stream::poll_next(Pin::new(&mut listener.receiver), cx) {
Poll::Pending => None,
Poll::Ready(Some((channel, dial_port))) => Some(TransportEvent::Incoming {
listener_id: listener.id,
upgrade: future::ready(Ok(channel)),
local_addr: listener.addr.clone(),
send_back_addr: Protocol::Memory(dial_port.get()).into(),
}),
Poll::Ready(None) => {
// Listener was closed.
return Poll::Ready(TransportEvent::ListenerClosed {
listener_id: listener.id,
reason: Ok(()),
});
}
};
self.listeners.push_front(listener);
if let Some(event) = event {
return Poll::Ready(event);
} else {
remaining -= 1;
if remaining == 0 {
break;
}
}
}
Poll::Pending
}
}
/// Error that can be produced from the `MemoryTransport`.
@ -245,51 +316,17 @@ impl error::Error for MemoryTransportError {}
/// Listener for memory connections.
pub struct Listener {
id: ListenerId,
/// Port we're listening on.
port: NonZeroU64,
/// The address we are listening on.
addr: Multiaddr,
/// Receives incoming connections.
receiver: ChannelReceiver,
/// Generate `ListenerEvent::NewAddress` to inform about our listen address.
/// Generate [`TransportEvent::NewAddress`] to inform about our listen address.
tell_listen_addr: bool,
}
impl Stream for Listener {
type Item = Result<
ListenerEvent<Ready<Result<Channel<Vec<u8>>, MemoryTransportError>>, MemoryTransportError>,
MemoryTransportError,
>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if self.tell_listen_addr {
self.tell_listen_addr = false;
return Poll::Ready(Some(Ok(ListenerEvent::NewAddress(self.addr.clone()))));
}
let (channel, dial_port) = match Stream::poll_next(Pin::new(&mut self.receiver), cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(None) => panic!("Alive listeners always have a sender."),
Poll::Ready(Some(v)) => v,
};
let event = ListenerEvent::Upgrade {
upgrade: future::ready(Ok(channel)),
local_addr: self.addr.clone(),
remote_addr: Protocol::Memory(dial_port.get()).into(),
};
Poll::Ready(Some(Ok(event)))
}
}
impl Drop for Listener {
fn drop(&mut self) {
let val_in = HUB.unregister_port(&self.port);
debug_assert!(val_in.is_some());
}
}
/// If the address is `/memory/n`, returns the value of `n`.
fn parse_memory_addr(a: &Multiaddr) -> Result<u64, ()> {
let mut protocols = a.iter();
@ -418,28 +455,34 @@ mod tests {
#[test]
fn listening_twice() {
let mut transport = MemoryTransport::default();
assert!(transport
.listen_on("/memory/1639174018481".parse().unwrap())
.is_ok());
assert!(transport
.listen_on("/memory/1639174018481".parse().unwrap())
.is_ok());
let _listener = transport
.listen_on("/memory/1639174018481".parse().unwrap())
.unwrap();
assert!(transport
.listen_on("/memory/1639174018481".parse().unwrap())
.is_err());
assert!(transport
.listen_on("/memory/1639174018481".parse().unwrap())
.is_err());
drop(_listener);
assert!(transport
.listen_on("/memory/1639174018481".parse().unwrap())
.is_ok());
assert!(transport
.listen_on("/memory/1639174018481".parse().unwrap())
.is_ok());
let addr_1: Multiaddr = "/memory/1639174018481".parse().unwrap();
let addr_2: Multiaddr = "/memory/8459375923478".parse().unwrap();
let listener_id_1 = transport.listen_on(addr_1.clone()).unwrap();
assert!(
transport.remove_listener(listener_id_1),
"Listener doesn't exist."
);
let listener_id_2 = transport.listen_on(addr_1.clone()).unwrap();
let listener_id_3 = transport.listen_on(addr_2.clone()).unwrap();
assert!(transport.listen_on(addr_1.clone()).is_err());
assert!(transport.listen_on(addr_2.clone()).is_err());
assert!(
transport.remove_listener(listener_id_2),
"Listener doesn't exist."
);
assert!(transport.listen_on(addr_1).is_ok());
assert!(transport.listen_on(addr_2.clone()).is_err());
assert!(
transport.remove_listener(listener_id_3),
"Listener doesn't exist."
);
assert!(transport.listen_on(addr_2).is_ok());
}
#[test]
@ -456,6 +499,35 @@ mod tests {
.is_ok());
}
#[test]
fn stop_listening() {
let rand_port = rand::random::<u64>().saturating_add(1);
let addr: Multiaddr = format!("/memory/{}", rand_port).parse().unwrap();
let mut transport = MemoryTransport::default().boxed();
futures::executor::block_on(async {
let listener_id = transport.listen_on(addr.clone()).unwrap();
let reported_addr = transport
.select_next_some()
.await
.into_new_address()
.expect("new address");
assert_eq!(addr, reported_addr);
assert!(transport.remove_listener(listener_id));
match transport.select_next_some().await {
TransportEvent::ListenerClosed {
listener_id: id,
reason,
} => {
assert_eq!(id, listener_id);
assert!(reason.is_ok())
}
other => panic!("Unexpected transport event: {:?}", other),
}
assert!(!transport.remove_listener(listener_id));
})
}
#[test]
fn communicating_between_dialer_and_listener() {
let msg = [1, 2, 3];
@ -466,16 +538,16 @@ mod tests {
let t1_addr: Multiaddr = format!("/memory/{}", rand_port).parse().unwrap();
let cloned_t1_addr = t1_addr.clone();
let mut t1 = MemoryTransport::default();
let mut t1 = MemoryTransport::default().boxed();
let listener = async move {
let listener = t1.listen_on(t1_addr.clone()).unwrap();
let upgrade = listener
.filter_map(|ev| futures::future::ready(ListenerEvent::into_upgrade(ev.unwrap())))
.next()
.await
.unwrap();
t1.listen_on(t1_addr.clone()).unwrap();
let upgrade = loop {
let event = t1.select_next_some().await;
if let Some(upgrade) = event.into_incoming() {
break upgrade;
}
};
let mut socket = upgrade.0.await.unwrap();
@ -504,14 +576,16 @@ mod tests {
Protocol::Memory(rand::random::<u64>().saturating_add(1)).into();
let listener_addr_cloned = listener_addr.clone();
let mut listener_transport = MemoryTransport::default();
let mut listener_transport = MemoryTransport::default().boxed();
let listener = async move {
let mut listener = listener_transport.listen_on(listener_addr.clone()).unwrap();
while let Some(ev) = listener.next().await {
if let ListenerEvent::Upgrade { remote_addr, .. } = ev.unwrap() {
listener_transport.listen_on(listener_addr.clone()).unwrap();
loop {
if let TransportEvent::Incoming { send_back_addr, .. } =
listener_transport.select_next_some().await
{
assert!(
remote_addr != listener_addr,
send_back_addr != listener_addr,
"Expect dialer address not to equal listener address."
);
return;
@ -539,14 +613,16 @@ mod tests {
Protocol::Memory(rand::random::<u64>().saturating_add(1)).into();
let listener_addr_cloned = listener_addr.clone();
let mut listener_transport = MemoryTransport::default();
let mut listener_transport = MemoryTransport::default().boxed();
let listener = async move {
let mut listener = listener_transport.listen_on(listener_addr.clone()).unwrap();
while let Some(ev) = listener.next().await {
if let ListenerEvent::Upgrade { remote_addr, .. } = ev.unwrap() {
listener_transport.listen_on(listener_addr.clone()).unwrap();
loop {
if let TransportEvent::Incoming { send_back_addr, .. } =
listener_transport.select_next_some().await
{
let dialer_port =
NonZeroU64::new(parse_memory_addr(&remote_addr).unwrap()).unwrap();
NonZeroU64::new(parse_memory_addr(&send_back_addr).unwrap()).unwrap();
assert!(
HUB.get(&dialer_port).is_some(),

31
core/src/transport/optional.rs
View File

@ -18,8 +18,9 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use crate::transport::{Transport, TransportError};
use crate::transport::{ListenerId, Transport, TransportError, TransportEvent};
use multiaddr::Multiaddr;
use std::{pin::Pin, task::Context, task::Poll};
/// Transport that is possibly disabled.
///
@ -28,7 +29,8 @@ use multiaddr::Multiaddr;
/// enabled (read: contains `Some`), then dialing and listening will be handled by the inner
/// transport.
#[derive(Debug, Copy, Clone)]
pub struct OptionalTransport<T>(Option<T>);
#[pin_project::pin_project]
pub struct OptionalTransport<T>(#[pin] Option<T>);
impl<T> OptionalTransport<T> {
/// Builds an `OptionalTransport` with the given transport in an enabled
@ -55,14 +57,10 @@ where
{
type Output = T::Output;
type Error = T::Error;
type Listener = T::Listener;
type ListenerUpgrade = T::ListenerUpgrade;
type Dial = T::Dial;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
if let Some(inner) = self.0.as_mut() {
inner.listen_on(addr)
} else {
@ -70,6 +68,14 @@ where
}
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
if let Some(inner) = self.0.as_mut() {
inner.remove_listener(id)
} else {
false
}
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
if let Some(inner) = self.0.as_mut() {
inner.dial(addr)
@ -96,4 +102,15 @@ where
None
}
}
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
if let Some(inner) = self.project().0.as_pin_mut() {
inner.poll(cx)
} else {
Poll::Pending
}
}
}

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

@ -25,7 +25,7 @@
// TODO: add example
use crate::{
transport::{ListenerEvent, TransportError},
transport::{ListenerId, TransportError, TransportEvent},
Multiaddr, Transport,
};
use futures::prelude::*;
@ -38,7 +38,9 @@ use std::{error, fmt, io, pin::Pin, task::Context, task::Poll, time::Duration};
/// **Note**: `listen_on` is never subject to a timeout, only the setup of each
/// individual accepted connection.
#[derive(Debug, Copy, Clone)]
#[pin_project::pin_project]
pub struct TransportTimeout<InnerTrans> {
#[pin]
inner: InnerTrans,
outgoing_timeout: Duration,
incoming_timeout: Duration,
@ -80,25 +82,17 @@ where
{
type Output = InnerTrans::Output;
type Error = TransportTimeoutError<InnerTrans::Error>;
type Listener = TimeoutListener<InnerTrans::Listener>;
type ListenerUpgrade = Timeout<InnerTrans::ListenerUpgrade>;
type Dial = Timeout<InnerTrans::Dial>;
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
let listener = self
.inner
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.inner
.listen_on(addr)
.map_err(|err| err.map(TransportTimeoutError::Other))?;
.map_err(|err| err.map(TransportTimeoutError::Other))
}
let listener = TimeoutListener {
inner: listener,
timeout: self.incoming_timeout,
};
Ok(listener)
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.inner.remove_listener(id)
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
@ -129,45 +123,21 @@ where
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.inner.address_translation(server, observed)
}
}
// TODO: can be removed and replaced with an `impl Stream` once impl Trait is fully stable
// in Rust (https://github.com/rust-lang/rust/issues/34511)
#[pin_project::pin_project]
pub struct TimeoutListener<InnerStream> {
#[pin]
inner: InnerStream,
timeout: Duration,
}
impl<InnerStream, O, E> Stream for TimeoutListener<InnerStream>
where
InnerStream: TryStream<Ok = ListenerEvent<O, E>, Error = E>,
{
type Item =
Result<ListenerEvent<Timeout<O>, TransportTimeoutError<E>>, TransportTimeoutError<E>>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
let this = self.project();
let poll_out = match TryStream::try_poll_next(this.inner, cx) {
Poll::Ready(Some(Err(err))) => {
return Poll::Ready(Some(Err(TransportTimeoutError::Other(err))))
}
Poll::Ready(Some(Ok(v))) => v,
Poll::Ready(None) => return Poll::Ready(None),
Poll::Pending => return Poll::Pending,
};
let timeout = *this.timeout;
let event = poll_out
.map(move |inner_fut| Timeout {
inner: inner_fut,
timer: Delay::new(timeout),
})
.map_err(TransportTimeoutError::Other);
Poll::Ready(Some(Ok(event)))
let timeout = *this.incoming_timeout;
this.inner.poll(cx).map(|event| {
event
.map_upgrade(move |inner_fut| Timeout {
inner: inner_fut,
timer: Delay::new(timeout),
})
.map_err(TransportTimeoutError::Other)
})
}
}

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

@ -26,8 +26,8 @@ use crate::{
connection::ConnectedPoint,
muxing::{StreamMuxer, StreamMuxerBox},
transport::{
and_then::AndThen, boxed::boxed, timeout::TransportTimeout, ListenerEvent, Transport,
TransportError,
and_then::AndThen, boxed::boxed, timeout::TransportTimeout, ListenerId, Transport,
TransportError, TransportEvent,
},
upgrade::{
self, apply_inbound, apply_outbound, InboundUpgrade, InboundUpgradeApply, OutboundUpgrade,
@ -287,16 +287,16 @@ where
/// A authenticated and multiplexed transport, obtained from
/// [`Authenticated::multiplex`].
#[derive(Clone)]
pub struct Multiplexed<T>(T);
#[pin_project::pin_project]
pub struct Multiplexed<T>(#[pin] T);
impl<T> Multiplexed<T> {
/// Boxes the authenticated, multiplexed transport, including
/// the [`StreamMuxer`] and custom transport errors.
pub fn boxed<M>(self) -> super::Boxed<(PeerId, StreamMuxerBox)>
where
T: Transport<Output = (PeerId, M)> + Sized + Send + 'static,
T: Transport<Output = (PeerId, M)> + Sized + Send + Unpin + 'static,
T::Dial: Send + 'static,
T::Listener: Send + 'static,
T::ListenerUpgrade: Send + 'static,
T::Error: Send + Sync,
M: StreamMuxer + Send + Sync + 'static,
@ -332,7 +332,6 @@ where
{
type Output = T::Output;
type Error = T::Error;
type Listener = T::Listener;
type ListenerUpgrade = T::ListenerUpgrade;
type Dial = T::Dial;
@ -340,6 +339,10 @@ where
self.0.dial(addr)
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.0.remove_listener(id)
}
fn dial_as_listener(
&mut self,
addr: Multiaddr,
@ -347,16 +350,20 @@ where
self.0.dial_as_listener(addr)
}
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.0.listen_on(addr)
}
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.0.address_translation(server, observed)
}
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
self.project().0.poll(cx)
}
}
/// An inbound or outbound upgrade.
@ -366,7 +373,9 @@ type EitherUpgrade<C, U> = future::Either<InboundUpgradeApply<C, U>, OutboundUpg
///
/// See [`Transport::upgrade`]
#[derive(Debug, Copy, Clone)]
#[pin_project::pin_project]
pub struct Upgrade<T, U> {
#[pin]
inner: T,
upgrade: U,
}
@ -388,7 +397,6 @@ where
{
type Output = (PeerId, D);
type Error = TransportUpgradeError<T::Error, E>;
type Listener = ListenerStream<T::Listener, U>;
type ListenerUpgrade = ListenerUpgradeFuture<T::ListenerUpgrade, U, C>;
type Dial = DialUpgradeFuture<T::Dial, U, C>;
@ -403,6 +411,10 @@ where
})
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
self.inner.remove_listener(id)
}
fn dial_as_listener(
&mut self,
addr: Multiaddr,
@ -417,23 +429,31 @@ where
})
}
fn listen_on(
&mut self,
addr: Multiaddr,
) -> Result<Self::Listener, TransportError<Self::Error>> {
let stream = self
.inner
fn listen_on(&mut self, addr: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
self.inner
.listen_on(addr)
.map_err(|err| err.map(TransportUpgradeError::Transport))?;
Ok(ListenerStream {
stream: Box::pin(stream),
upgrade: self.upgrade.clone(),
})
.map_err(|err| err.map(TransportUpgradeError::Transport))
}
fn address_translation(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
self.inner.address_translation(server, observed)
}
fn poll(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
let this = self.project();
let upgrade = this.upgrade.clone();
this.inner.poll(cx).map(|event| {
event
.map_upgrade(move |future| ListenerUpgradeFuture {
future: Box::pin(future),
upgrade: future::Either::Left(Some(upgrade)),
})
.map_err(TransportUpgradeError::Transport)
})
}
}
/// Errors produced by a transport upgrade.
@ -478,7 +498,7 @@ where
C: AsyncRead + AsyncWrite + Unpin,
{
future: Pin<Box<F>>,
upgrade: future::Either<Option<U>, (Option<PeerId>, OutboundUpgradeApply<C, U>)>,
upgrade: future::Either<Option<U>, (PeerId, OutboundUpgradeApply<C, U>)>,
}
impl<F, U, C, D> Future for DialUpgradeFuture<F, U, C>
@ -507,18 +527,15 @@ where
let u = up
.take()
.expect("DialUpgradeFuture is constructed with Either::Left(Some).");
future::Either::Right((Some(i), apply_outbound(c, u, upgrade::Version::V1)))
future::Either::Right((i, apply_outbound(c, u, upgrade::Version::V1)))
}
future::Either::Right((ref mut i, ref mut up)) => {
future::Either::Right((i, ref mut up)) => {
let d = match ready!(
Future::poll(Pin::new(up), cx).map_err(TransportUpgradeError::Upgrade)
) {
Ok(d) => d,
Err(err) => return Poll::Ready(Err(err)),
};
let i = i
.take()
.expect("DialUpgradeFuture polled after completion.");
return Poll::Ready(Ok((i, d)));
}
}
@ -533,43 +550,6 @@ where
{
}
/// The [`Transport::Listener`] stream of an [`Upgrade`]d transport.
pub struct ListenerStream<S, U> {
stream: Pin<Box<S>>,
upgrade: U,
}
impl<S, U, F, C, D, E> Stream for ListenerStream<S, U>
where
S: TryStream<Ok = ListenerEvent<F, E>, Error = E>,
F: TryFuture<Ok = (PeerId, C)>,
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>, Output = D> + Clone,
{
type Item = Result<
ListenerEvent<ListenerUpgradeFuture<F, U, C>, TransportUpgradeError<E, U::Error>>,
TransportUpgradeError<E, U::Error>,
>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
match ready!(TryStream::try_poll_next(self.stream.as_mut(), cx)) {
Some(Ok(event)) => {
let event = event
.map(move |future| ListenerUpgradeFuture {
future: Box::pin(future),
upgrade: future::Either::Left(Some(self.upgrade.clone())),
})
.map_err(TransportUpgradeError::Transport);
Poll::Ready(Some(Ok(event)))
}
Some(Err(err)) => Poll::Ready(Some(Err(TransportUpgradeError::Transport(err)))),
None => Poll::Ready(None),
}
}
}
impl<S, U> Unpin for ListenerStream<S, U> {}
/// The [`Transport::ListenerUpgrade`] future of an [`Upgrade`]d transport.
pub struct ListenerUpgradeFuture<F, U, C>
where
@ -577,7 +557,7 @@ where
U: InboundUpgrade<Negotiated<C>>,
{
future: Pin<Box<F>>,
upgrade: future::Either<Option<U>, (Option<PeerId>, InboundUpgradeApply<C, U>)>,
upgrade: future::Either<Option<U>, (PeerId, InboundUpgradeApply<C, U>)>,
}
impl<F, U, C, D> Future for ListenerUpgradeFuture<F, U, C>
@ -606,18 +586,15 @@ where
let u = up
.take()
.expect("ListenerUpgradeFuture is constructed with Either::Left(Some).");
future::Either::Right((Some(i), apply_inbound(c, u)))
future::Either::Right((i, apply_inbound(c, u)))
}
future::Either::Right((ref mut i, ref mut up)) => {
future::Either::Right((i, ref mut up)) => {
let d = match ready!(TryFuture::try_poll(Pin::new(up), cx)
.map_err(TransportUpgradeError::Upgrade))
{
Ok(v) => v,
Err(err) => return Poll::Ready(Err(err)),
};
let i = i
.take()
.expect("ListenerUpgradeFuture polled after completion.");
return Poll::Ready(Ok((i, d)));
}
}