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

Remove some Unpin requirements on Futures (#1384)

Browse Source 
* Remove lots of Unpin requirements

* Make Transport::and_then accept pinned futures

* Finish the PR

* Work on secio

* Fix BandwidthTransport

* Adjust ListenersStrema

* Fix nodes/tasks

* Fix nodes

* Various more fixes

* Fix yamux

* Fix Swarm

* Fix WebSockets

* Fix rw-stream-sink
This commit is contained in:
Pierre Krieger
2020-01-14 12:03:10 +01:00
committed by GitHub
parent 42a45e2630
commit 3f968cbf92
23 changed files with 249 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
core/src/either.rs
View File

@ -155,8 +155,8 @@ where
impl<A, B, I> Sink<I> for EitherOutput<A, B>
where
A: Sink<I> + Unpin,
B: Sink<I> + Unpin,
A: Sink<I>,
B: Sink<I>,
{
type Error = EitherError<A::Error, B::Error>;
@ -414,8 +414,8 @@ pub enum EitherFuture2<A, B> { A(#[pin] A), B(#[pin] B) }
impl<AFut, BFut, AItem, BItem, AError, BError> Future for EitherFuture2<AFut, BFut>
where
AFut: TryFuture<Ok = AItem, Error = AError> + Unpin,
BFut: TryFuture<Ok = BItem, Error = BError> + Unpin,
AFut: TryFuture<Ok = AItem, Error = AError>,
BFut: TryFuture<Ok = BItem, Error = BError>,
{
type Output = Result<EitherOutput<AItem, BItem>, EitherError<AError, BError>>;

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

@ -322,7 +322,7 @@ where
pub fn add_reach_attempt<TFut, TMuxer>(&mut self, future: TFut, handler: THandler)
-> ReachAttemptId
where
TFut: Future<Output = Result<(TConnInfo, TMuxer), TReachErr>> + Unpin + Send + 'static,
TFut: Future<Output = Result<(TConnInfo, TMuxer), TReachErr>> + Send + 'static,
THandler: IntoNodeHandler<TConnInfo> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,

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

@ -60,7 +60,7 @@ use std::{collections::VecDeque, fmt, pin::Pin};
/// ListenersEvent::AddressExpired { listener_id, listen_addr } => {
/// println!("Listener {:?} is no longer listening at address {}", listener_id, listen_addr);
/// },
/// ListenersEvent::Closed { listener_id, .. } => {
/// ListenersEvent::Closed { listener_id } => {
/// println!("Listener {:?} has been closed", listener_id);
/// },
/// ListenersEvent::Error { listener_id, error } => {
@ -84,7 +84,9 @@ where
/// Transport used to spawn listeners.
transport: TTrans,
/// All the active listeners.
listeners: VecDeque<Listener<TTrans>>,
/// The `Listener` struct contains a stream that we want to be pinned. Since the `VecDeque`
/// can be resized, the only way is to use a `Pin<Box<>>`.
listeners: VecDeque<Pin<Box<Listener<TTrans>>>>,
/// The next listener ID to assign.
next_id: ListenerId
}
@ -97,6 +99,7 @@ where
pub struct ListenerId(u64);
/// A single active listener.
#[pin_project::pin_project]
#[derive(Debug)]
struct Listener<TTrans>
where
@ -105,6 +108,7 @@ where
/// The ID of this listener.
id: ListenerId,
/// The object that actually listens.
#[pin]
listener: TTrans::Listener,
/// Addresses it is listening on.
addresses: SmallVec<[Multiaddr; 4]>
@ -144,8 +148,6 @@ where
Closed {
/// The ID of the listener that closed.
listener_id: ListenerId,
/// The listener that closed.
listener: TTrans::Listener,
},
/// A listener errored.
///
@ -190,22 +192,25 @@ where
TTrans: Clone,
{
let listener = self.transport.clone().listen_on(addr)?;
self.listeners.push_back(Listener {
self.listeners.push_back(Box::pin(Listener {
id: self.next_id,
listener,
addresses: SmallVec::new()
});
}));
let id = self.next_id;
self.next_id = ListenerId(self.next_id.0 + 1);
Ok(id)
}
/// Remove the listener matching the given `ListenerId`.
pub fn remove_listener(&mut self, id: ListenerId) -> Option<TTrans::Listener> {
///
/// Return `Ok(())` if a listener with this ID was in the list.
pub fn remove_listener(&mut self, id: ListenerId) -> Result<(), ()> {
if let Some(i) = self.listeners.iter().position(|l| l.id == id) {
self.listeners.remove(i).map(|l| l.listener)
self.listeners.remove(i);
Ok(())
} else {
None
Err(())
}
}
@ -220,21 +225,19 @@ where
}
/// Provides an API similar to `Stream`, except that it cannot end.
pub fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<ListenersEvent<TTrans>>
where
TTrans::Listener: Unpin,
{
pub fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<ListenersEvent<TTrans>> {
// We remove each element from `listeners` one by one and add them back.
let mut remaining = self.listeners.len();
while let Some(mut listener) = self.listeners.pop_back() {
match TryStream::try_poll_next(Pin::new(&mut listener.listener), cx) {
let mut listener_project = listener.as_mut().project();
match TryStream::try_poll_next(listener_project.listener.as_mut(), cx) {
Poll::Pending => {
self.listeners.push_front(listener);
remaining -= 1;
if remaining == 0 { break }
}
Poll::Ready(Some(Ok(ListenerEvent::Upgrade { upgrade, local_addr, remote_addr }))) => {
let id = listener.id;
let id = *listener_project.id;
self.listeners.push_front(listener);
return Poll::Ready(ListenersEvent::Incoming {
listener_id: id,
@ -244,13 +247,13 @@ where
})
}
Poll::Ready(Some(Ok(ListenerEvent::NewAddress(a)))) => {
if listener.addresses.contains(&a) {
if listener_project.addresses.contains(&a) {
debug!("Transport has reported address {} multiple times", a)
}
if !listener.addresses.contains(&a) {
listener.addresses.push(a.clone());
if !listener_project.addresses.contains(&a) {
listener_project.addresses.push(a.clone());
}
let id = listener.id;
let id = *listener_project.id;
self.listeners.push_front(listener);
return Poll::Ready(ListenersEvent::NewAddress {
listener_id: id,
@ -258,8 +261,8 @@ where
})
}
Poll::Ready(Some(Ok(ListenerEvent::AddressExpired(a)))) => {
listener.addresses.retain(|x| x != &a);
let id = listener.id;
listener_project.addresses.retain(|x| x != &a);
let id = *listener_project.id;
self.listeners.push_front(listener);
return Poll::Ready(ListenersEvent::AddressExpired {
listener_id: id,
@ -268,13 +271,12 @@ where
}
Poll::Ready(None) => {
return Poll::Ready(ListenersEvent::Closed {
listener_id: listener.id,
listener: listener.listener
listener_id: *listener_project.id,
})
}
Poll::Ready(Some(Err(err))) => {
return Poll::Ready(ListenersEvent::Error {
listener_id: listener.id,
listener_id: *listener_project.id,
error: err
})
}
@ -289,7 +291,6 @@ where
impl<TTrans> Stream for ListenersStream<TTrans>
where
TTrans: Transport,
TTrans::Listener: Unpin,
{
type Item = ListenersEvent<TTrans>;
@ -338,7 +339,7 @@ where
.field("listener_id", listener_id)
.field("local_addr", local_addr)
.finish(),
ListenersEvent::Closed { listener_id, .. } => f
ListenersEvent::Closed { listener_id } => f
.debug_struct("ListenersEvent::Closed")
.field("listener_id", listener_id)
.finish(),

25
core/src/nodes/network.rs
View File

@ -171,8 +171,6 @@ where
ListenerClosed {
/// The listener ID that closed.
listener_id: ListenerId,
/// The listener which closed.
listener: TTrans::Listener,
},
/// One of the listeners errored.
@ -309,7 +307,7 @@ where
.field("listen_addr", listen_addr)
.finish()
}
NetworkEvent::ListenerClosed { listener_id, .. } => {
NetworkEvent::ListenerClosed { listener_id } => {
f.debug_struct("ListenerClosed")
.field("listener_id", listener_id)
.finish()
@ -580,7 +578,7 @@ impl<'a, TTrans, TInEvent, TOutEvent, TMuxer, THandler, THandlerErr, TConnInfo,
where
TTrans: Transport<Output = (TConnInfo, TMuxer)>,
TTrans::Error: Send + 'static,
TTrans::ListenerUpgrade: Unpin + Send + 'static,
TTrans::ListenerUpgrade: Send + 'static,
THandler: IntoNodeHandler<(TConnInfo, ConnectedPoint)> + Send + 'static,
THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
<THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static, // TODO: shouldn't be necessary
@ -735,7 +733,9 @@ where
}
/// Remove a previously added listener.
pub fn remove_listener(&mut self, id: ListenerId) -> Option<TTrans::Listener> {
///
/// Returns `Ok(())` if a listener with this ID was in the list.
pub fn remove_listener(&mut self, id: ListenerId) -> Result<(), ()> {
self.listeners.remove_listener(id)
}
@ -788,7 +788,7 @@ where
where
TTrans: Transport<Output = (TConnInfo, TMuxer)>,
TTrans::Error: Send + 'static,
TTrans::Dial: Unpin + Send + 'static,
TTrans::Dial: Send + 'static,
TMuxer: Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TInEvent: Send + 'static,
@ -936,7 +936,7 @@ where
fn start_dial_out(&mut self, peer_id: TPeerId, handler: THandler, first: Multiaddr, rest: Vec<Multiaddr>)
where
TTrans: Transport<Output = (TConnInfo, TMuxer)>,
TTrans::Dial: Unpin + Send + 'static,
TTrans::Dial: Send + 'static,
TTrans::Error: Send + 'static,
TMuxer: Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
@ -982,8 +982,7 @@ where
where
TTrans: Transport<Output = (TConnInfo, TMuxer)>,
TTrans::Error: Send + 'static,
TTrans::Dial: Unpin + Send + 'static,
TTrans::Listener: Unpin,
TTrans::Dial: Send + 'static,
TTrans::ListenerUpgrade: Send + 'static,
TMuxer: Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
@ -1021,8 +1020,8 @@ where
Poll::Ready(ListenersEvent::AddressExpired { listener_id, listen_addr }) => {
return Poll::Ready(NetworkEvent::ExpiredListenerAddress { listener_id, listen_addr })
}
Poll::Ready(ListenersEvent::Closed { listener_id, listener }) => {
return Poll::Ready(NetworkEvent::ListenerClosed { listener_id, listener })
Poll::Ready(ListenersEvent::Closed { listener_id }) => {
return Poll::Ready(NetworkEvent::ListenerClosed { listener_id })
}
Poll::Ready(ListenersEvent::Error { listener_id, error }) => {
return Poll::Ready(NetworkEvent::ListenerError { listener_id, error })
@ -1462,7 +1461,7 @@ impl<'a, TTrans, TMuxer, TInEvent, TOutEvent, THandler, THandlerErr, TConnInfo,
where
TTrans: Transport<Output = (TConnInfo, TMuxer)> + Clone,
TTrans::Error: Send + 'static,
TTrans::Dial: Unpin + Send + 'static,
TTrans::Dial: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,
@ -1759,7 +1758,7 @@ impl<'a, TTrans, TInEvent, TOutEvent, TMuxer, THandler, THandlerErr, TConnInfo,
where
TTrans: Transport<Output = (TConnInfo, TMuxer)> + Clone,
TTrans::Error: Send + 'static,
TTrans::Dial: Unpin + Send + 'static,
TTrans::Dial: Send + 'static,
TMuxer: StreamMuxer + Send + Sync + 'static,
TMuxer::OutboundSubstream: Send,
TMuxer::Substream: Send,

2
core/src/nodes/tasks/manager.rs
View File

@ -157,7 +157,7 @@ impl<I, O, H, E, HE, T, C> Manager<I, O, H, E, HE, T, C> {
/// processing the node's events.
pub fn add_reach_attempt<F, M>(&mut self, future: F, user_data: T, handler: H) -> TaskId
where
F: Future<Output = Result<(C, M), E>> + Unpin + Send + 'static,
F: Future<Output = Result<(C, M), E>> + Send + 'static,
H: IntoNodeHandler<C> + Send + 'static,
H::Handler: NodeHandler<Substream = Substream<M>, InEvent = I, OutEvent = O, Error = HE> + Send + 'static,
E: error::Error + Send + 'static,

7
core/src/nodes/tasks/task.rs
View File

@ -92,7 +92,7 @@ where
id: i,
sender: s,
receiver: r.fuse(),
state: State::Future { future: f, handler: h, events_buffer: Vec::new() },
state: State::Future { future: Box::pin(f), handler: h, events_buffer: Vec::new() },
taken_over: SmallVec::new()
}
}
@ -124,7 +124,8 @@ where
/// Future to resolve to connect to the node.
Future {
/// The future that will attempt to reach the node.
future: F,
// TODO: don't pin this Future; this requires deeper changes though
future: Pin<Box<F>>,
/// The handler that will be used to build the `HandledNode`.
handler: H,
/// While we are dialing the future, we need to buffer the events received on
@ -163,7 +164,7 @@ where
impl<F, M, H, I, O, E, C> Future for Task<F, M, H, I, O, E, C>
where
M: StreamMuxer,
F: Future<Output = Result<(C, M), E>> + Unpin,
F: Future<Output = Result<(C, M), E>>,
H: IntoNodeHandler<C>,
H::Handler: NodeHandler<Substream = Substream<M>, InEvent = I, OutEvent = O>
{

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

@ -25,7 +25,7 @@ use crate::{
};
use futures::{future::Either, prelude::*};
use multiaddr::Multiaddr;
use std::{error, pin::Pin, task::Context, task::Poll};
use std::{error, marker::PhantomPinned, pin::Pin, task::Context, task::Poll};
/// See the `Transport::and_then` method.
#[derive(Debug, Clone)]
@ -40,11 +40,8 @@ impl<T, C> AndThen<T, C> {
impl<T, C, F, O> Transport for AndThen<T, C>
where
T: Transport,
T::Dial: Unpin,
T::Listener: Unpin,
T::ListenerUpgrade: Unpin,
C: FnOnce(T::Output, ConnectedPoint) -> F + Clone,
F: TryFuture<Ok = O> + Unpin,
F: TryFuture<Ok = O>,
F::Error: error::Error,
{
type Output = O;
@ -66,8 +63,9 @@ where
fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
let dialed_fut = self.transport.dial(addr.clone()).map_err(|err| err.map(EitherError::A))?;
let future = AndThenFuture {
inner: Either::Left(dialed_fut),
args: Some((self.fun, ConnectedPoint::Dialer { address: addr }))
inner: Either::Left(Box::pin(dialed_fut)),
args: Some((self.fun, ConnectedPoint::Dialer { address: addr })),
marker: PhantomPinned,
};
Ok(future)
}
@ -76,18 +74,17 @@ where
/// 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> Unpin for AndThenStream<TListener, TMap> {
}
impl<TListener, TMap, TTransOut, TMapOut, TListUpgr, TTransErr> Stream for AndThenStream<TListener, TMap>
where
TListener: TryStream<Ok = ListenerEvent<TListUpgr>, Error = TTransErr> + Unpin,
TListener: TryStream<Ok = ListenerEvent<TListUpgr>, Error = TTransErr>,
TListUpgr: TryFuture<Ok = TTransOut, Error = TTransErr>,
TMap: FnOnce(TTransOut, ConnectedPoint) -> TMapOut + Clone,
TMapOut: TryFuture
@ -97,8 +94,9 @@ where
EitherError<TTransErr, TMapOut::Error>
>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
match TryStream::try_poll_next(Pin::new(&mut self.stream), cx) {
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
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 } => {
@ -108,8 +106,9 @@ where
};
ListenerEvent::Upgrade {
upgrade: AndThenFuture {
inner: Either::Left(upgrade),
args: Some((self.fun.clone(), point))
inner: Either::Left(Box::pin(upgrade)),
args: Some((this.fun.clone(), point)),
marker: PhantomPinned,
},
local_addr,
remote_addr
@ -132,26 +131,24 @@ where
/// Applies a function to the result of the inner future.
#[derive(Debug)]
pub struct AndThenFuture<TFut, TMap, TMapOut> {
inner: Either<TFut, TMapOut>,
args: Option<(TMap, ConnectedPoint)>
}
impl<TFut, TMap, TMapOut> Unpin for AndThenFuture<TFut, TMap, TMapOut> {
inner: Either<Pin<Box<TFut>>, Pin<Box<TMapOut>>>,
args: Option<(TMap, ConnectedPoint)>,
marker: PhantomPinned,
}
impl<TFut, TMap, TMapOut> Future for AndThenFuture<TFut, TMap, TMapOut>
where
TFut: TryFuture + Unpin,
TFut: TryFuture,
TMap: FnOnce(TFut::Ok, ConnectedPoint) -> TMapOut,
TMapOut: TryFuture + Unpin
TMapOut: TryFuture,
{
type Output = Result<TMapOut::Ok, EitherError<TFut::Error, TMapOut::Error>>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
loop {
let future = match (*self).inner {
Either::Left(ref mut future) => {
let item = match TryFuture::try_poll(Pin::new(future), cx) {
let future = match &mut self.inner {
Either::Left(future) => {
let item = match TryFuture::try_poll(future.as_mut(), cx) {
Poll::Ready(Ok(v)) => v,
Poll::Ready(Err(err)) => return Poll::Ready(Err(EitherError::A(err))),
Poll::Pending => return Poll::Pending,
@ -159,8 +156,8 @@ where
let (f, a) = self.args.take().expect("AndThenFuture has already finished.");
f(item, a)
}
Either::Right(ref mut future) => {
return match TryFuture::try_poll(Pin::new(future), cx) {
Either::Right(future) => {
return match TryFuture::try_poll(future.as_mut(), cx) {
Poll::Ready(Ok(v)) => Poll::Ready(Ok(v)),
Poll::Ready(Err(err)) => return Poll::Ready(Err(EitherError::B(err))),
Poll::Pending => Poll::Pending,
@ -168,7 +165,10 @@ where
}
};
(*self).inner = Either::Right(future);
self.inner = Either::Right(Box::pin(future));
}
}
}
impl<TFut, TMap, TMapOut> Unpin for AndThenFuture<TFut, TMap, TMapOut> {
}

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

@ -101,9 +101,6 @@ where
AndThen<T, impl FnOnce(C, ConnectedPoint) -> Authenticate<C, U> + Clone>
> where
T: Transport<Output = C>,
T::Dial: Unpin,
T::Listener: Unpin,
T::ListenerUpgrade: Unpin,
I: ConnectionInfo,
C: AsyncRead + AsyncWrite + Unpin,
D: AsyncRead + AsyncWrite + Unpin,
@ -133,9 +130,6 @@ where
pub fn apply<C, D, U, I, E>(self, upgrade: U) -> Builder<Upgrade<T, U>>
where
T: Transport<Output = (I, C)>,
T::Dial: Unpin,
T::Listener: Unpin,
T::ListenerUpgrade: Unpin,
C: AsyncRead + AsyncWrite + Unpin,
D: AsyncRead + AsyncWrite + Unpin,
I: ConnectionInfo,
@ -161,9 +155,6 @@ where
-> AndThen<T, impl FnOnce((I, C), ConnectedPoint) -> Multiplex<C, U, I> + Clone>
where
T: Transport<Output = (I, C)>,
T::Dial: Unpin,
T::Listener: Unpin,
T::ListenerUpgrade: Unpin,
C: AsyncRead + AsyncWrite + Unpin,
M: StreamMuxer,
I: ConnectionInfo,
@ -183,11 +174,13 @@ where
/// in the context of negotiating a secure channel.
///
/// Configured through [`Builder::authenticate`].
#[pin_project::pin_project]
pub struct Authenticate<C, U>
where
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>> + OutboundUpgrade<Negotiated<C>>
{
#[pin]
inner: EitherUpgrade<C, U>
}
@ -201,8 +194,9 @@ where
{
type Output = <EitherUpgrade<C, U> as Future>::Output;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
Future::poll(Pin::new(&mut self.inner), cx)
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
Future::poll(this.inner, cx)
}
}
@ -210,12 +204,14 @@ where
/// top of an authenticated transport.
///
/// Configured through [`Builder::multiplex`].
#[pin_project::pin_project]
pub struct Multiplex<C, U, I>
where
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>> + OutboundUpgrade<Negotiated<C>>,
{
info: Option<I>,
#[pin]
upgrade: EitherUpgrade<C, U>,
}
@ -227,23 +223,17 @@ where
{
type Output = Result<(I, M), UpgradeError<E>>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let m = match ready!(Future::poll(Pin::new(&mut self.upgrade), cx)) {
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let this = self.project();
let m = match ready!(Future::poll(this.upgrade, cx)) {
Ok(m) => m,
Err(err) => return Poll::Ready(Err(err)),
};
let i = self.info.take().expect("Multiplex future polled after completion.");
let i = this.info.take().expect("Multiplex future polled after completion.");
Poll::Ready(Ok((i, m)))
}
}
impl<C, U, I> Unpin for Multiplex<C, U, I>
where
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>> + OutboundUpgrade<Negotiated<C>>,
{
}
/// An inbound or outbound upgrade.
type EitherUpgrade<C, U> = future::Either<InboundUpgradeApply<C, U>, OutboundUpgradeApply<C, U>>;
@ -262,9 +252,6 @@ impl<T, U> Upgrade<T, U> {
impl<T, C, D, U, I, E> Transport for Upgrade<T, U>
where
T: Transport<Output = (I, C)>,
T::Dial: Unpin,
T::Listener: Unpin,
T::ListenerUpgrade: Unpin,
T::Error: 'static,
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>, Output = D, Error = E>,
@ -281,7 +268,7 @@ where
let future = self.inner.dial(addr.clone())
.map_err(|err| err.map(TransportUpgradeError::Transport))?;
Ok(DialUpgradeFuture {
future,
future: Box::pin(future),
upgrade: future::Either::Left(Some(self.upgrade))
})
}
@ -290,7 +277,7 @@ where
let stream = self.inner.listen_on(addr)
.map_err(|err| err.map(TransportUpgradeError::Transport))?;
Ok(ListenerStream {
stream,
stream: Box::pin(stream),
upgrade: self.upgrade
})
}
@ -337,13 +324,13 @@ where
U: OutboundUpgrade<Negotiated<C>>,
C: AsyncRead + AsyncWrite + Unpin,
{
future: F,
future: Pin<Box<F>>,
upgrade: future::Either<Option<U>, (Option<I>, OutboundUpgradeApply<C, U>)>
}
impl<F, U, I, C, D> Future for DialUpgradeFuture<F, U, I, C>
where
F: TryFuture<Ok = (I, C)> + Unpin,
F: TryFuture<Ok = (I, C)>,
C: AsyncRead + AsyncWrite + Unpin,
U: OutboundUpgrade<Negotiated<C>, Output = D>,
U::Error: Error
@ -358,7 +345,7 @@ where
loop {
this.upgrade = match this.upgrade {
future::Either::Left(ref mut up) => {
let (i, c) = match ready!(TryFuture::try_poll(Pin::new(&mut this.future), cx).map_err(TransportUpgradeError::Transport)) {
let (i, c) = match ready!(TryFuture::try_poll(this.future.as_mut(), cx).map_err(TransportUpgradeError::Transport)) {
Ok(v) => v,
Err(err) => return Poll::Ready(Err(err)),
};
@ -387,13 +374,13 @@ where
/// The [`Transport::Listener`] stream of an [`Upgrade`]d transport.
pub struct ListenerStream<S, U> {
stream: S,
stream: Pin<Box<S>>,
upgrade: U
}
impl<S, U, F, I, C, D> Stream for ListenerStream<S, U>
where
S: TryStream<Ok = ListenerEvent<F>> + Unpin,
S: TryStream<Ok = ListenerEvent<F>>,
F: TryFuture<Ok = (I, C)>,
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>, Output = D> + Clone
@ -401,11 +388,11 @@ where
type Item = Result<ListenerEvent<ListenerUpgradeFuture<F, U, I, C>>, TransportUpgradeError<S::Error, U::Error>>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
match ready!(TryStream::try_poll_next(Pin::new(&mut self.stream), cx)) {
match ready!(TryStream::try_poll_next(self.stream.as_mut(), cx)) {
Some(Ok(event)) => {
let event = event.map(move |future| {
ListenerUpgradeFuture {
future,
future: Box::pin(future),
upgrade: future::Either::Left(Some(self.upgrade.clone()))
}
});
@ -428,13 +415,13 @@ where
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>>
{
future: F,
future: Pin<Box<F>>,
upgrade: future::Either<Option<U>, (Option<I>, InboundUpgradeApply<C, U>)>
}
impl<F, U, I, C, D> Future for ListenerUpgradeFuture<F, U, I, C>
where
F: TryFuture<Ok = (I, C)> + Unpin,
F: TryFuture<Ok = (I, C)>,
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>, Output = D>,
U::Error: Error
@ -449,7 +436,7 @@ where
loop {
this.upgrade = match this.upgrade {
future::Either::Left(ref mut up) => {
let (i, c) = match ready!(TryFuture::try_poll(Pin::new(&mut this.future), cx).map_err(TransportUpgradeError::Transport)) {
let (i, c) = match ready!(TryFuture::try_poll(this.future.as_mut(), cx).map_err(TransportUpgradeError::Transport)) {
Ok(v) => v,
Err(err) => return Poll::Ready(Err(err))
};

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

@ -86,7 +86,7 @@ where
upgrade: U,
},
Upgrade {
future: U::Future
future: Pin<Box<U::Future>>
},
Undefined
}
@ -102,7 +102,6 @@ impl<C, U> Future for InboundUpgradeApply<C, U>
where
C: AsyncRead + AsyncWrite + Unpin,
U: InboundUpgrade<Negotiated<C>>,
U::Future: Unpin,
{
type Output = Result<U::Output, UpgradeError<U::Error>>;
@ -118,7 +117,7 @@ where
}
};
self.inner = InboundUpgradeApplyState::Upgrade {
future: upgrade.upgrade_inbound(Compat01As03::new(io), info.0)
future: Box::pin(upgrade.upgrade_inbound(Compat01As03::new(io), info.0))
};
}
InboundUpgradeApplyState::Upgrade { mut future } => {
@ -163,7 +162,7 @@ where
upgrade: U
},
Upgrade {
future: U::Future
future: Pin<Box<U::Future>>
},
Undefined
}
@ -179,7 +178,6 @@ impl<C, U> Future for OutboundUpgradeApply<C, U>
where
C: AsyncRead + AsyncWrite + Unpin,
U: OutboundUpgrade<Negotiated<C>>,
U::Future: Unpin,
{
type Output = Result<U::Output, UpgradeError<U::Error>>;
@ -195,7 +193,7 @@ where
}
};
self.inner = OutboundUpgradeApplyState::Upgrade {
future: upgrade.upgrade_outbound(Compat01As03::new(connection), info.0)
future: Box::pin(upgrade.upgrade_outbound(Compat01As03::new(connection), info.0))
};
}
OutboundUpgradeApplyState::Upgrade { mut future } => {

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

@ -144,7 +144,7 @@ pub trait InboundUpgrade<C>: UpgradeInfo {
/// Possible error during the handshake.
type Error;
/// Future that performs the handshake with the remote.
type Future: Future<Output = Result<Self::Output, Self::Error>> + Unpin;
type Future: Future<Output = Result<Self::Output, Self::Error>>;
/// After we have determined that the remote supports one of the protocols we support, this
/// method is called to start the handshake.
@ -184,7 +184,7 @@ pub trait OutboundUpgrade<C>: UpgradeInfo {
/// Possible error during the handshake.
type Error;
/// Future that performs the handshake with the remote.
type Future: Future<Output = Result<Self::Output, Self::Error>> + Unpin;
type Future: Future<Output = Result<Self::Output, Self::Error>>;
/// After we have determined that the remote supports one of the protocols we support, this
/// method is called to start the handshake.