Revert "refactor(lib): Switch from pin-project to pin-project-lite"

This reverts commit 43412a950f.
This commit is contained in:
Sean McArthur
2021-02-05 15:37:21 -08:00
parent 48d4594930
commit 7390f026d7
14 changed files with 262 additions and 343 deletions

View File

@@ -34,7 +34,7 @@ httparse = "1.0"
h2 = { version = "0.3", optional = true } h2 = { version = "0.3", optional = true }
itoa = "0.4.1" itoa = "0.4.1"
tracing = { version = "0.1", default-features = false, features = ["std"] } tracing = { version = "0.1", default-features = false, features = ["std"] }
pin-project-lite = "0.2.4" pin-project = "1.0"
tower-service = "0.3" tower-service = "0.3"
tokio = { version = "1", features = ["sync"] } tokio = { version = "1", features = ["sync"] }
want = "0.3" want = "0.3"

View File

@@ -56,7 +56,7 @@ use std::time::Duration;
use bytes::Bytes; use bytes::Bytes;
use futures_util::future::{self, Either, FutureExt as _}; use futures_util::future::{self, Either, FutureExt as _};
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncRead, AsyncWrite};
use tower_service::Service; use tower_service::Service;
@@ -75,23 +75,15 @@ use crate::{Body, Request, Response};
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
type Http1Dispatcher<T, B, R> = proto::dispatch::Dispatcher<proto::dispatch::Client<B>, B, T, R>; type Http1Dispatcher<T, B, R> = proto::dispatch::Dispatcher<proto::dispatch::Client<B>, B, T, R>;
pin_project! { #[pin_project(project = ProtoClientProj)]
#[project = ProtoClientProj]
enum ProtoClient<T, B> enum ProtoClient<T, B>
where where
B: HttpBody, B: HttpBody,
{ {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
H1 { H1(#[pin] Http1Dispatcher<T, B, proto::h1::ClientTransaction>),
#[pin]
h1: Http1Dispatcher<T, B, proto::h1::ClientTransaction>,
},
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
H2 { H2(#[pin] proto::h2::ClientTask<B>, PhantomData<fn(T)>),
#[pin]
h2: proto::h2::ClientTask<B>, _phantom: PhantomData<fn(T)>,
},
}
} }
/// Returns a handshake future over some IO. /// Returns a handshake future over some IO.
@@ -408,7 +400,7 @@ where
pub fn into_parts(self) -> Parts<T> { pub fn into_parts(self) -> Parts<T> {
match self.inner.expect("already upgraded") { match self.inner.expect("already upgraded") {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
ProtoClient::H1 { h1 } => { ProtoClient::H1(h1) => {
let (io, read_buf, _) = h1.into_inner(); let (io, read_buf, _) = h1.into_inner();
Parts { Parts {
io, io,
@@ -417,7 +409,7 @@ where
} }
} }
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
ProtoClient::H2 { .. } => { ProtoClient::H2(..) => {
panic!("http2 cannot into_inner"); panic!("http2 cannot into_inner");
} }
} }
@@ -437,9 +429,9 @@ where
pub fn poll_without_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> { pub fn poll_without_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
match *self.inner.as_mut().expect("already upgraded") { match *self.inner.as_mut().expect("already upgraded") {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
ProtoClient::H1 { ref mut h1 } => h1.poll_without_shutdown(cx), ProtoClient::H1(ref mut h1) => h1.poll_without_shutdown(cx),
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
ProtoClient::H2 { ref mut h2, .. } => Pin::new(h2).poll(cx).map_ok(|_| ()), ProtoClient::H2(ref mut h2, _) => Pin::new(h2).poll(cx).map_ok(|_| ()),
} }
} }
@@ -468,7 +460,7 @@ where
proto::Dispatched::Shutdown => Poll::Ready(Ok(())), proto::Dispatched::Shutdown => Poll::Ready(Ok(())),
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
proto::Dispatched::Upgrade(pending) => match self.inner.take() { proto::Dispatched::Upgrade(pending) => match self.inner.take() {
Some(ProtoClient::H1 { h1 }) => { Some(ProtoClient::H1(h1)) => {
let (io, buf, _) = h1.into_inner(); let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(io, buf)); pending.fulfill(Upgraded::new(io, buf));
Poll::Ready(Ok(())) Poll::Ready(Ok(()))
@@ -715,17 +707,14 @@ impl Builder {
} }
let cd = proto::h1::dispatch::Client::new(rx); let cd = proto::h1::dispatch::Client::new(rx);
let dispatch = proto::h1::Dispatcher::new(cd, conn); let dispatch = proto::h1::Dispatcher::new(cd, conn);
ProtoClient::H1 { h1: dispatch } ProtoClient::H1(dispatch)
} }
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
Proto::Http2 => { Proto::Http2 => {
let h2 = let h2 =
proto::h2::client::handshake(io, rx, &opts.h2_builder, opts.exec.clone()) proto::h2::client::handshake(io, rx, &opts.h2_builder, opts.exec.clone())
.await?; .await?;
ProtoClient::H2 { ProtoClient::H2(h2, PhantomData)
h2,
_phantom: PhantomData,
}
} }
}; };
@@ -779,9 +768,9 @@ where
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match self.project() { match self.project() {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
ProtoClientProj::H1 { h1 } => h1.poll(cx), ProtoClientProj::H1(c) => c.poll(cx),
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
ProtoClientProj::H2 { h2, .. } => h2.poll(cx), ProtoClientProj::H2(c, _) => c.poll(cx),
} }
} }
} }

View File

@@ -11,7 +11,7 @@ use std::time::Duration;
use futures_util::future::Either; use futures_util::future::Either;
use http::uri::{Scheme, Uri}; use http::uri::{Scheme, Uri};
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::net::{TcpSocket, TcpStream}; use tokio::net::{TcpSocket, TcpStream};
use tokio::time::Sleep; use tokio::time::Sleep;
@@ -373,20 +373,19 @@ impl HttpInfo {
} }
} }
pin_project! {
// Not publicly exported (so missing_docs doesn't trigger). // Not publicly exported (so missing_docs doesn't trigger).
// //
// We return this `Future` instead of the `Pin<Box<dyn Future>>` directly // We return this `Future` instead of the `Pin<Box<dyn Future>>` directly
// so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot // so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot
// (and thus we can change the type in the future). // (and thus we can change the type in the future).
#[must_use = "futures do nothing unless polled"] #[must_use = "futures do nothing unless polled"]
#[pin_project]
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
pub struct HttpConnecting<R> { pub struct HttpConnecting<R> {
#[pin] #[pin]
fut: BoxConnecting, fut: BoxConnecting,
_marker: PhantomData<R>, _marker: PhantomData<R>,
} }
}
type ConnectResult = Result<TcpStream, ConnectError>; type ConnectResult = Result<TcpStream, ConnectError>;
type BoxConnecting = Pin<Box<dyn Future<Output = ConnectResult> + Send>>; type BoxConnecting = Pin<Box<dyn Future<Output = ConnectResult> + Send>>;

View File

@@ -11,7 +11,7 @@ use futures_channel::oneshot;
use tokio::time::{Duration, Instant, Interval}; use tokio::time::{Duration, Instant, Interval};
use super::client::Ver; use super::client::Ver;
use crate::common::{exec::Exec, task, Future, Pin, Poll, Unpin}; use crate::common::{task, exec::Exec, Future, Pin, Poll, Unpin};
// FIXME: allow() required due to `impl Trait` leaking types to this lint // FIXME: allow() required due to `impl Trait` leaking types to this lint
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
@@ -714,7 +714,7 @@ impl Expiration {
} }
#[cfg(feature = "runtime")] #[cfg(feature = "runtime")]
pin_project_lite::pin_project! { #[pin_project::pin_project]
struct IdleTask<T> { struct IdleTask<T> {
#[pin] #[pin]
interval: Interval, interval: Interval,
@@ -725,7 +725,6 @@ pin_project_lite::pin_project! {
#[pin] #[pin]
pool_drop_notifier: oneshot::Receiver<crate::common::Never>, pool_drop_notifier: oneshot::Receiver<crate::common::Never>,
} }
}
#[cfg(feature = "runtime")] #[cfg(feature = "runtime")]
impl<T: Poolable + 'static> Future for IdleTask<T> { impl<T: Poolable + 'static> Future for IdleTask<T> {
@@ -777,7 +776,7 @@ mod tests {
use std::time::Duration; use std::time::Duration;
use super::{Connecting, Key, Pool, Poolable, Reservation, WeakOpt}; use super::{Connecting, Key, Pool, Poolable, Reservation, WeakOpt};
use crate::common::{exec::Exec, task, Future, Pin}; use crate::common::{task, exec::Exec, Future, Pin};
/// Test unique reservations. /// Test unique reservations.
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]

View File

@@ -1,6 +1,6 @@
use std::mem; use std::mem;
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::sync::watch; use tokio::sync::watch;
use super::{task, Future, Pin, Poll}; use super::{task, Future, Pin, Poll};
@@ -21,8 +21,8 @@ pub(crate) struct Watch {
rx: watch::Receiver<()>, rx: watch::Receiver<()>,
} }
pin_project! {
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
#[pin_project]
pub struct Watching<F, FN> { pub struct Watching<F, FN> {
#[pin] #[pin]
future: F, future: F,
@@ -30,7 +30,6 @@ pin_project! {
watch: Pin<Box<dyn Future<Output = ()> + Send + Sync>>, watch: Pin<Box<dyn Future<Output = ()> + Send + Sync>>,
_rx: watch::Receiver<()>, _rx: watch::Receiver<()>,
} }
}
enum State<F> { enum State<F> {
Watch(F), Watch(F),

View File

@@ -44,14 +44,11 @@ cfg_server! {
} }
cfg_client! { cfg_client! {
pin_project_lite::pin_project! {
pub(crate) struct Client<B> { pub(crate) struct Client<B> {
callback: Option<crate::client::dispatch::Callback<Request<B>, http::Response<Body>>>, callback: Option<crate::client::dispatch::Callback<Request<B>, http::Response<Body>>>,
#[pin]
rx: ClientRx<B>, rx: ClientRx<B>,
rx_closed: bool, rx_closed: bool,
} }
}
type ClientRx<B> = crate::client::dispatch::Receiver<Request<B>, http::Response<Body>>; type ClientRx<B> = crate::client::dispatch::Receiver<Request<B>, http::Response<Body>>;
} }

View File

@@ -5,7 +5,7 @@ use http::header::{
TRANSFER_ENCODING, UPGRADE, TRANSFER_ENCODING, UPGRADE,
}; };
use http::HeaderMap; use http::HeaderMap;
use pin_project_lite::pin_project; use pin_project::pin_project;
use std::error::Error as StdError; use std::error::Error as StdError;
use std::io::IoSlice; use std::io::IoSlice;
@@ -94,7 +94,7 @@ fn decode_content_length(headers: &HeaderMap) -> DecodedLength {
// body adapters used by both Client and Server // body adapters used by both Client and Server
pin_project! { #[pin_project]
struct PipeToSendStream<S> struct PipeToSendStream<S>
where where
S: HttpBody, S: HttpBody,
@@ -104,7 +104,6 @@ pin_project! {
#[pin] #[pin]
stream: S, stream: S,
} }
}
impl<S> PipeToSendStream<S> impl<S> PipeToSendStream<S>
where where

View File

@@ -5,7 +5,7 @@ use std::time::Duration;
use h2::server::{Connection, Handshake, SendResponse}; use h2::server::{Connection, Handshake, SendResponse};
use h2::Reason; use h2::Reason;
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncRead, AsyncWrite};
use super::{decode_content_length, ping, PipeToSendStream, SendBuf}; use super::{decode_content_length, ping, PipeToSendStream, SendBuf};
@@ -57,7 +57,7 @@ impl Default for Config {
} }
} }
pin_project! { #[pin_project]
pub(crate) struct Server<T, S, B, E> pub(crate) struct Server<T, S, B, E>
where where
S: HttpService<Body>, S: HttpService<Body>,
@@ -67,7 +67,6 @@ pin_project! {
service: S, service: S,
state: State<T, B>, state: State<T, B>,
} }
}
enum State<T, B> enum State<T, B>
where where
@@ -316,8 +315,8 @@ where
} }
} }
pin_project! {
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
#[pin_project]
pub struct H2Stream<F, B> pub struct H2Stream<F, B>
where where
B: HttpBody, B: HttpBody,
@@ -326,23 +325,14 @@ pin_project! {
#[pin] #[pin]
state: H2StreamState<F, B>, state: H2StreamState<F, B>,
} }
}
pin_project! { #[pin_project(project = H2StreamStateProj)]
#[project = H2StreamStateProj]
enum H2StreamState<F, B> enum H2StreamState<F, B>
where where
B: HttpBody, B: HttpBody,
{ {
Service { Service(#[pin] F),
#[pin] Body(#[pin] PipeToSendStream<B>),
fut: F,
},
Body {
#[pin]
pipe: PipeToSendStream<B>,
},
}
} }
impl<F, B> H2Stream<F, B> impl<F, B> H2Stream<F, B>
@@ -352,7 +342,7 @@ where
fn new(fut: F, respond: SendResponse<SendBuf<B::Data>>) -> H2Stream<F, B> { fn new(fut: F, respond: SendResponse<SendBuf<B::Data>>) -> H2Stream<F, B> {
H2Stream { H2Stream {
reply: respond, reply: respond,
state: H2StreamState::Service { fut }, state: H2StreamState::Service(fut),
} }
} }
} }
@@ -381,7 +371,7 @@ where
let mut me = self.project(); let mut me = self.project();
loop { loop {
let next = match me.state.as_mut().project() { let next = match me.state.as_mut().project() {
H2StreamStateProj::Service { fut: h } => { H2StreamStateProj::Service(h) => {
let res = match h.poll(cx) { let res = match h.poll(cx) {
Poll::Ready(Ok(r)) => r, Poll::Ready(Ok(r)) => r,
Poll::Pending => { Poll::Pending => {
@@ -419,15 +409,13 @@ where
if !body.is_end_stream() { if !body.is_end_stream() {
let body_tx = reply!(me, res, false); let body_tx = reply!(me, res, false);
H2StreamState::Body { H2StreamState::Body(PipeToSendStream::new(body, body_tx))
pipe: PipeToSendStream::new(body, body_tx),
}
} else { } else {
reply!(me, res, true); reply!(me, res, true);
return Poll::Ready(Ok(())); return Poll::Ready(Ok(()));
} }
} }
H2StreamStateProj::Body { pipe } => { H2StreamStateProj::Body(pipe) => {
return pipe.poll(cx); return pipe.poll(cx);
} }
}; };

View File

@@ -9,7 +9,7 @@
#[cfg(feature = "stream")] #[cfg(feature = "stream")]
use futures_core::Stream; use futures_core::Stream;
#[cfg(feature = "stream")] #[cfg(feature = "stream")]
use pin_project_lite::pin_project; use pin_project::pin_project;
use crate::common::{ use crate::common::{
task::{self, Poll}, task::{self, Poll},
@@ -86,12 +86,8 @@ pub fn from_stream<S, IO, E>(stream: S) -> impl Accept<Conn = IO, Error = E>
where where
S: Stream<Item = Result<IO, E>>, S: Stream<Item = Result<IO, E>>,
{ {
pin_project! { #[pin_project]
struct FromStream<S> { struct FromStream<S>(#[pin] S);
#[pin]
stream: S,
}
}
impl<S, IO, E> Accept for FromStream<S> impl<S, IO, E> Accept for FromStream<S>
where where
@@ -103,9 +99,9 @@ where
self: Pin<&mut Self>, self: Pin<&mut Self>,
cx: &mut task::Context<'_>, cx: &mut task::Context<'_>,
) -> Poll<Option<Result<Self::Conn, Self::Error>>> { ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
self.project().stream.poll_next(cx) self.project().0.poll_next(cx)
} }
} }
FromStream { stream } FromStream(stream)
} }

View File

@@ -45,6 +45,7 @@
use std::error::Error as StdError; use std::error::Error as StdError;
use std::fmt; use std::fmt;
#[cfg(feature = "http1")]
use std::marker::PhantomData; use std::marker::PhantomData;
#[cfg(feature = "tcp")] #[cfg(feature = "tcp")]
use std::net::SocketAddr; use std::net::SocketAddr;
@@ -52,7 +53,7 @@ use std::net::SocketAddr;
use std::time::Duration; use std::time::Duration;
use bytes::Bytes; use bytes::Bytes;
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncRead, AsyncWrite};
use super::accept::Accept; use super::accept::Accept;
@@ -108,11 +109,11 @@ enum ConnectionMode {
Fallback, Fallback,
} }
pin_project! {
/// A stream mapping incoming IOs to new services. /// A stream mapping incoming IOs to new services.
/// ///
/// Yields `Connecting`s that are futures that should be put on a reactor. /// Yields `Connecting`s that are futures that should be put on a reactor.
#[must_use = "streams do nothing unless polled"] #[must_use = "streams do nothing unless polled"]
#[pin_project]
#[derive(Debug)] #[derive(Debug)]
pub(super) struct Serve<I, S, E = Exec> { pub(super) struct Serve<I, S, E = Exec> {
#[pin] #[pin]
@@ -120,14 +121,13 @@ pin_project! {
make_service: S, make_service: S,
protocol: Http<E>, protocol: Http<E>,
} }
}
pin_project! {
/// A future building a new `Service` to a `Connection`. /// A future building a new `Service` to a `Connection`.
/// ///
/// Wraps the future returned from `MakeService` into one that returns /// Wraps the future returned from `MakeService` into one that returns
/// a `Connection`. /// a `Connection`.
#[must_use = "futures do nothing unless polled"] #[must_use = "futures do nothing unless polled"]
#[pin_project]
#[derive(Debug)] #[derive(Debug)]
pub struct Connecting<I, F, E = Exec> { pub struct Connecting<I, F, E = Exec> {
#[pin] #[pin]
@@ -135,10 +135,9 @@ pin_project! {
io: Option<I>, io: Option<I>,
protocol: Http<E>, protocol: Http<E>,
} }
}
pin_project! {
#[must_use = "futures do nothing unless polled"] #[must_use = "futures do nothing unless polled"]
#[pin_project]
#[derive(Debug)] #[derive(Debug)]
pub(super) struct SpawnAll<I, S, E> { pub(super) struct SpawnAll<I, S, E> {
// TODO: re-add `pub(super)` once rustdoc can handle this. // TODO: re-add `pub(super)` once rustdoc can handle this.
@@ -147,46 +146,40 @@ pin_project! {
#[pin] #[pin]
pub(super) serve: Serve<I, S, E>, pub(super) serve: Serve<I, S, E>,
} }
}
pin_project! {
/// A future binding a connection with a Service. /// A future binding a connection with a Service.
/// ///
/// Polling this future will drive HTTP forward. /// Polling this future will drive HTTP forward.
#[must_use = "futures do nothing unless polled"] #[must_use = "futures do nothing unless polled"]
#[pin_project]
pub struct Connection<T, S, E = Exec> pub struct Connection<T, S, E = Exec>
where where
S: HttpService<Body>, S: HttpService<Body>,
{ {
pub(super) conn: Option<ProtoServer<T, S::ResBody, S, E>>, pub(super) conn: Option<ProtoServer<T, S::ResBody, S, E>>,
#[cfg(all(feature = "http1", feature = "http2"))]
fallback: Fallback<E>, fallback: Fallback<E>,
} }
}
pin_project! { #[pin_project(project = ProtoServerProj)]
#[project = ProtoServerProj]
pub(super) enum ProtoServer<T, B, S, E = Exec> pub(super) enum ProtoServer<T, B, S, E = Exec>
where where
S: HttpService<Body>, S: HttpService<Body>,
B: HttpBody, B: HttpBody,
{ {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
H1 { H1(
#[pin] #[pin]
h1: proto::h1::Dispatcher< proto::h1::Dispatcher<
proto::h1::dispatch::Server<S, Body>, proto::h1::dispatch::Server<S, Body>,
B, B,
T, T,
proto::ServerTransaction, proto::ServerTransaction,
>, >,
_phantom: PhantomData<E>, PhantomData<E>,
}, ),
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
H2 { H2(#[pin] proto::h2::Server<Rewind<T>, S, B, E>),
#[pin]
h2: proto::h2::Server<Rewind<T>, S, B, E>,
},
}
} }
#[cfg(all(feature = "http1", feature = "http2"))] #[cfg(all(feature = "http1", feature = "http2"))]
@@ -196,10 +189,6 @@ enum Fallback<E> {
Http1Only, Http1Only,
} }
#[cfg(not(all(feature = "http1", feature = "http2")))]
#[derive(Clone, Debug)]
struct Fallback<E>(PhantomData<E>);
#[cfg(all(feature = "http1", feature = "http2"))] #[cfg(all(feature = "http1", feature = "http2"))]
impl<E> Fallback<E> { impl<E> Fallback<E> {
fn to_h2(&self) -> bool { fn to_h2(&self) -> bool {
@@ -530,10 +519,7 @@ impl<E> Http<E> {
conn.set_max_buf_size(max); conn.set_max_buf_size(max);
} }
let sd = proto::h1::dispatch::Server::new(service); let sd = proto::h1::dispatch::Server::new(service);
ProtoServer::H1 { ProtoServer::H1(proto::h1::Dispatcher::new(sd, conn), PhantomData)
h1: proto::h1::Dispatcher::new(sd, conn),
_phantom: PhantomData,
}
}}; }};
} }
@@ -549,7 +535,7 @@ impl<E> Http<E> {
let rewind_io = Rewind::new(io); let rewind_io = Rewind::new(io);
let h2 = let h2 =
proto::h2::Server::new(rewind_io, service, &self.h2_builder, self.exec.clone()); proto::h2::Server::new(rewind_io, service, &self.h2_builder, self.exec.clone());
ProtoServer::H2 { h2 } ProtoServer::H2(h2)
} }
}; };
@@ -604,14 +590,14 @@ where
/// This should only be called while the `Connection` future is still /// This should only be called while the `Connection` future is still
/// pending. If called after `Connection::poll` has resolved, this does /// pending. If called after `Connection::poll` has resolved, this does
/// nothing. /// nothing.
pub fn graceful_shutdown(mut self: Pin<&mut Self>) { pub fn graceful_shutdown(self: Pin<&mut Self>) {
match self.conn { match self.project().conn {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
Some(ProtoServer::H1 { ref mut h1, .. }) => { Some(ProtoServer::H1(ref mut h1, _)) => {
h1.disable_keep_alive(); h1.disable_keep_alive();
} }
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
Some(ProtoServer::H2 { ref mut h2 }) => { Some(ProtoServer::H2(ref mut h2)) => {
h2.graceful_shutdown(); h2.graceful_shutdown();
} }
None => (), None => (),
@@ -638,7 +624,7 @@ where
pub fn try_into_parts(self) -> Option<Parts<I, S>> { pub fn try_into_parts(self) -> Option<Parts<I, S>> {
match self.conn.unwrap() { match self.conn.unwrap() {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
ProtoServer::H1 { h1, .. } => { ProtoServer::H1(h1, _) => {
let (io, read_buf, dispatch) = h1.into_inner(); let (io, read_buf, dispatch) = h1.into_inner();
Some(Parts { Some(Parts {
io, io,
@@ -648,7 +634,7 @@ where
}) })
} }
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
ProtoServer::H2 { .. } => None, ProtoServer::H2(_h2) => None,
} }
} }
@@ -672,7 +658,7 @@ where
loop { loop {
match *self.conn.as_mut().unwrap() { match *self.conn.as_mut().unwrap() {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
ProtoServer::H1 { ref mut h1, .. } => match ready!(h1.poll_without_shutdown(cx)) { ProtoServer::H1(ref mut h1, _) => match ready!(h1.poll_without_shutdown(cx)) {
Ok(()) => return Poll::Ready(Ok(())), Ok(()) => return Poll::Ready(Ok(())),
Err(e) => { Err(e) => {
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
@@ -688,7 +674,7 @@ where
} }
}, },
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
ProtoServer::H2 { ref mut h2 } => return Pin::new(h2).poll(cx).map_ok(|_| ()), ProtoServer::H2(ref mut h2) => return Pin::new(h2).poll(cx).map_ok(|_| ()),
}; };
} }
} }
@@ -714,8 +700,8 @@ where
let conn = self.conn.take(); let conn = self.conn.take();
let (io, read_buf, dispatch) = match conn.unwrap() { let (io, read_buf, dispatch) = match conn.unwrap() {
ProtoServer::H1 { h1, .. } => h1.into_inner(), ProtoServer::H1(h1, _) => h1.into_inner(),
ProtoServer::H2 { .. } => { ProtoServer::H2(_h2) => {
panic!("h2 cannot into_inner"); panic!("h2 cannot into_inner");
} }
}; };
@@ -728,7 +714,7 @@ where
let h2 = proto::h2::Server::new(rewind_io, dispatch.into_service(), builder, exec.clone()); let h2 = proto::h2::Server::new(rewind_io, dispatch.into_service(), builder, exec.clone());
debug_assert!(self.conn.is_none()); debug_assert!(self.conn.is_none());
self.conn = Some(ProtoServer::H2 { h2 }); self.conn = Some(ProtoServer::H2(h2));
} }
/// Enable this connection to support higher-level HTTP upgrades. /// Enable this connection to support higher-level HTTP upgrades.
@@ -962,9 +948,9 @@ where
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match self.project() { match self.project() {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
ProtoServerProj::H1 { h1, .. } => h1.poll(cx), ProtoServerProj::H1(s, _) => s.poll(cx),
#[cfg(feature = "http2")] #[cfg(feature = "http2")]
ProtoServerProj::H2 { h2 } => h2.poll(cx), ProtoServerProj::H2(s) => s.poll(cx),
} }
} }
} }
@@ -978,7 +964,7 @@ pub(crate) mod spawn_all {
use crate::common::exec::ConnStreamExec; use crate::common::exec::ConnStreamExec;
use crate::common::{task, Future, Pin, Poll, Unpin}; use crate::common::{task, Future, Pin, Poll, Unpin};
use crate::service::HttpService; use crate::service::HttpService;
use pin_project_lite::pin_project; use pin_project::pin_project;
// Used by `SpawnAll` to optionally watch a `Connection` future. // Used by `SpawnAll` to optionally watch a `Connection` future.
// //
@@ -1023,36 +1009,23 @@ pub(crate) mod spawn_all {
// Users cannot import this type, nor the associated `NewSvcExec`. Instead, // Users cannot import this type, nor the associated `NewSvcExec`. Instead,
// a blanket implementation for `Executor<impl Future>` is sufficient. // a blanket implementation for `Executor<impl Future>` is sufficient.
pin_project! { #[pin_project]
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
pub struct NewSvcTask<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> { pub struct NewSvcTask<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> {
#[pin] #[pin]
state: State<I, N, S, E, W>, state: State<I, N, S, E, W>,
} }
}
pin_project! { #[pin_project(project = StateProj)]
#[project = StateProj]
pub(super) enum State<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> { pub(super) enum State<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> {
Connecting { Connecting(#[pin] Connecting<I, N, E>, W),
#[pin] Connected(#[pin] W::Future),
connecting: Connecting<I, N, E>,
watcher: W,
},
Connected {
#[pin]
future: W::Future,
},
}
} }
impl<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> NewSvcTask<I, N, S, E, W> { impl<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> NewSvcTask<I, N, S, E, W> {
pub(super) fn new(connecting: Connecting<I, N, E>, watcher: W) -> Self { pub(super) fn new(connecting: Connecting<I, N, E>, watcher: W) -> Self {
NewSvcTask { NewSvcTask {
state: State::Connecting { state: State::Connecting(connecting, watcher),
connecting,
watcher,
},
} }
} }
} }
@@ -1079,10 +1052,7 @@ pub(crate) mod spawn_all {
loop { loop {
let next = { let next = {
match me.state.as_mut().project() { match me.state.as_mut().project() {
StateProj::Connecting { StateProj::Connecting(connecting, watcher) => {
connecting,
watcher,
} => {
let res = ready!(connecting.poll(cx)); let res = ready!(connecting.poll(cx));
let conn = match res { let conn = match res {
Ok(conn) => conn, Ok(conn) => conn,
@@ -1092,10 +1062,10 @@ pub(crate) mod spawn_all {
return Poll::Ready(()); return Poll::Ready(());
} }
}; };
let future = watcher.watch(conn.with_upgrades()); let connected = watcher.watch(conn.with_upgrades());
State::Connected { future } State::Connected(connected)
} }
StateProj::Connected { future } => { StateProj::Connected(future) => {
return future.poll(cx).map(|res| { return future.poll(cx).map(|res| {
if let Err(err) = res { if let Err(err) = res {
debug!("connection error: {}", err); debug!("connection error: {}", err);
@@ -1163,7 +1133,7 @@ mod upgrades {
#[cfg(feature = "http1")] #[cfg(feature = "http1")]
Ok(proto::Dispatched::Upgrade(pending)) => { Ok(proto::Dispatched::Upgrade(pending)) => {
match self.inner.conn.take() { match self.inner.conn.take() {
Some(ProtoServer::H1 { h1, .. }) => { Some(ProtoServer::H1(h1, _)) => {
let (io, buf, _) = h1.into_inner(); let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(io, buf)); pending.fulfill(Upgraded::new(io, buf));
return Poll::Ready(Ok(())); return Poll::Ready(Ok(()));

View File

@@ -6,7 +6,7 @@ use std::net::{SocketAddr, TcpListener as StdTcpListener};
#[cfg(feature = "tcp")] #[cfg(feature = "tcp")]
use std::time::Duration; use std::time::Duration;
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncRead, AsyncWrite};
use super::accept::Accept; use super::accept::Accept;
@@ -21,18 +21,17 @@ use super::shutdown::{Graceful, GracefulWatcher};
#[cfg(feature = "tcp")] #[cfg(feature = "tcp")]
use super::tcp::AddrIncoming; use super::tcp::AddrIncoming;
pin_project! {
/// A listening HTTP server that accepts connections in both HTTP1 and HTTP2 by default. /// A listening HTTP server that accepts connections in both HTTP1 and HTTP2 by default.
/// ///
/// `Server` is a `Future` mapping a bound listener with a set of service /// `Server` is a `Future` mapping a bound listener with a set of service
/// handlers. It is built using the [`Builder`](Builder), and the future /// handlers. It is built using the [`Builder`](Builder), and the future
/// completes when the server has been shutdown. It should be run by an /// completes when the server has been shutdown. It should be run by an
/// `Executor`. /// `Executor`.
#[pin_project]
pub struct Server<I, S, E = Exec> { pub struct Server<I, S, E = Exec> {
#[pin] #[pin]
spawn_all: SpawnAll<I, S, E>, spawn_all: SpawnAll<I, S, E>,
} }
}
/// A builder for a [`Server`](Server). /// A builder for a [`Server`](Server).
#[derive(Debug)] #[derive(Debug)]

View File

@@ -1,26 +1,24 @@
use std::error::Error as StdError; use std::error::Error as StdError;
use pin_project_lite::pin_project; use pin_project::pin_project;
use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncRead, AsyncWrite};
use super::accept::Accept;
use super::conn::{SpawnAll, UpgradeableConnection, Watcher}; use super::conn::{SpawnAll, UpgradeableConnection, Watcher};
use super::accept::Accept;
use crate::body::{Body, HttpBody}; use crate::body::{Body, HttpBody};
use crate::common::drain::{self, Draining, Signal, Watch, Watching}; use crate::common::drain::{self, Draining, Signal, Watch, Watching};
use crate::common::exec::{ConnStreamExec, NewSvcExec}; use crate::common::exec::{ConnStreamExec, NewSvcExec};
use crate::common::{task, Future, Pin, Poll, Unpin}; use crate::common::{task, Future, Pin, Poll, Unpin};
use crate::service::{HttpService, MakeServiceRef}; use crate::service::{HttpService, MakeServiceRef};
pin_project! {
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
#[pin_project]
pub struct Graceful<I, S, F, E> { pub struct Graceful<I, S, F, E> {
#[pin] #[pin]
state: State<I, S, F, E>, state: State<I, S, F, E>,
} }
}
pin_project! { #[pin_project(project = StateProj)]
#[project = StateProj]
pub(super) enum State<I, S, F, E> { pub(super) enum State<I, S, F, E> {
Running { Running {
drain: Option<(Signal, Watch)>, drain: Option<(Signal, Watch)>,
@@ -29,8 +27,7 @@ pin_project! {
#[pin] #[pin]
signal: F, signal: F,
}, },
Draining { draining: Draining }, Draining(Draining),
}
} }
impl<I, S, F, E> Graceful<I, S, F, E> { impl<I, S, F, E> Graceful<I, S, F, E> {
@@ -74,16 +71,14 @@ where
Poll::Ready(()) => { Poll::Ready(()) => {
debug!("signal received, starting graceful shutdown"); debug!("signal received, starting graceful shutdown");
let sig = drain.take().expect("drain channel").0; let sig = drain.take().expect("drain channel").0;
State::Draining { State::Draining(sig.drain())
draining: sig.drain(),
}
} }
Poll::Pending => { Poll::Pending => {
let watch = drain.as_ref().expect("drain channel").1.clone(); let watch = drain.as_ref().expect("drain channel").1.clone();
return spawn_all.poll_watch(cx, &GracefulWatcher(watch)); return spawn_all.poll_watch(cx, &GracefulWatcher(watch));
} }
}, },
StateProj::Draining { ref mut draining } => { StateProj::Draining(ref mut draining) => {
return Pin::new(draining).poll(cx).map(Ok); return Pin::new(draining).poll(cx).map(Ok);
} }
} }

View File

@@ -229,15 +229,14 @@ mod addr_stream {
use crate::common::{task, Pin, Poll}; use crate::common::{task, Pin, Poll};
pin_project_lite::pin_project! {
/// A transport returned yieled by `AddrIncoming`. /// A transport returned yieled by `AddrIncoming`.
#[pin_project::pin_project]
#[derive(Debug)] #[derive(Debug)]
pub struct AddrStream { pub struct AddrStream {
#[pin] #[pin]
inner: TcpStream, inner: TcpStream,
pub(super) remote_addr: SocketAddr, pub(super) remote_addr: SocketAddr,
} }
}
impl AddrStream { impl AddrStream {
pub(super) fn new(tcp: TcpStream, addr: SocketAddr) -> AddrStream { pub(super) fn new(tcp: TcpStream, addr: SocketAddr) -> AddrStream {

View File

@@ -1,6 +1,6 @@
// TODO: Eventually to be replaced with tower_util::Oneshot. // TODO: Eventually to be replaced with tower_util::Oneshot.
use pin_project_lite::pin_project; use pin_project::pin_project;
use tower_service::Service; use tower_service::Service;
use crate::common::{task, Future, Pin, Poll}; use crate::common::{task, Future, Pin, Poll};
@@ -10,36 +10,26 @@ where
S: Service<Req>, S: Service<Req>,
{ {
Oneshot { Oneshot {
state: State::NotReady { svc, req }, state: State::NotReady(svc, req),
} }
} }
pin_project! {
// A `Future` consuming a `Service` and request, waiting until the `Service` // A `Future` consuming a `Service` and request, waiting until the `Service`
// is ready, and then calling `Service::call` with the request, and // is ready, and then calling `Service::call` with the request, and
// waiting for that `Future`. // waiting for that `Future`.
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
#[pin_project]
pub struct Oneshot<S: Service<Req>, Req> { pub struct Oneshot<S: Service<Req>, Req> {
#[pin] #[pin]
state: State<S, Req>, state: State<S, Req>,
} }
}
pin_project! { #[pin_project(project = StateProj, project_replace = StateProjOwn)]
#[project = StateProj]
#[project_replace = StateProjOwn]
enum State<S: Service<Req>, Req> { enum State<S: Service<Req>, Req> {
NotReady { NotReady(S, Req),
svc: S, Called(#[pin] S::Future),
req: Req,
},
Called {
#[pin]
fut: S::Future,
},
Tmp, Tmp,
} }
}
impl<S, Req> Future for Oneshot<S, Req> impl<S, Req> Future for Oneshot<S, Req>
where where
@@ -52,19 +42,19 @@ where
loop { loop {
match me.state.as_mut().project() { match me.state.as_mut().project() {
StateProj::NotReady { ref mut svc, .. } => { StateProj::NotReady(ref mut svc, _) => {
ready!(svc.poll_ready(cx))?; ready!(svc.poll_ready(cx))?;
// fallthrough out of the match's borrow // fallthrough out of the match's borrow
} }
StateProj::Called { fut } => { StateProj::Called(fut) => {
return fut.poll(cx); return fut.poll(cx);
} }
StateProj::Tmp => unreachable!(), StateProj::Tmp => unreachable!(),
} }
match me.state.as_mut().project_replace(State::Tmp) { match me.state.as_mut().project_replace(State::Tmp) {
StateProjOwn::NotReady { mut svc, req } => { StateProjOwn::NotReady(mut svc, req) => {
me.state.set(State::Called { fut: svc.call(req) }); me.state.set(State::Called(svc.call(req)));
} }
_ => unreachable!(), _ => unreachable!(),
} }