use std::error::Error as StdError;
use std::fmt;
use std::mem;
#[cfg(feature = "tcp")] use std::net::SocketAddr;
#[cfg(feature = "tcp")] use std::time::Duration;
use bytes::Bytes;
use futures_core::Stream;
use tokio_io::{AsyncRead, AsyncWrite};
use pin_project::{pin_project, project};
#[cfg(feature = "tcp")] use tokio_net::driver::Handle;
use crate::body::{Body, Payload};
use crate::common::exec::{Exec, H2Exec, NewSvcExec};
use crate::common::io::Rewind;
use crate::common::{Future, Pin, Poll, Unpin, task};
use crate::error::{Kind, Parse};
use crate::proto;
use crate::service::{MakeServiceRef, Service};
use crate::upgrade::Upgraded;
use super::Accept;
pub(super) use self::spawn_all::NoopWatcher;
use self::spawn_all::NewSvcTask;
pub(super) use self::spawn_all::Watcher;
pub(super) use self::upgrades::UpgradeableConnection;
#[cfg(feature = "tcp")] pub use super::tcp::{AddrIncoming, AddrStream};
#[derive(Clone, Debug)]
pub struct Http<E = Exec> {
exec: E,
h1_half_close: bool,
h1_writev: bool,
h2_builder: h2::server::Builder,
mode: ConnectionMode,
keep_alive: bool,
max_buf_size: Option<usize>,
pipeline_flush: bool,
}
#[derive(Clone, Debug, PartialEq)]
enum ConnectionMode {
H1Only,
H2Only,
Fallback,
}
#[must_use = "streams do nothing unless polled"]
#[pin_project]
#[derive(Debug)]
pub struct Serve<I, S, E = Exec> {
#[pin]
incoming: I,
make_service: S,
protocol: Http<E>,
}
#[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> {
#[pin]
pub(super) serve: Serve<I, S, E>,
}
#[must_use = "futures do nothing unless polled"]
#[pin_project]
pub struct Connection<T, S, E = Exec>
where
S: Service<Body>,
{
pub(super) conn: Option<Either<
proto::h1::Dispatcher<
proto::h1::dispatch::Server<S, Body>,
S::ResBody,
T,
proto::ServerTransaction,
>,
proto::h2::Server<
Rewind<T>,
S,
S::ResBody,
E,
>,
>>,
fallback: Fallback<E>,
}
#[pin_project]
pub(super) enum Either<A, B> {
A(#[pin] A),
B(#[pin] B),
}
#[derive(Clone, Debug)]
enum Fallback<E> {
ToHttp2(h2::server::Builder, E),
Http1Only,
}
impl<E> Fallback<E> {
fn to_h2(&self) -> bool {
match *self {
Fallback::ToHttp2(..) => true,
Fallback::Http1Only => false,
}
}
}
impl<E> Unpin for Fallback<E> {}
#[derive(Debug)]
pub struct Parts<T, S> {
pub io: T,
pub read_buf: Bytes,
pub service: S,
_inner: (),
}
impl Http {
pub fn new() -> Http {
Http {
exec: Exec::Default,
h1_half_close: true,
h1_writev: true,
h2_builder: h2::server::Builder::default(),
mode: ConnectionMode::Fallback,
keep_alive: true,
max_buf_size: None,
pipeline_flush: false,
}
}
}
impl<E> Http<E> {
pub fn http1_only(&mut self, val: bool) -> &mut Self {
if val {
self.mode = ConnectionMode::H1Only;
} else {
self.mode = ConnectionMode::Fallback;
}
self
}
#[inline]
pub fn http1_half_close(&mut self, val: bool) -> &mut Self {
self.h1_half_close = val;
self
}
#[inline]
pub fn http1_writev(&mut self, val: bool) -> &mut Self {
self.h1_writev = val;
self
}
pub fn http2_only(&mut self, val: bool) -> &mut Self {
if val {
self.mode = ConnectionMode::H2Only;
} else {
self.mode = ConnectionMode::Fallback;
}
self
}
pub fn http2_initial_stream_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
if let Some(sz) = sz.into() {
self.h2_builder.initial_window_size(sz);
}
self
}
pub fn http2_initial_connection_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
if let Some(sz) = sz.into() {
self.h2_builder.initial_connection_window_size(sz);
}
self
}
pub fn http2_max_concurrent_streams(&mut self, max: impl Into<Option<u32>>) -> &mut Self {
if let Some(max) = max.into() {
self.h2_builder.max_concurrent_streams(max);
}
self
}
pub fn keep_alive(&mut self, val: bool) -> &mut Self {
self.keep_alive = val;
self
}
pub fn max_buf_size(&mut self, max: usize) -> &mut Self {
assert!(
max >= proto::h1::MINIMUM_MAX_BUFFER_SIZE,
"the max_buf_size cannot be smaller than the minimum that h1 specifies."
);
self.max_buf_size = Some(max);
self
}
pub fn pipeline_flush(&mut self, enabled: bool) -> &mut Self {
self.pipeline_flush = enabled;
self
}
pub fn with_executor<E2>(self, exec: E2) -> Http<E2> {
Http {
exec,
h1_half_close: self.h1_half_close,
h1_writev: self.h1_writev,
h2_builder: self.h2_builder,
mode: self.mode,
keep_alive: self.keep_alive,
max_buf_size: self.max_buf_size,
pipeline_flush: self.pipeline_flush,
}
}
pub fn serve_connection<S, I, Bd>(&self, io: I, service: S) -> Connection<I, S, E>
where
S: Service<Body, ResBody=Bd>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
Bd: Payload,
Bd::Data: Unpin,
I: AsyncRead + AsyncWrite + Unpin,
E: H2Exec<S::Future, Bd>,
{
let either = match self.mode {
ConnectionMode::H1Only | ConnectionMode::Fallback => {
let mut conn = proto::Conn::new(io);
if !self.keep_alive {
conn.disable_keep_alive();
}
if !self.h1_half_close {
conn.set_disable_half_close();
}
if !self.h1_writev {
conn.set_write_strategy_flatten();
}
conn.set_flush_pipeline(self.pipeline_flush);
if let Some(max) = self.max_buf_size {
conn.set_max_buf_size(max);
}
let sd = proto::h1::dispatch::Server::new(service);
Either::A(proto::h1::Dispatcher::new(sd, conn))
}
ConnectionMode::H2Only => {
let rewind_io = Rewind::new(io);
let h2 = proto::h2::Server::new(rewind_io, service, &self.h2_builder, self.exec.clone());
Either::B(h2)
}
};
Connection {
conn: Some(either),
fallback: if self.mode == ConnectionMode::Fallback {
Fallback::ToHttp2(self.h2_builder.clone(), self.exec.clone())
} else {
Fallback::Http1Only
},
}
}
#[cfg(feature = "tcp")]
#[doc(hidden)]
#[deprecated]
#[allow(deprecated)]
pub fn serve_addr<S, Bd>(&self, addr: &SocketAddr, make_service: S) -> crate::Result<Serve<AddrIncoming, S, E>>
where
S: MakeServiceRef<
AddrStream,
Body,
ResBody=Bd,
>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
Bd: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, Bd>,
{
let mut incoming = AddrIncoming::new(addr, None)?;
if self.keep_alive {
incoming.set_keepalive(Some(Duration::from_secs(90)));
}
Ok(self.serve_incoming(incoming, make_service))
}
#[cfg(feature = "tcp")]
#[doc(hidden)]
#[deprecated]
#[allow(deprecated)]
pub fn serve_addr_handle<S, Bd>(&self, addr: &SocketAddr, handle: &Handle, make_service: S) -> crate::Result<Serve<AddrIncoming, S, E>>
where
S: MakeServiceRef<
AddrStream,
Body,
ResBody=Bd,
>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
Bd: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, Bd>,
{
let mut incoming = AddrIncoming::new(addr, Some(handle))?;
if self.keep_alive {
incoming.set_keepalive(Some(Duration::from_secs(90)));
}
Ok(self.serve_incoming(incoming, make_service))
}
#[doc(hidden)]
#[deprecated]
pub fn serve_incoming<I, IO, IE, S, Bd>(&self, incoming: I, make_service: S) -> Serve<I, S, E>
where
I: Accept<Conn=IO, Error=IE>,
IE: Into<Box<dyn StdError + Send + Sync>>,
IO: AsyncRead + AsyncWrite + Unpin,
S: MakeServiceRef<
IO,
Body,
ResBody=Bd,
>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
Bd: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, Bd>,
{
Serve {
incoming,
make_service,
protocol: self.clone(),
}
}
pub(super) fn serve<I, IO, IE, S, Bd>(&self, incoming: I, make_service: S) -> Serve<I, S, E>
where
I: Accept<Conn=IO, Error=IE>,
IE: Into<Box<dyn StdError + Send + Sync>>,
IO: AsyncRead + AsyncWrite + Unpin,
S: MakeServiceRef<
IO,
Body,
ResBody=Bd,
>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
Bd: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, Bd>,
{
Serve {
incoming,
make_service,
protocol: self.clone(),
}
}
}
impl<I, B, S, E> Connection<I, S, E>
where
S: Service<Body, ResBody=B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin,
B: Payload + 'static,
B::Data: Unpin,
E: H2Exec<S::Future, B>,
{
pub fn graceful_shutdown(self: Pin<&mut Self>) {
match self.project().conn.as_mut().unwrap() {
Either::A(ref mut h1) => {
h1.disable_keep_alive();
},
Either::B(ref mut h2) => {
h2.graceful_shutdown();
}
}
}
pub fn into_parts(self) -> Parts<I, S> {
self.try_into_parts().unwrap_or_else(|| panic!("h2 cannot into_inner"))
}
pub fn try_into_parts(self) -> Option<Parts<I, S>> {
match self.conn.unwrap() {
Either::A(h1) => {
let (io, read_buf, dispatch) = h1.into_inner();
Some(Parts {
io: io,
read_buf: read_buf,
service: dispatch.into_service(),
_inner: (),
})
},
Either::B(_h2) => None,
}
}
pub fn poll_without_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>>
where
S: Unpin,
S::Future: Unpin,
B: Unpin,
{
loop {
let polled = match *self.conn.as_mut().unwrap() {
Either::A(ref mut h1) => h1.poll_without_shutdown(cx),
Either::B(ref mut h2) => return Pin::new(h2).poll(cx).map_ok(|_| ()),
};
match ready!(polled) {
Ok(x) => return Poll::Ready(Ok(x)),
Err(e) => {
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.fallback.to_h2() => {
self.upgrade_h2();
continue;
}
_ => return Poll::Ready(Err(e)),
}
}
}
}
}
pub fn without_shutdown(self) -> impl Future<Output=crate::Result<Parts<I, S>>>
where
S: Unpin,
S::Future: Unpin,
B: Unpin,
{
let mut conn = Some(self);
futures_util::future::poll_fn(move |cx| {
ready!(conn.as_mut().unwrap().poll_without_shutdown(cx))?;
Poll::Ready(Ok(conn.take().unwrap().into_parts()))
})
}
fn upgrade_h2(&mut self) {
trace!("Trying to upgrade connection to h2");
let conn = self.conn.take();
let (io, read_buf, dispatch) = match conn.unwrap() {
Either::A(h1) => {
h1.into_inner()
},
Either::B(_h2) => {
panic!("h2 cannot into_inner");
}
};
let mut rewind_io = Rewind::new(io);
rewind_io.rewind(read_buf);
let (builder, exec) = match self.fallback {
Fallback::ToHttp2(ref builder, ref exec) => (builder, exec),
Fallback::Http1Only => unreachable!("upgrade_h2 with Fallback::Http1Only"),
};
let h2 = proto::h2::Server::new(
rewind_io,
dispatch.into_service(),
builder,
exec.clone(),
);
debug_assert!(self.conn.is_none());
self.conn = Some(Either::B(h2));
}
pub fn with_upgrades(self) -> UpgradeableConnection<I, S, E>
where
I: Send,
{
UpgradeableConnection {
inner: self,
}
}
}
impl<I, B, S, E> Future for Connection<I, S, E>
where
S: Service<Body, ResBody=B> + 'static,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin + 'static,
B: Payload + 'static,
B::Data: Unpin,
E: H2Exec<S::Future, B>,
{
type Output = crate::Result<()>;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
loop {
match ready!(Pin::new(self.conn.as_mut().unwrap()).poll(cx)) {
Ok(done) => {
if let proto::Dispatched::Upgrade(pending) = done {
pending.manual();
}
return Poll::Ready(Ok(()));
},
Err(e) => {
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.fallback.to_h2() => {
self.upgrade_h2();
continue;
}
_ => return Poll::Ready(Err(e)),
}
}
}
}
}
}
impl<I, S> fmt::Debug for Connection<I, S>
where
S: Service<Body>,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Connection")
.finish()
}
}
impl<I, S, E> Serve<I, S, E> {
#[inline]
pub fn incoming_ref(&self) -> &I {
&self.incoming
}
#[inline]
pub fn incoming_mut(&mut self) -> &mut I {
&mut self.incoming
}
pub(super) fn spawn_all(self) -> SpawnAll<I, S, E> {
SpawnAll {
serve: self,
}
}
}
impl<I, IO, IE, S, B, E> Serve<I, S, E>
where
I: Accept<Conn=IO, Error=IE>,
IO: AsyncRead + AsyncWrite + Unpin,
IE: Into<Box<dyn StdError + Send + Sync>>,
S: MakeServiceRef<IO, Body, ResBody=B>,
B: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, B>,
{
fn poll_next_(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<crate::Result<Connecting<IO, S::Future, E>>>> {
let me = self.project();
match ready!(me.make_service.poll_ready_ref(cx)) {
Ok(()) => (),
Err(e) => {
trace!("make_service closed");
return Poll::Ready(Some(Err(crate::Error::new_user_make_service(e))));
}
}
if let Some(item) = ready!(me.incoming.poll_accept(cx)) {
let io = item.map_err(crate::Error::new_accept)?;
let new_fut = me.make_service.make_service_ref(&io);
Poll::Ready(Some(Ok(Connecting {
future: new_fut,
io: Some(io),
protocol: me.protocol.clone(),
})))
} else {
Poll::Ready(None)
}
}
}
impl<I, IO, IE, S, B, E> Stream for Serve<I, S, E>
where
I: Accept<Conn=IO, Error=IE>,
IO: AsyncRead + AsyncWrite + Unpin,
IE: Into<Box<dyn StdError + Send + Sync>>,
S: MakeServiceRef<IO, Body, ResBody=B>,
B: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, B>,
{
type Item = crate::Result<Connecting<IO, S::Future, E>>;
fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
self.poll_next_(cx)
}
}
impl<I, F, S, FE, E, B> Future for Connecting<I, F, E>
where
I: AsyncRead + AsyncWrite + Unpin,
F: Future<Output=Result<S, FE>>,
S: Service<Body, ResBody=B>,
B: Payload,
B::Data: Unpin,
E: H2Exec<S::Future, B>,
{
type Output = Result<Connection<I, S, E>, FE>;
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
let me = self.project();
let service = ready!(me.future.poll(cx))?;
let io = me.io.take().expect("polled after complete");
Poll::Ready(Ok(me.protocol.serve_connection(io, service)))
}
}
#[cfg(feature = "tcp")]
impl<S, E> SpawnAll<AddrIncoming, S, E> {
pub(super) fn local_addr(&self) -> SocketAddr {
self.serve.incoming.local_addr()
}
}
impl<I, S, E> SpawnAll<I, S, E> {
pub(super) fn incoming_ref(&self) -> &I {
self.serve.incoming_ref()
}
}
impl<I, IO, IE, S, B, E> SpawnAll<I, S, E>
where
I: Accept<Conn=IO, Error=IE>,
IE: Into<Box<dyn StdError + Send + Sync>>,
IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
S: MakeServiceRef<
IO,
Body,
ResBody=B,
>,
B: Payload,
E: H2Exec<<S::Service as Service<Body>>::Future, B>,
{
pub(super) fn poll_watch<W>(self: Pin<&mut Self>, cx: &mut task::Context<'_>, watcher: &W) -> Poll<crate::Result<()>>
where
E: NewSvcExec<IO, S::Future, S::Service, E, W>,
W: Watcher<IO, S::Service, E>,
{
let mut me = self.project();
loop {
if let Some(connecting) = ready!(me.serve.as_mut().poll_next_(cx)?) {
let fut = NewSvcTask::new(connecting, watcher.clone());
me.serve.as_mut().project().protocol.exec.execute_new_svc(fut)?;
} else {
return Poll::Ready(Ok(()));
}
}
}
}
impl<A, B> Future for Either<A, B>
where
A: Future,
B: Future<Output=A::Output>,
{
type Output = A::Output;
#[project]
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
#[project]
match self.project() {
Either::A(a) => a.poll(cx),
Either::B(b) => b.poll(cx),
}
}
}
pub(crate) mod spawn_all {
use std::error::Error as StdError;
use tokio_io::{AsyncRead, AsyncWrite};
use crate::body::{Body, Payload};
use crate::common::exec::H2Exec;
use crate::common::{Future, Pin, Poll, Unpin, task};
use crate::service::Service;
use super::{Connecting, UpgradeableConnection};
use pin_project::{pin_project, project};
pub trait Watcher<I, S: Service<Body>, E>: Clone {
type Future: Future<Output = crate::Result<()>>;
fn watch(&self, conn: UpgradeableConnection<I, S, E>) -> Self::Future;
}
#[allow(missing_debug_implementations)]
#[derive(Copy, Clone)]
pub struct NoopWatcher;
impl<I, S, E> Watcher<I, S, E> for NoopWatcher
where
I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
S: Service<Body> + 'static,
<S::ResBody as Payload>::Data: Unpin,
E: H2Exec<S::Future, S::ResBody>,
{
type Future = UpgradeableConnection<I, S, E>;
fn watch(&self, conn: UpgradeableConnection<I, S, E>) -> Self::Future {
conn
}
}
#[pin_project]
#[allow(missing_debug_implementations)]
pub struct NewSvcTask<I, N, S: Service<Body>, E, W: Watcher<I, S, E>> {
#[pin]
state: State<I, N, S, E, W>,
}
#[pin_project]
pub enum State<I, N, S: Service<Body>, E, W: Watcher<I, S, E>> {
Connecting(#[pin] Connecting<I, N, E>, W),
Connected(#[pin] W::Future),
}
impl<I, N, S: Service<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),
}
}
}
impl<I, N, S, NE, B, E, W> Future for NewSvcTask<I, N, S, E, W>
where
I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
N: Future<Output=Result<S, NE>>,
NE: Into<Box<dyn StdError + Send + Sync>>,
S: Service<Body, ResBody=B>,
B: Payload,
B::Data: Unpin,
E: H2Exec<S::Future, B>,
W: Watcher<I, S, E>,
{
type Output = ();
#[project]
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
let mut me = self.project();
loop {
let next = {
#[project]
match me.state.as_mut().project() {
State::Connecting(connecting, watcher) => {
let res = ready!(connecting.poll(cx));
let conn = match res {
Ok(conn) => conn,
Err(err) => {
let err = crate::Error::new_user_make_service(err);
debug!("connecting error: {}", err);
return Poll::Ready(());
}
};
let connected = watcher.watch(conn.with_upgrades());
State::Connected(connected)
},
State::Connected(future) => {
return future
.poll(cx)
.map(|res| {
if let Err(err) = res {
debug!("connection error: {}", err);
}
});
}
}
};
me.state.set(next);
}
}
}
}
mod upgrades {
use super::*;
#[must_use = "futures do nothing unless polled"]
#[allow(missing_debug_implementations)]
pub struct UpgradeableConnection<T, S, E>
where
S: Service<Body>,
{
pub(super) inner: Connection<T, S, E>,
}
impl<I, B, S, E> UpgradeableConnection<I, S, E>
where
S: Service<Body, ResBody=B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin,
B: Payload + 'static,
B::Data: Unpin,
E: H2Exec<S::Future, B>,
{
pub fn graceful_shutdown(mut self: Pin<&mut Self>) {
Pin::new(&mut self.inner).graceful_shutdown()
}
}
impl<I, B, S, E> Future for UpgradeableConnection<I, S, E>
where
S: Service<Body, ResBody=B> + 'static,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
B: Payload + 'static,
B::Data: Unpin,
E: super::H2Exec<S::Future, B>,
{
type Output = crate::Result<()>;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
loop {
match ready!(Pin::new(self.inner.conn.as_mut().unwrap()).poll(cx)) {
Ok(proto::Dispatched::Shutdown) => return Poll::Ready(Ok(())),
Ok(proto::Dispatched::Upgrade(pending)) => {
let h1 = match mem::replace(&mut self.inner.conn, None) {
Some(Either::A(h1)) => h1,
_ => unreachable!("Upgrade expects h1"),
};
let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(Box::new(io), buf));
return Poll::Ready(Ok(()));
},
Err(e) => {
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.inner.fallback.to_h2() => {
self.inner.upgrade_h2();
continue;
}
_ => return Poll::Ready(Err(e)),
}
}
}
}
}
}
}