forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(lib): update to std::future::Future

Browse Source 
BREAKING CHANGE: All usage of async traits (`Future`, `Stream`,
`AsyncRead`, `AsyncWrite`, etc) are updated to newer versions.
This commit is contained in:
Sean McArthur
2019-07-09 15:37:43 -07:00
parent da9b0319ef
commit 8f4b05ae78
37 changed files with 1526 additions and 1548 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
src/client/conn.rs
View File

@@ -13,13 +13,12 @@ use std::mem;
use std::sync::Arc;
use bytes::Bytes;
use futures::{Async, Future, Poll};
use futures::future::{self, Either, Executor};
use futures_util::future::{self, Either, FutureExt as _};
use h2;
use tokio_io::{AsyncRead, AsyncWrite};
use crate::body::Payload;
use crate::common::Exec;
use crate::common::{Exec, Future, Pin, Poll, task};
use crate::upgrade::Upgraded;
use crate::proto;
use super::dispatch;
@@ -41,7 +40,7 @@ type ConnEither<T, B> = Either<
/// This is a shortcut for `Builder::new().handshake(io)`.
pub fn handshake<T>(io: T) -> Handshake<T, crate::Body>
where
T: AsyncRead + AsyncWrite + Send + 'static,
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
Builder::new()
.handshake(io)
@@ -88,7 +87,7 @@ pub struct Builder {
pub struct Handshake<T, B> {
builder: Builder,
io: Option<T>,
_marker: PhantomData<B>,
_marker: PhantomData<fn(B)>,
}
/// A future returned by `SendRequest::send_request`.
@@ -96,9 +95,13 @@ pub struct Handshake<T, B> {
/// Yields a `Response` if successful.
#[must_use = "futures do nothing unless polled"]
pub struct ResponseFuture {
// for now, a Box is used to hide away the internal `B`
// that can be returned if canceled
inner: Box<dyn Future<Item=Response<Body>, Error=crate::Error> + Send>,
inner: ResponseFutureState
}
enum ResponseFutureState {
Waiting(dispatch::Promise<Response<Body>>),
// Option is to be able to `take()` it in `poll`
Error(Option<crate::Error>),
}
/// Deconstructed parts of a `Connection`.
@@ -123,14 +126,6 @@ pub struct Parts<T> {
// ========== internal client api
/// A `Future` for when `SendRequest::poll_ready()` is ready.
// FIXME: allow() required due to `impl Trait` leaking types to this lint
#[allow(missing_debug_implementations)]
#[must_use = "futures do nothing unless polled"]
pub(super) struct WhenReady<B> {
tx: Option<SendRequest<B>>,
}
// A `SendRequest` that can be cloned to send HTTP2 requests.
// private for now, probably not a great idea of a type...
#[must_use = "futures do nothing unless polled"]
@@ -145,14 +140,16 @@ impl<B> SendRequest<B>
/// Polls to determine whether this sender can be used yet for a request.
///
/// If the associated connection is closed, this returns an Error.
pub fn poll_ready(&mut self) -> Poll<(), crate::Error> {
self.dispatch.poll_ready()
pub fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
self.dispatch.poll_ready(cx)
}
pub(super) fn when_ready(self) -> WhenReady<B> {
WhenReady {
tx: Some(self),
}
pub(super) fn when_ready(self) -> impl Future<Output=crate::Result<Self>> {
let mut me = Some(self);
future::poll_fn(move |cx| {
ready!(me.as_mut().unwrap().poll_ready(cx))?;
Poll::Ready(Ok(me.take().unwrap()))
})
}
pub(super) fn is_ready(&self) -> bool {
@@ -224,37 +221,30 @@ where
pub fn send_request(&mut self, req: Request<B>) -> ResponseFuture {
let inner = match self.dispatch.send(req) {
Ok(rx) => {
Either::A(rx.then(move |res| {
match res {
Ok(Ok(res)) => Ok(res),
Ok(Err(err)) => Err(err),
// this is definite bug if it happens, but it shouldn't happen!
Err(_) => panic!("dispatch dropped without returning error"),
}
}))
ResponseFutureState::Waiting(rx)
},
Err(_req) => {
debug!("connection was not ready");
let err = crate::Error::new_canceled().with("connection was not ready");
Either::B(future::err(err))
ResponseFutureState::Error(Some(err))
}
};
ResponseFuture {
inner: Box::new(inner),
inner,
}
}
pub(crate) fn send_request_retryable(&mut self, req: Request<B>) -> impl Future<Item = Response<Body>, Error = (crate::Error, Option<Request<B>>)>
pub(crate) fn send_request_retryable(&mut self, req: Request<B>) -> impl Future<Output = Result<Response<Body>, (crate::Error, Option<Request<B>>)>> + Unpin
where
B: Send,
{
match self.dispatch.try_send(req) {
Ok(rx) => {
Either::A(rx.then(move |res| {
Either::Left(rx.then(move |res| {
match res {
Ok(Ok(res)) => Ok(res),
Ok(Err(err)) => Err(err),
Ok(Ok(res)) => future::ok(res),
Ok(Err(err)) => future::err(err),
// this is definite bug if it happens, but it shouldn't happen!
Err(_) => panic!("dispatch dropped without returning error"),
}
@@ -263,7 +253,7 @@ where
Err(req) => {
debug!("connection was not ready");
let err = crate::Error::new_canceled().with("connection was not ready");
Either::B(future::err((err, Some(req))))
Either::Right(future::err((err, Some(req))))
}
}
}
@@ -305,16 +295,16 @@ impl<B> Http2SendRequest<B>
where
B: Payload + 'static,
{
pub(super) fn send_request_retryable(&mut self, req: Request<B>) -> impl Future<Item=Response<Body>, Error=(crate::Error, Option<Request<B>>)>
pub(super) fn send_request_retryable(&mut self, req: Request<B>) -> impl Future<Output=Result<Response<Body>, (crate::Error, Option<Request<B>>)>>
where
B: Send,
{
match self.dispatch.try_send(req) {
Ok(rx) => {
Either::A(rx.then(move |res| {
Either::Left(rx.then(move |res| {
match res {
Ok(Ok(res)) => Ok(res),
Ok(Err(err)) => Err(err),
Ok(Ok(res)) => future::ok(res),
Ok(Err(err)) => future::err(err),
// this is definite bug if it happens, but it shouldn't happen!
Err(_) => panic!("dispatch dropped without returning error"),
}
@@ -323,7 +313,7 @@ where
Err(req) => {
debug!("connection was not ready");
let err = crate::Error::new_canceled().with("connection was not ready");
Either::B(future::err((err, Some(req))))
Either::Right(future::err((err, Some(req))))
}
}
}
@@ -348,7 +338,7 @@ impl<B> Clone for Http2SendRequest<B> {
impl<T, B> Connection<T, B>
where
T: AsyncRead + AsyncWrite + Send + 'static,
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
B: Payload + 'static,
{
/// Return the inner IO object, and additional information.
@@ -356,8 +346,8 @@ where
/// Only works for HTTP/1 connections. HTTP/2 connections will panic.
pub fn into_parts(self) -> Parts<T> {
let (io, read_buf, _) = match self.inner.expect("already upgraded") {
Either::A(h1) => h1.into_inner(),
Either::B(_h2) => {
Either::Left(h1) => h1.into_inner(),
Either::Right(_h2) => {
panic!("http2 cannot into_inner");
}
};
@@ -380,51 +370,58 @@ where
/// Use [`poll_fn`](https://docs.rs/futures/0.1.25/futures/future/fn.poll_fn.html)
/// and [`try_ready!`](https://docs.rs/futures/0.1.25/futures/macro.try_ready.html)
/// to work with this function; or use the `without_shutdown` wrapper.
pub fn poll_without_shutdown(&mut self) -> Poll<(), crate::Error> {
pub fn poll_without_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>>
where
B: Unpin,
{
match self.inner.as_mut().expect("already upgraded") {
&mut Either::A(ref mut h1) => {
h1.poll_without_shutdown()
&mut Either::Left(ref mut h1) => {
h1.poll_without_shutdown(cx)
},
&mut Either::B(ref mut h2) => {
&mut Either::Right(ref mut h2) => {
unimplemented!("h2 poll_without_shutdown");
/*
h2.poll().map(|x| x.map(|_| ()))
*/
}
}
}
/// Prevent shutdown of the underlying IO object at the end of service the request,
/// instead run `into_parts`. This is a convenience wrapper over `poll_without_shutdown`.
pub fn without_shutdown(self) -> impl Future<Item=Parts<T>, Error=crate::Error> {
pub fn without_shutdown(self) -> impl Future<Output=crate::Result<Parts<T>>>
where
B: Unpin,
{
let mut conn = Some(self);
::futures::future::poll_fn(move || -> crate::Result<_> {
try_ready!(conn.as_mut().unwrap().poll_without_shutdown());
Ok(conn.take().unwrap().into_parts().into())
future::poll_fn(move |cx| -> Poll<crate::Result<Parts<T>>> {
ready!(conn.as_mut().unwrap().poll_without_shutdown(cx))?;
Poll::Ready(Ok(conn.take().unwrap().into_parts()))
})
}
}
impl<T, B> Future for Connection<T, B>
where
T: AsyncRead + AsyncWrite + Send + 'static,
B: Payload + 'static,
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
B: Payload + Unpin + 'static,
{
type Item = ();
type Error = crate::Error;
type Output = crate::Result<()>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match try_ready!(self.inner.poll()) {
Some(proto::Dispatched::Shutdown) |
None => {
Ok(Async::Ready(()))
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match ready!(Pin::new(self.inner.as_mut().unwrap()).poll(cx))? {
proto::Dispatched::Shutdown => {
Poll::Ready(Ok(()))
},
Some(proto::Dispatched::Upgrade(pending)) => {
proto::Dispatched::Upgrade(pending) => {
let h1 = match mem::replace(&mut self.inner, None) {
Some(Either::A(h1)) => h1,
Some(Either::Left(h1)) => h1,
_ => unreachable!("Upgrade expects h1"),
};
let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(Box::new(io), buf));
Ok(Async::Ready(()))
Poll::Ready(Ok(()))
}
}
}
@@ -461,6 +458,7 @@ impl Builder {
}
}
/*
/// Provide an executor to execute background HTTP2 tasks.
pub fn executor<E>(&mut self, exec: E) -> &mut Builder
where
@@ -469,6 +467,7 @@ impl Builder {
self.exec = Exec::Executor(Arc::new(exec));
self
}
*/
pub(super) fn h1_writev(&mut self, enabled: bool) -> &mut Builder {
self.h1_writev = enabled;
@@ -532,7 +531,7 @@ impl Builder {
#[inline]
pub fn handshake<T, B>(&self, io: T) -> Handshake<T, B>
where
T: AsyncRead + AsyncWrite + Send + 'static,
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
B: Payload + 'static,
{
trace!("client handshake HTTP/{}", if self.http2 { 2 } else { 1 });
@@ -548,13 +547,12 @@ impl Builder {
impl<T, B> Future for Handshake<T, B>
where
T: AsyncRead + AsyncWrite + Send + 'static,
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
B: Payload + 'static,
{
type Item = (SendRequest<B>, Connection<T, B>);
type Error = crate::Error;
type Output = crate::Result<(SendRequest<B>, Connection<T, B>)>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
let io = self.io.take().expect("polled more than once");
let (tx, rx) = dispatch::channel();
let either = if !self.builder.http2 {
@@ -573,13 +571,13 @@ where
}
let cd = proto::h1::dispatch::Client::new(rx);
let dispatch = proto::h1::Dispatcher::new(cd, conn);
Either::A(dispatch)
Either::Left(dispatch)
} else {
let h2 = proto::h2::Client::new(io, rx, &self.builder.h2_builder, self.builder.exec.clone());
Either::B(h2)
Either::Right(h2)
};
Ok(Async::Ready((
Poll::Ready(Ok((
SendRequest {
dispatch: tx,
},
@@ -600,12 +598,22 @@ impl<T, B> fmt::Debug for Handshake<T, B> {
// ===== impl ResponseFuture
impl Future for ResponseFuture {
type Item = Response<Body>;
type Error = crate::Error;
type Output = crate::Result<Response<Body>>;
#[inline]
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match self.inner {
ResponseFutureState::Waiting(ref mut rx) => {
Pin::new(rx).poll(cx).map(|res| match res {
Ok(Ok(resp)) => Ok(resp),
Ok(Err(err)) => Err(err),
// this is definite bug if it happens, but it shouldn't happen!
Err(_canceled) => panic!("dispatch dropped without returning error"),
})
},
ResponseFutureState::Error(ref mut err) => {
Poll::Ready(Err(err.take().expect("polled after ready")))
}
}
}
}
@@ -616,24 +624,6 @@ impl fmt::Debug for ResponseFuture {
}
}
// ===== impl WhenReady
impl<B> Future for WhenReady<B> {
type Item = SendRequest<B>;
type Error = crate::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let mut tx = self.tx.take().expect("polled after complete");
match tx.poll_ready()? {
Async::Ready(()) => Ok(Async::Ready(tx)),
Async::NotReady => {
self.tx = Some(tx);
Ok(Async::NotReady)
}
}
}
}
// assert trait markers
trait AssertSend: Send {}

83
src/client/connect/dns.rs
View File

@@ -16,20 +16,19 @@ use std::net::{
use std::str::FromStr;
use std::sync::Arc;
use futures::{Async, Future, Poll};
use futures::future::{Executor, ExecuteError};
use futures::sync::oneshot;
use futures_cpupool::{Builder as CpuPoolBuilder};
use tokio_executor::TypedExecutor;
use tokio_sync::oneshot;
use tokio_threadpool;
use crate::common::{Future, Pin, Poll, Unpin, task};
use self::sealed::GaiTask;
/// Resolve a hostname to a set of IP addresses.
pub trait Resolve {
pub trait Resolve: Unpin {
/// The set of IP addresses to try to connect to.
type Addrs: Iterator<Item=IpAddr>;
/// A Future of the resolved set of addresses.
type Future: Future<Item=Self::Addrs, Error=io::Error>;
type Future: Future<Output=Result<Self::Addrs, io::Error>> + Unpin;
/// Resolve a hostname.
fn resolve(&self, name: Name) -> Self::Future;
}
@@ -53,7 +52,8 @@ pub struct GaiAddrs {
/// A future to resole a name returned by `GaiResolver`.
pub struct GaiFuture {
rx: oneshot::SpawnHandle<IpAddrs, io::Error>,
//rx: oneshot::SpawnHandle<IpAddrs, io::Error>,
blocking: GaiBlocking,
}
impl Name {
@@ -112,24 +112,36 @@ impl GaiResolver {
///
/// Takes number of DNS worker threads.
pub fn new(threads: usize) -> Self {
let pool = CpuPoolBuilder::new()
.name_prefix("hyper-dns")
.pool_size(threads)
.create();
GaiResolver {
executor: GaiExecutor,
}
/*
use tokio_threadpool::Builder;
let pool = Builder::new()
.name_prefix("hyper-dns-gai-resolver")
// not for CPU tasks, so only spawn workers
// in blocking mode
.pool_size(1)
.max_blocking(threads)
.build();
GaiResolver::new_with_executor(pool)
*/
}
/*
/// Construct a new `GaiResolver` with a shared thread pool executor.
///
/// Takes an executor to run blocking `getaddrinfo` tasks on.
pub fn new_with_executor<E: 'static>(executor: E) -> Self
/*pub */fn new_with_executor<E: 'static>(executor: E) -> Self
where
E: Executor<GaiTask> + Send + Sync,
E: TypedExecutor<GaiTask> + Send + Sync,
{
GaiResolver {
executor: GaiExecutor(Arc::new(executor)),
}
}
*/
}
impl Resolve for GaiResolver {
@@ -138,9 +150,10 @@ impl Resolve for GaiResolver {
fn resolve(&self, name: Name) -> Self::Future {
let blocking = GaiBlocking::new(name.host);
let rx = oneshot::spawn(blocking, &self.executor);
//let rx = oneshot::spawn(blocking, &self.executor);
GaiFuture {
rx,
//rx,
blocking,
}
}
}
@@ -152,14 +165,16 @@ impl fmt::Debug for GaiResolver {
}
impl Future for GaiFuture {
type Item = GaiAddrs;
type Error = io::Error;
type Output = Result<GaiAddrs, io::Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
/*
let addrs = try_ready!(self.rx.poll());
Ok(Async::Ready(GaiAddrs {
inner: addrs,
}))
*/
Poll::Ready(self.blocking.block().map(|addrs| GaiAddrs { inner: addrs }))
}
}
@@ -184,14 +199,16 @@ impl fmt::Debug for GaiAddrs {
}
#[derive(Clone)]
struct GaiExecutor(Arc<dyn Executor<GaiTask> + Send + Sync>);
struct GaiExecutor/*(Arc<dyn Executor<GaiTask> + Send + Sync>)*/;
/*
impl Executor<oneshot::Execute<GaiBlocking>> for GaiExecutor {
fn execute(&self, future: oneshot::Execute<GaiBlocking>) -> Result<(), ExecuteError<oneshot::Execute<GaiBlocking>>> {
self.0.execute(GaiTask { work: future })
.map_err(|err| ExecuteError::new(err.kind(), err.into_future().work))
}
}
*/
pub(super) struct GaiBlocking {
host: String,
@@ -201,8 +218,16 @@ impl GaiBlocking {
pub(super) fn new(host: String) -> GaiBlocking {
GaiBlocking { host }
}
fn block(&self) -> io::Result<IpAddrs> {
debug!("resolving host={:?}", self.host);
(&*self.host, 0).to_socket_addrs()
.map(|i| IpAddrs { iter: i })
}
}
/*
impl Future for GaiBlocking {
type Item = IpAddrs;
type Error = io::Error;
@@ -213,6 +238,7 @@ impl Future for GaiBlocking {
.map(|i| Async::Ready(IpAddrs { iter: i }))
}
}
*/
pub(super) struct IpAddrs {
iter: vec::IntoIter<SocketAddr>,
@@ -276,7 +302,7 @@ pub(super) mod sealed {
use super::*;
// Blocking task to be executed on a thread pool.
pub struct GaiTask {
pub(super) work: oneshot::Execute<GaiBlocking>
//pub(super) work: oneshot::Execute<GaiBlocking>
}
impl fmt::Debug for GaiTask {
@@ -285,6 +311,7 @@ pub(super) mod sealed {
}
}
/*
impl Future for GaiTask {
type Item = ();
type Error = ();
@@ -293,6 +320,7 @@ pub(super) mod sealed {
self.work.poll()
}
}
*/
}
@@ -329,15 +357,14 @@ impl Resolve for TokioThreadpoolGaiResolver {
}
impl Future for TokioThreadpoolGaiFuture {
type Item = GaiAddrs;
type Error = io::Error;
type Output = Result<GaiAddrs, io::Error>;
fn poll(&mut self) -> Poll<GaiAddrs, io::Error> {
match tokio_threadpool::blocking(|| (self.name.as_str(), 0).to_socket_addrs()) {
Ok(Async::Ready(Ok(iter))) => Ok(Async::Ready(GaiAddrs { inner: IpAddrs { iter } })),
Ok(Async::Ready(Err(e))) => Err(e),
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)),
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match ready!(tokio_threadpool::blocking(|| (self.name.as_str(), 0).to_socket_addrs())) {
Ok(Ok(iter)) => Poll::Ready(Ok(GaiAddrs { inner: IpAddrs { iter } })),
Ok(Err(e)) => Poll::Ready(Err(e)),
// a BlockingError, meaning not on a tokio_threadpool :(
Err(e) => Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
}
}
}

123
src/client/connect/http.rs
View File

@@ -6,17 +6,19 @@ use std::mem;
use std::net::{IpAddr, SocketAddr};
use std::time::{Duration, Instant};
use futures::{Async, Future, Poll};
use futures::future::{Executor};
use http::uri::Scheme;
use net2::TcpBuilder;
use tokio_reactor::Handle;
use tokio_tcp::{TcpStream, ConnectFuture};
use tokio_tcp::{TcpStream/*, ConnectFuture*/};
use tokio_timer::Delay;
use crate::common::{Future, Pin, Poll, task};
use super::{Connect, Connected, Destination};
use super::dns::{self, GaiResolver, Resolve, TokioThreadpoolGaiResolver};
// TODO: unbox me?
type ConnectFuture = Pin<Box<dyn Future<Output = io::Result<TcpStream>> + Send>>;
/// A connector for the `http` scheme.
///
/// Performs DNS resolution in a thread pool, and then connects over TCP.
@@ -89,6 +91,7 @@ impl HttpConnector {
http
}
/*
/// Construct a new HttpConnector.
///
/// Takes an executor to run blocking `getaddrinfo` tasks on.
@@ -100,6 +103,7 @@ impl HttpConnector {
http.set_reactor(handle);
http
}
*/
}
impl HttpConnector<TokioThreadpoolGaiResolver> {
@@ -335,55 +339,54 @@ enum State<R: Resolve> {
Error(Option<io::Error>),
}
impl<R: Resolve> Future for HttpConnecting<R> {
type Item = (TcpStream, Connected);
type Error = io::Error;
impl<R: Resolve> Future for HttpConnecting<R>
where
R::Future: Unpin,
{
type Output = Result<(TcpStream, Connected), io::Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
let me = &mut *self;
loop {
let state;
match self.state {
match me.state {
State::Lazy(ref resolver, ref mut host, local_addr) => {
// If the host is already an IP addr (v4 or v6),
// skip resolving the dns and start connecting right away.
if let Some(addrs) = dns::IpAddrs::try_parse(host, self.port) {
if let Some(addrs) = dns::IpAddrs::try_parse(host, me.port) {
state = State::Connecting(ConnectingTcp::new(
local_addr, addrs, self.happy_eyeballs_timeout, self.reuse_address));
local_addr, addrs, me.happy_eyeballs_timeout, me.reuse_address));
} else {
let name = dns::Name::new(mem::replace(host, String::new()));
state = State::Resolving(resolver.resolve(name), local_addr);
}
},
State::Resolving(ref mut future, local_addr) => {
match future.poll()? {
Async::NotReady => return Ok(Async::NotReady),
Async::Ready(addrs) => {
let port = self.port;
let addrs = addrs
.map(|addr| SocketAddr::new(addr, port))
.collect();
let addrs = dns::IpAddrs::new(addrs);
state = State::Connecting(ConnectingTcp::new(
local_addr, addrs, self.happy_eyeballs_timeout, self.reuse_address));
}
};
let addrs = ready!(Pin::new(future).poll(cx))?;
let port = me.port;
let addrs = addrs
.map(|addr| SocketAddr::new(addr, port))
.collect();
let addrs = dns::IpAddrs::new(addrs);
state = State::Connecting(ConnectingTcp::new(
local_addr, addrs, me.happy_eyeballs_timeout, me.reuse_address));
},
State::Connecting(ref mut c) => {
let sock = try_ready!(c.poll(&self.handle));
let sock = ready!(c.poll(cx, &me.handle))?;
if let Some(dur) = self.keep_alive_timeout {
if let Some(dur) = me.keep_alive_timeout {
sock.set_keepalive(Some(dur))?;
}
if let Some(size) = self.send_buffer_size {
if let Some(size) = me.send_buffer_size {
sock.set_send_buffer_size(size)?;
}
if let Some(size) = self.recv_buffer_size {
if let Some(size) = me.recv_buffer_size {
sock.set_recv_buffer_size(size)?;
}
sock.set_nodelay(self.nodelay)?;
sock.set_nodelay(me.nodelay)?;
let extra = HttpInfo {
remote_addr: sock.peer_addr()?,
@@ -391,11 +394,11 @@ impl<R: Resolve> Future for HttpConnecting<R> {
let connected = Connected::new()
.extra(extra);
return Ok(Async::Ready((sock, connected)));
return Poll::Ready(Ok((sock, connected)));
},
State::Error(ref mut e) => return Err(e.take().expect("polled more than once")),
State::Error(ref mut e) => return Poll::Ready(Err(e.take().expect("polled more than once"))),
}
self.state = state;
me.state = state;
}
}
}
@@ -474,20 +477,21 @@ impl ConnectingTcpRemote {
// not a Future, since passing a &Handle to poll
fn poll(
&mut self,
cx: &mut task::Context<'_>,
local_addr: &Option<IpAddr>,
handle: &Option<Handle>,
reuse_address: bool,
) -> Poll<TcpStream, io::Error> {
) -> Poll<io::Result<TcpStream>> {
let mut err = None;
loop {
if let Some(ref mut current) = self.current {
match current.poll() {
Ok(Async::Ready(tcp)) => {
match current.as_mut().poll(cx) {
Poll::Ready(Ok(tcp)) => {
debug!("connected to {:?}", tcp.peer_addr().ok());
return Ok(Async::Ready(tcp));
return Poll::Ready(Ok(tcp));
},
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(e) => {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(e)) => {
trace!("connect error {:?}", e);
err = Some(e);
if let Some(addr) = self.addrs.next() {
@@ -503,7 +507,7 @@ impl ConnectingTcpRemote {
continue;
}
return Err(err.take().expect("missing connect error"));
return Poll::Ready(Err(err.take().expect("missing connect error")));
}
}
}
@@ -540,51 +544,46 @@ fn connect(addr: &SocketAddr, local_addr: &Option<IpAddr>, handle: &Option<Handl
None => Cow::Owned(Handle::default()),
};
Ok(TcpStream::connect_std(builder.to_tcp_stream()?, addr, &handle))
Ok(Box::pin(TcpStream::connect_std(builder.to_tcp_stream()?, addr, &handle)))
}
impl ConnectingTcp {
// not a Future, since passing a &Handle to poll
fn poll(&mut self, handle: &Option<Handle>) -> Poll<TcpStream, io::Error> {
fn poll(&mut self, cx: &mut task::Context<'_>, handle: &Option<Handle>) -> Poll<io::Result<TcpStream>> {
match self.fallback.take() {
None => self.preferred.poll(&self.local_addr, handle, self.reuse_address),
Some(mut fallback) => match self.preferred.poll(&self.local_addr, handle, self.reuse_address) {
Ok(Async::Ready(stream)) => {
None => self.preferred.poll(cx, &self.local_addr, handle, self.reuse_address),
Some(mut fallback) => match self.preferred.poll(cx, &self.local_addr, handle, self.reuse_address) {
Poll::Ready(Ok(stream)) => {
// Preferred successful - drop fallback.
Ok(Async::Ready(stream))
Poll::Ready(Ok(stream))
}
Ok(Async::NotReady) => match fallback.delay.poll() {
Ok(Async::Ready(_)) => match fallback.remote.poll(&self.local_addr, handle, self.reuse_address) {
Ok(Async::Ready(stream)) => {
Poll::Pending => match Pin::new(&mut fallback.delay).poll(cx) {
Poll::Ready(()) => match fallback.remote.poll(cx, &self.local_addr, handle, self.reuse_address) {
Poll::Ready(Ok(stream)) => {
// Fallback successful - drop current preferred,
// but keep fallback as new preferred.
self.preferred = fallback.remote;
Ok(Async::Ready(stream))
Poll::Ready(Ok(stream))
}
Ok(Async::NotReady) => {
Poll::Pending => {
// Neither preferred nor fallback are ready.
self.fallback = Some(fallback);
Ok(Async::NotReady)
Poll::Pending
}
Err(_) => {
Poll::Ready(Err(_)) => {
// Fallback failed - resume with preferred only.
Ok(Async::NotReady)
Poll::Pending
}
},
Ok(Async::NotReady) => {
Poll::Pending => {
// Too early to attempt fallback.
self.fallback = Some(fallback);
Ok(Async::NotReady)
}
Err(_) => {
// Fallback delay failed - resume with preferred only.
Ok(Async::NotReady)
Poll::Pending
}
}
Err(_) => {
Poll::Ready(Err(_)) => {
// Preferred failed - use fallback as new preferred.
self.preferred = fallback.remote;
self.preferred.poll(&self.local_addr, handle, self.reuse_address)
self.preferred.poll(cx, &self.local_addr, handle, self.reuse_address)
}
}
}
@@ -734,10 +733,10 @@ mod tests {
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.0.poll(&Some(Handle::default())) {
Ok(Async::Ready(stream)) => Ok(Async::Ready(
Poll::Ready(Ok(stream)) => Poll::Ready(Ok(
if stream.peer_addr().unwrap().is_ipv4() { 4 } else { 6 }
)),
Ok(Async::NotReady) => Ok(Async::NotReady),
Poll::Pending => Poll::Pending,
Err(err) => Err(err),
}
}

7
src/client/connect/mod.rs
View File

@@ -10,10 +10,11 @@ use std::{fmt, mem};
#[cfg(try_from)] use std::convert::TryFrom;
use bytes::{BufMut, Bytes, BytesMut};
use futures::Future;
use ::http::{uri, Response, Uri};
use tokio_io::{AsyncRead, AsyncWrite};
use crate::common::{Future, Unpin};
#[cfg(feature = "runtime")] pub mod dns;
#[cfg(feature = "runtime")] mod http;
#[cfg(feature = "runtime")] pub use self::http::{HttpConnector, HttpInfo};
@@ -25,11 +26,11 @@ use tokio_io::{AsyncRead, AsyncWrite};
/// ready connection.
pub trait Connect: Send + Sync {
/// The connected IO Stream.
type Transport: AsyncRead + AsyncWrite + Send + 'static;
type Transport: AsyncRead + AsyncWrite + Unpin + Send + 'static;
/// An error occured when trying to connect.
type Error: Into<Box<dyn StdError + Send + Sync>>;
/// A Future that will resolve to the connected Transport.
type Future: Future<Item=(Self::Transport, Connected), Error=Self::Error> + Send;
type Future: Future<Output=Result<(Self::Transport, Connected), Self::Error>> + Unpin + Send;
/// Connect to a destination.
fn connect(&self, dst: Destination) -> Self::Future;
}

73
src/client/dispatch.rs
View File

@@ -1,8 +1,9 @@
use futures::{future, Async, Future, Poll, Stream};
use futures::sync::{mpsc, oneshot};
use futures_core::Stream;
use futures_channel::{mpsc, oneshot};
use futures_util::future;
use want;
use crate::common::Never;
use crate::common::{Future, Never, Pin, Poll, task};
pub type RetryPromise<T, U> = oneshot::Receiver<Result<U, (crate::Error, Option<T>)>>;
pub type Promise<T> = oneshot::Receiver<Result<T, crate::Error>>;
@@ -51,8 +52,8 @@ pub struct UnboundedSender<T, U> {
}
impl<T, U> Sender<T, U> {
pub fn poll_ready(&mut self) -> Poll<(), crate::Error> {
self.giver.poll_want()
pub fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
self.giver.poll_want(cx)
.map_err(|_| crate::Error::new_closed())
}
@@ -136,20 +137,32 @@ pub struct Receiver<T, U> {
taker: want::Taker,
}
impl<T, U> Stream for Receiver<T, U> {
type Item = (T, Callback<T, U>);
type Error = Never;
//impl<T, U> Stream for Receiver<T, U> {
// type Item = (T, Callback<T, U>);
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
match self.inner.poll() {
Ok(Async::Ready(item)) => Ok(Async::Ready(item.map(|mut env| {
impl<T, U> Receiver<T, U> {
pub(crate) fn poll_next(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<(T, Callback<T, U>)>> {
match Pin::new(&mut self.inner).poll_next(cx) {
Poll::Ready(item) => Poll::Ready(item.map(|mut env| {
env.0.take().expect("envelope not dropped")
}))),
Ok(Async::NotReady) => {
})),
Poll::Pending => {
self.taker.want();
Ok(Async::NotReady)
}
Err(()) => unreachable!("mpsc never errors"),
Poll::Pending
},
}
}
pub(crate) fn close(&mut self) {
self.taker.cancel();
self.inner.close();
}
pub(crate) fn try_recv(&mut self) -> Option<(T, Callback<T, U>)> {
match self.inner.try_next() {
Ok(Some(mut env)) => env.0.take(),
Ok(None) => None,
Err(_) => None,
}
}
}
@@ -185,10 +198,10 @@ impl<T, U> Callback<T, U> {
}
}
pub(crate) fn poll_cancel(&mut self) -> Poll<(), ()> {
pub(crate) fn poll_cancel(&mut self, cx: &mut task::Context<'_>) -> Poll<()> {
match *self {
Callback::Retry(ref mut tx) => tx.poll_cancel(),
Callback::NoRetry(ref mut tx) => tx.poll_cancel(),
Callback::Retry(ref mut tx) => tx.poll_cancel(cx),
Callback::NoRetry(ref mut tx) => tx.poll_cancel(cx),
}
}
@@ -205,30 +218,30 @@ impl<T, U> Callback<T, U> {
pub(crate) fn send_when(
self,
mut when: impl Future<Item=U, Error=(crate::Error, Option<T>)>,
) -> impl Future<Item=(), Error=()> {
mut when: impl Future<Output=Result<U, (crate::Error, Option<T>)>> + Unpin,
) -> impl Future<Output=()> {
let mut cb = Some(self);
// "select" on this callback being canceled, and the future completing
future::poll_fn(move || {
match when.poll() {
Ok(Async::Ready(res)) => {
future::poll_fn(move |cx| {
match Pin::new(&mut when).poll(cx) {
Poll::Ready(Ok(res)) => {
cb.take()
.expect("polled after complete")
.send(Ok(res));
Ok(().into())
Poll::Ready(())
},
Ok(Async::NotReady) => {
Poll::Pending => {
// check if the callback is canceled
try_ready!(cb.as_mut().unwrap().poll_cancel());
ready!(cb.as_mut().unwrap().poll_cancel(cx));
trace!("send_when canceled");
Ok(().into())
Poll::Ready(())
},
Err(err) => {
Poll::Ready(Err(err)) => {
cb.take()
.expect("polled after complete")
.send(Err(err));
Ok(().into())
Poll::Ready(())
}
}
})

156
src/client/mod.rs
View File

@@ -82,15 +82,15 @@ use std::mem;
use std::sync::Arc;
use std::time::Duration;
use futures::{Async, Future, Poll};
use futures::future::{self, Either, Executor};
use futures::sync::oneshot;
use futures_channel::oneshot;
use futures_util::future::{self, FutureExt as _, Either};
use futures_util::try_future::TryFutureExt as _;
use http::{Method, Request, Response, Uri, Version};
use http::header::{HeaderValue, HOST};
use http::uri::Scheme;
use crate::body::{Body, Payload};
use crate::common::{lazy as hyper_lazy, Lazy};
use crate::common::{lazy as hyper_lazy, Lazy, Future, Pin, Poll, task};
use self::connect::{Alpn, Connect, Connected, Destination};
use self::pool::{Key as PoolKey, Pool, Poolable, Pooled, Reservation};
@@ -118,6 +118,16 @@ struct Config {
ver: Ver,
}
/// A `Future` that will resolve to an HTTP Response.
///
/// This is returned by `Client::request` (and `Client::get`).
#[must_use = "futures do nothing unless polled"]
pub struct ResponseFuture {
inner: Pin<Box<dyn Future<Output=crate::Result<Response<Body>>> + Send>>,
}
// ===== impl Client =====
#[cfg(feature = "runtime")]
impl Client<HttpConnector, Body> {
/// Create a new Client with the default [config](Builder).
@@ -170,7 +180,7 @@ impl<C, B> Client<C, B>
where C: Connect + Sync + 'static,
C::Transport: 'static,
C::Future: 'static,
B: Payload + Send + 'static,
B: Payload + Unpin + Send + 'static,
B::Data: Send,
{
/// Send a `GET` request to the supplied `Uri`.
@@ -257,16 +267,15 @@ where C: Connect + Sync + 'static,
ResponseFuture::new(Box::new(self.retryably_send_request(req, pool_key)))
}
fn retryably_send_request(&self, req: Request<B>, pool_key: PoolKey) -> impl Future<Item=Response<Body>, Error=crate::Error> {
fn retryably_send_request(&self, req: Request<B>, pool_key: PoolKey) -> impl Future<Output=crate::Result<Response<Body>>> {
let client = self.clone();
let uri = req.uri().clone();
let mut send_fut = client.send_request(req, pool_key.clone());
future::poll_fn(move || loop {
match send_fut.poll() {
Ok(Async::Ready(resp)) => return Ok(Async::Ready(resp)),
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(ClientError::Normal(err)) => return Err(err),
future::poll_fn(move |cx| loop {
match ready!(Pin::new(&mut send_fut).poll(cx)) {
Ok(resp) => return Poll::Ready(Ok(resp)),
Err(ClientError::Normal(err)) => return Poll::Ready(Err(err)),
Err(ClientError::Canceled {
connection_reused,
mut req,
@@ -275,7 +284,7 @@ where C: Connect + Sync + 'static,
if !client.config.retry_canceled_requests || !connection_reused {
// if client disabled, don't retry
// a fresh connection means we definitely can't retry
return Err(reason);
return Poll::Ready(Err(reason));
}
trace!("unstarted request canceled, trying again (reason={:?})", reason);
@@ -286,7 +295,7 @@ where C: Connect + Sync + 'static,
})
}
fn send_request(&self, mut req: Request<B>, pool_key: PoolKey) -> impl Future<Item=Response<Body>, Error=ClientError<B>> {
fn send_request(&self, mut req: Request<B>, pool_key: PoolKey) -> impl Future<Output=Result<Response<Body>, ClientError<B>>> + Unpin {
let conn = self.connection_for(req.uri().clone(), pool_key);
let set_host = self.config.set_host;
@@ -320,7 +329,7 @@ where C: Connect + Sync + 'static,
};
} else if req.method() == &Method::CONNECT {
debug!("client does not support CONNECT requests over HTTP2");
return Either::A(future::err(ClientError::Normal(crate::Error::new_user_unsupported_request_method())));
return Either::Left(future::err(ClientError::Normal(crate::Error::new_user_unsupported_request_method())));
}
let fut = pooled.send_request_retryable(req)
@@ -328,7 +337,7 @@ where C: Connect + Sync + 'static,
// If the Connector included 'extra' info, add to Response...
let extra_info = pooled.conn_info.extra.clone();
let fut = fut.map(move |mut res| {
let fut = fut.map_ok(move |mut res| {
if let Some(extra) = extra_info {
extra.set(&mut res);
}
@@ -343,11 +352,11 @@ where C: Connect + Sync + 'static,
// To counteract this, we must check if our senders 'want' channel
// has been closed after having tried to send. If so, error out...
if pooled.is_closed() {
return Either::B(Either::A(fut));
return Either::Right(Either::Left(fut));
}
Either::B(Either::B(fut
.and_then(move |mut res| {
Either::Right(Either::Right(fut
.map_ok(move |mut res| {
// If pooled is HTTP/2, we can toss this reference immediately.
//
// when pooled is dropped, it will try to insert back into the
@@ -363,14 +372,13 @@ where C: Connect + Sync + 'static,
} else if !res.body().is_end_stream() {
let (delayed_tx, delayed_rx) = oneshot::channel();
res.body_mut().delayed_eof(delayed_rx);
let on_idle = future::poll_fn(move || {
pooled.poll_ready()
let on_idle = future::poll_fn(move |cx| {
pooled.poll_ready(cx)
})
.then(move |_| {
.map(move |_| {
// At this point, `pooled` is dropped, and had a chance
// to insert into the pool (if conn was idle)
drop(delayed_tx);
Ok(())
});
if let Err(err) = executor.execute(on_idle) {
@@ -380,23 +388,23 @@ where C: Connect + Sync + 'static,
} else {
// There's no body to delay, but the connection isn't
// ready yet. Only re-insert when it's ready
let on_idle = future::poll_fn(move || {
pooled.poll_ready()
let on_idle = future::poll_fn(move |cx| {
pooled.poll_ready(cx)
})
.then(|_| Ok(()));
.map(|_| ());
if let Err(err) = executor.execute(on_idle) {
// This task isn't critical, so just log and ignore.
warn!("error spawning task to insert idle connection: {}", err);
}
}
Ok(res)
res
})))
})
}
fn connection_for(&self, uri: Uri, pool_key: PoolKey)
-> impl Future<Item=Pooled<PoolClient<B>>, Error=ClientError<B>>
-> impl Future<Output=Result<Pooled<PoolClient<B>>, ClientError<B>>>
{
// This actually races 2 different futures to try to get a ready
// connection the fastest, and to reduce connection churn.
@@ -415,15 +423,14 @@ where C: Connect + Sync + 'static,
let connect = self.connect_to(uri, pool_key);
let executor = self.conn_builder.exec.clone();
checkout
// The order of the `select` is depended on below...
.select2(connect)
.map(move |either| match either {
// The order of the `select` is depended on below...
future::select(checkout, connect)
.then(move |either| match either {
// Checkout won, connect future may have been started or not.
//
// If it has, let it finish and insert back into the pool,
// so as to not waste the socket...
Either::A((checked_out, connecting)) => {
Either::Left((Ok(checked_out), connecting)) => {
// This depends on the `select` above having the correct
// order, such that if the checkout future were ready
// immediately, the connect future will never have been
@@ -433,25 +440,23 @@ where C: Connect + Sync + 'static,
// have been started...
if connecting.started() {
let bg = connecting
.map_err(|err| {
trace!("background connect error: {}", err);
})
.map(|_pooled| {
// dropping here should just place it in
// the Pool for us...
})
.map_err(|err| {
trace!("background connect error: {}", err);
});
// An execute error here isn't important, we're just trying
// to prevent a waste of a socket...
let _ = executor.execute(bg);
}
checked_out
Either::Left(future::ok(checked_out))
},
// Connect won, checkout can just be dropped.
Either::B((connected, _checkout)) => {
connected
Either::Right((Ok(connected), _checkout)) => {
Either::Left(future::ok(connected))
},
})
.or_else(|either| match either {
// Either checkout or connect could get canceled:
//
// 1. Connect is canceled if this is HTTP/2 and there is
@@ -460,25 +465,25 @@ where C: Connect + Sync + 'static,
// idle connection reliably.
//
// In both cases, we should just wait for the other future.
Either::A((err, connecting)) => {
Either::Left((Err(err), connecting)) => Either::Right(Either::Left({
if err.is_canceled() {
Either::A(Either::A(connecting.map_err(ClientError::Normal)))
Either::Left(connecting.map_err(ClientError::Normal))
} else {
Either::B(future::err(ClientError::Normal(err)))
Either::Right(future::err(ClientError::Normal(err)))
}
},
Either::B((err, checkout)) => {
})),
Either::Right((Err(err), checkout)) => Either::Right(Either::Right({
if err.is_canceled() {
Either::A(Either::B(checkout.map_err(ClientError::Normal)))
Either::Left(checkout.map_err(ClientError::Normal))
} else {
Either::B(future::err(ClientError::Normal(err)))
Either::Right(future::err(ClientError::Normal(err)))
}
}
})),
})
}
fn connect_to(&self, uri: Uri, pool_key: PoolKey)
-> impl Lazy<Item=Pooled<PoolClient<B>>, Error=crate::Error>
-> impl Lazy<Output=crate::Result<Pooled<PoolClient<B>>>> + Unpin
{
let executor = self.conn_builder.exec.clone();
let pool = self.pool.clone();
@@ -499,10 +504,10 @@ where C: Connect + Sync + 'static,
Some(lock) => lock,
None => {
let canceled = crate::Error::new_canceled().with("HTTP/2 connection in progress");
return Either::B(future::err(canceled));
return Either::Right(future::err(canceled));
}
};
Either::A(connector.connect(dst)
Either::Left(connector.connect(dst)
.map_err(crate::Error::new_connect)
.and_then(move |(io, connected)| {
// If ALPN is h2 and we aren't http2_only already,
@@ -518,34 +523,34 @@ where C: Connect + Sync + 'static,
// Another connection has already upgraded,
// the pool checkout should finish up for us.
let canceled = crate::Error::new_canceled().with("ALPN upgraded to HTTP/2");
return Either::B(future::err(canceled));
return Either::Right(future::err(canceled));
}
}
} else {
connecting
};
let is_h2 = is_ver_h2 || connected.alpn == Alpn::H2;
Either::A(conn_builder
Either::Left(conn_builder
.http2_only(is_h2)
.handshake(io)
.and_then(move |(tx, conn)| {
trace!("handshake complete, spawning background dispatcher task");
let bg = executor.execute(conn.map_err(|e| {
debug!("client connection error: {}", e)
}));
}).map(|_| ()));
// This task is critical, so an execute error
// should be returned.
if let Err(err) = bg {
warn!("error spawning critical client task: {}", err);
return Either::A(future::err(err));
return Either::Left(future::err(err));
}
// Wait for 'conn' to ready up before we
// declare this tx as usable
Either::B(tx.when_ready())
Either::Right(tx.when_ready())
})
.map(move |tx| {
.map_ok(move |tx| {
pool.pooled(connecting, PoolClient {
conn_info: connected,
tx: if is_h2 {
@@ -578,18 +583,12 @@ impl<C, B> fmt::Debug for Client<C, B> {
}
}
/// A `Future` that will resolve to an HTTP Response.
///
/// This is returned by `Client::request` (and `Client::get`).
#[must_use = "futures do nothing unless polled"]
pub struct ResponseFuture {
inner: Box<dyn Future<Item=Response<Body>, Error=crate::Error> + Send>,
}
// ===== impl ResponseFuture =====
impl ResponseFuture {
fn new(fut: Box<dyn Future<Item=Response<Body>, Error=crate::Error> + Send>) -> Self {
fn new(fut: Box<dyn Future<Output=crate::Result<Response<Body>>> + Send>) -> Self {
Self {
inner: fut,
inner: fut.into(),
}
}
@@ -606,14 +605,15 @@ impl fmt::Debug for ResponseFuture {
}
impl Future for ResponseFuture {
type Item = Response<Body>;
type Error = crate::Error;
type Output = crate::Result<Response<Body>>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
Pin::new(&mut self.inner).poll(cx)
}
}
// ===== impl PoolClient =====
// FIXME: allow() required due to `impl Trait` leaking types to this lint
#[allow(missing_debug_implementations)]
struct PoolClient<B> {
@@ -627,10 +627,10 @@ enum PoolTx<B> {
}
impl<B> PoolClient<B> {
fn poll_ready(&mut self) -> Poll<(), crate::Error> {
fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
match self.tx {
PoolTx::Http1(ref mut tx) => tx.poll_ready(),
PoolTx::Http2(_) => Ok(Async::Ready(())),
PoolTx::Http1(ref mut tx) => tx.poll_ready(cx),
PoolTx::Http2(_) => Poll::Ready(Ok(())),
}
}
@@ -661,13 +661,13 @@ impl<B> PoolClient<B> {
}
impl<B: Payload + 'static> PoolClient<B> {
fn send_request_retryable(&mut self, req: Request<B>) -> impl Future<Item = Response<Body>, Error = (crate::Error, Option<Request<B>>)>
fn send_request_retryable(&mut self, req: Request<B>) -> impl Future<Output = Result<Response<Body>, (crate::Error, Option<Request<B>>)>>
where
B: Send,
{
match self.tx {
PoolTx::Http1(ref mut tx) => Either::A(tx.send_request_retryable(req)),
PoolTx::Http2(ref mut tx) => Either::B(tx.send_request_retryable(req)),
PoolTx::Http1(ref mut tx) => Either::Left(tx.send_request_retryable(req)),
PoolTx::Http2(ref mut tx) => Either::Right(tx.send_request_retryable(req)),
}
}
}
@@ -710,6 +710,8 @@ where
}
}
// ===== impl ClientError =====
// FIXME: allow() required due to `impl Trait` leaking types to this lint
#[allow(missing_debug_implementations)]
enum ClientError<B> {
@@ -1027,6 +1029,7 @@ impl Builder {
self
}
/*
/// Provide an executor to execute background `Connection` tasks.
pub fn executor<E>(&mut self, exec: E) -> &mut Self
where
@@ -1035,6 +1038,7 @@ impl Builder {
self.conn_builder.executor(exec);
self
}
*/
/// Builder a client with this configuration and the default `HttpConnector`.
#[cfg(feature = "runtime")]

79
src/client/pool.rs
View File

@@ -4,12 +4,11 @@ use std::ops::{Deref, DerefMut};
use std::sync::{Arc, Mutex, Weak};
use std::time::{Duration, Instant};
use futures::{Future, Async, Poll};
use futures::sync::oneshot;
use futures_channel::oneshot;
#[cfg(feature = "runtime")]
use tokio_timer::Interval;
use crate::common::Exec;
use crate::common::{Exec, Future, Pin, Poll, Unpin, task};
use super::Ver;
// FIXME: allow() required due to `impl Trait` leaking types to this lint
@@ -24,7 +23,7 @@ pub(super) struct Pool<T> {
// This is a trait to allow the `client::pool::tests` to work for `i32`.
//
// See https://github.com/hyperium/hyper/issues/1429
pub(super) trait Poolable: Send + Sized + 'static {
pub(super) trait Poolable: Unpin + Send + Sized + 'static {
fn is_open(&self) -> bool;
/// Reserve this connection.
///
@@ -415,7 +414,7 @@ impl<T: Poolable> PoolInner<T> {
let start = Instant::now() + dur;
let interval = IdleInterval {
let interval = IdleTask {
interval: Interval::new(start, dur),
pool: WeakOpt::downgrade(pool_ref),
pool_drop_notifier: rx,
@@ -449,7 +448,7 @@ impl<T> PoolInner<T> {
#[cfg(feature = "runtime")]
impl<T: Poolable> PoolInner<T> {
/// This should *only* be called by the IdleInterval.
/// This should *only* be called by the IdleTask
fn clear_expired(&mut self) {
let dur = self.timeout.expect("interval assumes timeout");
@@ -569,25 +568,25 @@ pub(super) struct Checkout<T> {
}
impl<T: Poolable> Checkout<T> {
fn poll_waiter(&mut self) -> Poll<Option<Pooled<T>>, crate::Error> {
fn poll_waiter(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<crate::Result<Pooled<T>>>> {
static CANCELED: &str = "pool checkout failed";
if let Some(mut rx) = self.waiter.take() {
match rx.poll() {
Ok(Async::Ready(value)) => {
match Pin::new(&mut rx).poll(cx) {
Poll::Ready(Ok(value)) => {
if value.is_open() {
Ok(Async::Ready(Some(self.pool.reuse(&self.key, value))))
Poll::Ready(Some(Ok(self.pool.reuse(&self.key, value))))
} else {
Err(crate::Error::new_canceled().with(CANCELED))
Poll::Ready(Some(Err(crate::Error::new_canceled().with(CANCELED))))
}
},
Ok(Async::NotReady) => {
Poll::Pending => {
self.waiter = Some(rx);
Ok(Async::NotReady)
Poll::Pending
},
Err(_canceled) => Err(crate::Error::new_canceled().with(CANCELED)),
Poll::Ready(Err(_canceled)) => Poll::Ready(Some(Err(crate::Error::new_canceled().with(CANCELED)))),
}
} else {
Ok(Async::Ready(None))
Poll::Ready(None)
}
}
@@ -622,9 +621,7 @@ impl<T: Poolable> Checkout<T> {
}
if entry.is_none() && self.waiter.is_none() {
let (tx, mut rx) = oneshot::channel();
let _ = rx.poll(); // park this task
let (tx, rx) = oneshot::channel();
trace!("checkout waiting for idle connection: {:?}", self.key);
inner
.waiters
@@ -643,20 +640,21 @@ impl<T: Poolable> Checkout<T> {
}
impl<T: Poolable> Future for Checkout<T> {
type Item = Pooled<T>;
type Error = crate::Error;
type Output = crate::Result<Pooled<T>>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if let Some(pooled) = try_ready!(self.poll_waiter()) {
return Ok(Async::Ready(pooled));
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
if let Some(pooled) = ready!(self.poll_waiter(cx)?) {
return Poll::Ready(Ok(pooled));
}
if let Some(pooled) = self.checkout() {
Ok(Async::Ready(pooled))
Poll::Ready(Ok(pooled))
} else if !self.pool.is_enabled() {
Err(crate::Error::new_canceled().with("pool is disabled"))
Poll::Ready(Err(crate::Error::new_canceled().with("pool is disabled")))
} else {
Ok(Async::NotReady)
// There's a new waiter, but there's no way it should be ready yet.
// We just need to register the waker.
self.poll_waiter(cx).map(|_| unreachable!())
}
}
}
@@ -717,37 +715,34 @@ impl Expiration {
}
#[cfg(feature = "runtime")]
struct IdleInterval<T> {
struct IdleTask<T> {
interval: Interval,
pool: WeakOpt<Mutex<PoolInner<T>>>,
// This allows the IdleInterval to be notified as soon as the entire
// This allows the IdleTask to be notified as soon as the entire
// Pool is fully dropped, and shutdown. This channel is never sent on,
// but Err(Canceled) will be received when the Pool is dropped.
pool_drop_notifier: oneshot::Receiver<crate::common::Never>,
}
#[cfg(feature = "runtime")]
impl<T: Poolable + 'static> Future for IdleInterval<T> {
type Item = ();
type Error = ();
impl<T: Poolable + 'static> Future for IdleTask<T> {
type Output = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
// Interval is a Stream
use futures::Stream;
use futures_core::Stream;
loop {
match self.pool_drop_notifier.poll() {
Ok(Async::Ready(n)) => match n {},
Ok(Async::NotReady) => (),
Err(_canceled) => {
match Pin::new(&mut self.pool_drop_notifier).poll(cx) {
Poll::Ready(Ok(n)) => match n {},
Poll::Pending => (),
Poll::Ready(Err(_canceled)) => {
trace!("pool closed, canceling idle interval");
return Ok(Async::Ready(()));
return Poll::Ready(());
}
}
try_ready!(self.interval.poll().map_err(|err| {
error!("idle interval timer error: {}", err);
}));
ready!(Pin::new(&mut self.interval).poll_next(cx));
if let Some(inner) = self.pool.upgrade() {
if let Ok(mut inner) = inner.lock() {
@@ -756,7 +751,7 @@ impl<T: Poolable + 'static> Future for IdleInterval<T> {
continue;
}
}
return Ok(Async::Ready(()));
return Poll::Ready(());
}
}
}