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

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

Browse Source
This commit is contained in:
Jonas Platte
2021-01-12 00:12:21 +01:00
committed by Sean McArthur
parent 9dff00425d
commit 43412a950f
14 changed files with 343 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.toml
View File

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

49
src/client/conn.rs
View File

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

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

@@ -11,7 +11,7 @@ use std::time::Duration;
use futures_util::future::Either;
use http::uri::{Scheme, Uri};
use pin_project::pin_project;
use pin_project_lite::pin_project;
use tokio::net::{TcpSocket, TcpStream};
use tokio::time::Sleep;
@@ -373,18 +373,19 @@ impl HttpInfo {
}
}
// Not publicly exported (so missing_docs doesn't trigger).
//
// 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
// (and thus we can change the type in the future).
#[must_use = "futures do nothing unless polled"]
#[pin_project]
#[allow(missing_debug_implementations)]
pub struct HttpConnecting<R> {
#[pin]
fut: BoxConnecting,
_marker: PhantomData<R>,
pin_project! {
// Not publicly exported (so missing_docs doesn't trigger).
//
// 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
// (and thus we can change the type in the future).
#[must_use = "futures do nothing unless polled"]
#[allow(missing_debug_implementations)]
pub struct HttpConnecting<R> {
#[pin]
fut: BoxConnecting,
_marker: PhantomData<R>,
}
}
type ConnectResult = Result<TcpStream, ConnectError>;

25
src/client/pool.rs
View File

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

19
src/common/drain.rs
View File

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

11
src/proto/h1/dispatch.rs
View File

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

21
src/proto/h2/mod.rs
View File

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

72
src/proto/h2/server.rs
View File

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

14
src/server/accept.rs
View File

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

236
src/server/conn.rs
View File

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

23
src/server/server.rs
View File

@@ -6,7 +6,7 @@ use std::net::{SocketAddr, TcpListener as StdTcpListener};
#[cfg(feature = "tcp")]
use std::time::Duration;
use pin_project::pin_project;
use pin_project_lite::pin_project;
use tokio::io::{AsyncRead, AsyncWrite};
use super::accept::Accept;
@@ -21,16 +21,17 @@ use super::shutdown::{Graceful, GracefulWatcher};
#[cfg(feature = "tcp")]
use super::tcp::AddrIncoming;
/// 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
/// handlers. It is built using the [`Builder`](Builder), and the future
/// completes when the server has been shutdown. It should be run by an
/// `Executor`.
#[pin_project]
pub struct Server<I, S, E = Exec> {
#[pin]
spawn_all: SpawnAll<I, S, E>,
pin_project! {
/// 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
/// handlers. It is built using the [`Builder`](Builder), and the future
/// completes when the server has been shutdown. It should be run by an
/// `Executor`.
pub struct Server<I, S, E = Exec> {
#[pin]
spawn_all: SpawnAll<I, S, E>,
}
}
/// A builder for a [`Server`](Server).

43
src/server/shutdown.rs
View File

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

15
src/server/tcp.rs
View File

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

48
src/service/oneshot.rs
View File

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