Source code
Revision control
Copy as Markdown
Other Tools
//! Lower-level Server connection API.
//!
//! The types in this module are to provide a lower-level API based around a
//! single connection. Accepting a connection and binding it with a service
//! are not handled at this level. This module provides the building blocks to
//! customize those things externally.
//!
//! If you don't have need to manage connections yourself, consider using the
//! higher-level [Server](super) API.
//!
//! ## Example
//! A simple example that uses the `Http` struct to talk HTTP over a Tokio TCP stream
//! ```no_run
//! # #[cfg(all(feature = "http1", feature = "runtime"))]
//! # mod rt {
//! use http::{Request, Response, StatusCode};
//! use hyper::{server::conn::Http, service::service_fn, Body};
//! use std::{net::SocketAddr, convert::Infallible};
//! use tokio::net::TcpListener;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
//! let addr: SocketAddr = ([127, 0, 0, 1], 8080).into();
//!
//! let mut tcp_listener = TcpListener::bind(addr).await?;
//! loop {
//! let (tcp_stream, _) = tcp_listener.accept().await?;
//! tokio::task::spawn(async move {
//! if let Err(http_err) = Http::new()
//! .http1_only(true)
//! .http1_keep_alive(true)
//! .serve_connection(tcp_stream, service_fn(hello))
//! .await {
//! eprintln!("Error while serving HTTP connection: {}", http_err);
//! }
//! });
//! }
//! }
//!
//! async fn hello(_req: Request<Body>) -> Result<Response<Body>, Infallible> {
//! Ok(Response::new(Body::from("Hello World!")))
//! }
//! # }
//! ```
#[cfg(all(
any(feature = "http1", feature = "http2"),
not(all(feature = "http1", feature = "http2"))
))]
use std::marker::PhantomData;
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "runtime"))]
use std::time::Duration;
#[cfg(feature = "http2")]
use crate::common::io::Rewind;
#[cfg(all(feature = "http1", feature = "http2"))]
use crate::error::{Kind, Parse};
#[cfg(feature = "http1")]
use crate::upgrade::Upgraded;
cfg_feature! {
#![any(feature = "http1", feature = "http2")]
use std::error::Error as StdError;
use std::fmt;
use bytes::Bytes;
use pin_project_lite::pin_project;
use tokio::io::{AsyncRead, AsyncWrite};
use tracing::trace;
pub use super::server::Connecting;
use crate::body::{Body, HttpBody};
use crate::common::{task, Future, Pin, Poll, Unpin};
#[cfg(not(all(feature = "http1", feature = "http2")))]
use crate::common::Never;
use crate::common::exec::{ConnStreamExec, Exec};
use crate::proto;
use crate::service::HttpService;
pub(super) use self::upgrades::UpgradeableConnection;
}
#[cfg(feature = "tcp")]
pub use super::tcp::{AddrIncoming, AddrStream};
/// A lower-level configuration of the HTTP protocol.
///
/// This structure is used to configure options for an HTTP server connection.
///
/// If you don't have need to manage connections yourself, consider using the
/// higher-level [Server](super) API.
#[derive(Clone, Debug)]
#[cfg(any(feature = "http1", feature = "http2"))]
#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
pub struct Http<E = Exec> {
pub(crate) exec: E,
h1_half_close: bool,
h1_keep_alive: bool,
h1_title_case_headers: bool,
h1_preserve_header_case: bool,
#[cfg(all(feature = "http1", feature = "runtime"))]
h1_header_read_timeout: Option<Duration>,
h1_writev: Option<bool>,
#[cfg(feature = "http2")]
h2_builder: proto::h2::server::Config,
mode: ConnectionMode,
max_buf_size: Option<usize>,
pipeline_flush: bool,
}
/// The internal mode of HTTP protocol which indicates the behavior when a parse error occurs.
#[cfg(any(feature = "http1", feature = "http2"))]
#[derive(Clone, Debug, PartialEq)]
enum ConnectionMode {
/// Always use HTTP/1 and do not upgrade when a parse error occurs.
#[cfg(feature = "http1")]
H1Only,
/// Always use HTTP/2.
#[cfg(feature = "http2")]
H2Only,
/// Use HTTP/1 and try to upgrade to h2 when a parse error occurs.
#[cfg(all(feature = "http1", feature = "http2"))]
Fallback,
}
#[cfg(any(feature = "http1", feature = "http2"))]
pin_project! {
/// A future binding a connection with a Service.
///
/// Polling this future will drive HTTP forward.
#[must_use = "futures do nothing unless polled"]
#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
pub struct Connection<T, S, E = Exec>
where
S: HttpService<Body>,
{
pub(super) conn: Option<ProtoServer<T, S::ResBody, S, E>>,
fallback: Fallback<E>,
}
}
#[cfg(feature = "http1")]
type Http1Dispatcher<T, B, S> =
proto::h1::Dispatcher<proto::h1::dispatch::Server<S, Body>, B, T, proto::ServerTransaction>;
#[cfg(all(not(feature = "http1"), feature = "http2"))]
type Http1Dispatcher<T, B, S> = (Never, PhantomData<(T, Box<Pin<B>>, Box<Pin<S>>)>);
#[cfg(feature = "http2")]
type Http2Server<T, B, S, E> = proto::h2::Server<Rewind<T>, S, B, E>;
#[cfg(all(not(feature = "http2"), feature = "http1"))]
type Http2Server<T, B, S, E> = (
Never,
PhantomData<(T, Box<Pin<S>>, Box<Pin<B>>, Box<Pin<E>>)>,
);
#[cfg(any(feature = "http1", feature = "http2"))]
pin_project! {
#[project = ProtoServerProj]
pub(super) enum ProtoServer<T, B, S, E = Exec>
where
S: HttpService<Body>,
B: HttpBody,
{
H1 {
#[pin]
h1: Http1Dispatcher<T, B, S>,
},
H2 {
#[pin]
h2: Http2Server<T, B, S, E>,
},
}
}
#[cfg(all(feature = "http1", feature = "http2"))]
#[derive(Clone, Debug)]
enum Fallback<E> {
ToHttp2(proto::h2::server::Config, E),
Http1Only,
}
#[cfg(all(
any(feature = "http1", feature = "http2"),
not(all(feature = "http1", feature = "http2"))
))]
type Fallback<E> = PhantomData<E>;
#[cfg(all(feature = "http1", feature = "http2"))]
impl<E> Fallback<E> {
fn to_h2(&self) -> bool {
match *self {
Fallback::ToHttp2(..) => true,
Fallback::Http1Only => false,
}
}
}
#[cfg(all(feature = "http1", feature = "http2"))]
impl<E> Unpin for Fallback<E> {}
/// Deconstructed parts of a `Connection`.
///
/// This allows taking apart a `Connection` at a later time, in order to
/// reclaim the IO object, and additional related pieces.
#[derive(Debug)]
#[cfg(any(feature = "http1", feature = "http2"))]
#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
pub struct Parts<T, S> {
/// The original IO object used in the handshake.
pub io: T,
/// A buffer of bytes that have been read but not processed as HTTP.
///
/// If the client sent additional bytes after its last request, and
/// this connection "ended" with an upgrade, the read buffer will contain
/// those bytes.
///
/// You will want to check for any existing bytes if you plan to continue
/// communicating on the IO object.
pub read_buf: Bytes,
/// The `Service` used to serve this connection.
pub service: S,
_inner: (),
}
// ===== impl Http =====
#[cfg(any(feature = "http1", feature = "http2"))]
impl Http {
/// Creates a new instance of the HTTP protocol, ready to spawn a server or
/// start accepting connections.
pub fn new() -> Http {
Http {
exec: Exec::Default,
h1_half_close: false,
h1_keep_alive: true,
h1_title_case_headers: false,
h1_preserve_header_case: false,
#[cfg(all(feature = "http1", feature = "runtime"))]
h1_header_read_timeout: None,
h1_writev: None,
#[cfg(feature = "http2")]
h2_builder: Default::default(),
mode: ConnectionMode::default(),
max_buf_size: None,
pipeline_flush: false,
}
}
}
#[cfg(any(feature = "http1", feature = "http2"))]
impl<E> Http<E> {
/// Sets whether HTTP1 is required.
///
/// Default is false
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_only(&mut self, val: bool) -> &mut Self {
if val {
self.mode = ConnectionMode::H1Only;
} else {
#[cfg(feature = "http2")]
{
self.mode = ConnectionMode::Fallback;
}
}
self
}
/// Set whether HTTP/1 connections should support half-closures.
///
/// Clients can chose to shutdown their write-side while waiting
/// for the server to respond. Setting this to `true` will
/// prevent closing the connection immediately if `read`
/// detects an EOF in the middle of a request.
///
/// Default is `false`.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_half_close(&mut self, val: bool) -> &mut Self {
self.h1_half_close = val;
self
}
/// Enables or disables HTTP/1 keep-alive.
///
/// Default is true.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_keep_alive(&mut self, val: bool) -> &mut Self {
self.h1_keep_alive = val;
self
}
/// Set whether HTTP/1 connections will write header names as title case at
/// the socket level.
///
/// Note that this setting does not affect HTTP/2.
///
/// Default is false.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_title_case_headers(&mut self, enabled: bool) -> &mut Self {
self.h1_title_case_headers = enabled;
self
}
/// Set whether to support preserving original header cases.
///
/// Currently, this will record the original cases received, and store them
/// in a private extension on the `Request`. It will also look for and use
/// such an extension in any provided `Response`.
///
/// Since the relevant extension is still private, there is no way to
/// interact with the original cases. The only effect this can have now is
/// to forward the cases in a proxy-like fashion.
///
/// Note that this setting does not affect HTTP/2.
///
/// Default is false.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_preserve_header_case(&mut self, enabled: bool) -> &mut Self {
self.h1_preserve_header_case = enabled;
self
}
/// Set a timeout for reading client request headers. If a client does not
/// transmit the entire header within this time, the connection is closed.
///
/// Default is None.
#[cfg(all(feature = "http1", feature = "runtime"))]
#[cfg_attr(docsrs, doc(cfg(all(feature = "http1", feature = "runtime"))))]
pub fn http1_header_read_timeout(&mut self, read_timeout: Duration) -> &mut Self {
self.h1_header_read_timeout = Some(read_timeout);
self
}
/// Set whether HTTP/1 connections should try to use vectored writes,
/// or always flatten into a single buffer.
///
/// Note that setting this to false may mean more copies of body data,
/// but may also improve performance when an IO transport doesn't
/// support vectored writes well, such as most TLS implementations.
///
/// Setting this to true will force hyper to use queued strategy
/// which may eliminate unnecessary cloning on some TLS backends
///
/// Default is `auto`. In this mode hyper will try to guess which
/// mode to use
#[inline]
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_writev(&mut self, val: bool) -> &mut Self {
self.h1_writev = Some(val);
self
}
/// Sets whether HTTP2 is required.
///
/// Default is false
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_only(&mut self, val: bool) -> &mut Self {
if val {
self.mode = ConnectionMode::H2Only;
} else {
#[cfg(feature = "http1")]
{
self.mode = ConnectionMode::Fallback;
}
}
self
}
/// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
/// stream-level flow control.
///
/// Passing `None` will do nothing.
///
/// If not set, hyper will use a default.
///
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
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.adaptive_window = false;
self.h2_builder.initial_stream_window_size = sz;
}
self
}
/// Sets the max connection-level flow control for HTTP2.
///
/// Passing `None` will do nothing.
///
/// If not set, hyper will use a default.
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
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.adaptive_window = false;
self.h2_builder.initial_conn_window_size = sz;
}
self
}
/// Sets whether to use an adaptive flow control.
///
/// Enabling this will override the limits set in
/// `http2_initial_stream_window_size` and
/// `http2_initial_connection_window_size`.
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_adaptive_window(&mut self, enabled: bool) -> &mut Self {
use proto::h2::SPEC_WINDOW_SIZE;
self.h2_builder.adaptive_window = enabled;
if enabled {
self.h2_builder.initial_conn_window_size = SPEC_WINDOW_SIZE;
self.h2_builder.initial_stream_window_size = SPEC_WINDOW_SIZE;
}
self
}
/// Sets the maximum frame size to use for HTTP2.
///
/// Passing `None` will do nothing.
///
/// If not set, hyper will use a default.
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_max_frame_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
if let Some(sz) = sz.into() {
self.h2_builder.max_frame_size = sz;
}
self
}
/// Sets the [`SETTINGS_MAX_CONCURRENT_STREAMS`][spec] option for HTTP2
/// connections.
///
/// Default is no limit (`std::u32::MAX`). Passing `None` will do nothing.
///
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_max_concurrent_streams(&mut self, max: impl Into<Option<u32>>) -> &mut Self {
self.h2_builder.max_concurrent_streams = max.into();
self
}
/// Sets an interval for HTTP2 Ping frames should be sent to keep a
/// connection alive.
///
/// Pass `None` to disable HTTP2 keep-alive.
///
/// Default is currently disabled.
///
/// # Cargo Feature
///
/// Requires the `runtime` cargo feature to be enabled.
#[cfg(feature = "runtime")]
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_keep_alive_interval(
&mut self,
interval: impl Into<Option<Duration>>,
) -> &mut Self {
self.h2_builder.keep_alive_interval = interval.into();
self
}
/// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
///
/// If the ping is not acknowledged within the timeout, the connection will
/// be closed. Does nothing if `http2_keep_alive_interval` is disabled.
///
/// Default is 20 seconds.
///
/// # Cargo Feature
///
/// Requires the `runtime` cargo feature to be enabled.
#[cfg(feature = "runtime")]
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_keep_alive_timeout(&mut self, timeout: Duration) -> &mut Self {
self.h2_builder.keep_alive_timeout = timeout;
self
}
/// Set the maximum write buffer size for each HTTP/2 stream.
///
/// Default is currently ~400KB, but may change.
///
/// # Panics
///
/// The value must be no larger than `u32::MAX`.
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_max_send_buf_size(&mut self, max: usize) -> &mut Self {
assert!(max <= std::u32::MAX as usize);
self.h2_builder.max_send_buffer_size = max;
self
}
/// Enables the [extended CONNECT protocol].
///
#[cfg(feature = "http2")]
pub fn http2_enable_connect_protocol(&mut self) -> &mut Self {
self.h2_builder.enable_connect_protocol = true;
self
}
/// Sets the max size of received header frames.
///
/// Default is currently ~16MB, but may change.
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_max_header_list_size(&mut self, max: u32) -> &mut Self {
self.h2_builder.max_header_list_size = max;
self
}
/// Set the maximum buffer size for the connection.
///
/// Default is ~400kb.
///
/// # Panics
///
/// The minimum value allowed is 8192. This method panics if the passed `max` is less than the minimum.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
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
}
/// Aggregates flushes to better support pipelined responses.
///
/// Experimental, may have bugs.
///
/// Default is false.
pub fn pipeline_flush(&mut self, enabled: bool) -> &mut Self {
self.pipeline_flush = enabled;
self
}
/// Set the executor used to spawn background tasks.
///
/// Default uses implicit default (like `tokio::spawn`).
pub fn with_executor<E2>(self, exec: E2) -> Http<E2> {
Http {
exec,
h1_half_close: self.h1_half_close,
h1_keep_alive: self.h1_keep_alive,
h1_title_case_headers: self.h1_title_case_headers,
h1_preserve_header_case: self.h1_preserve_header_case,
#[cfg(all(feature = "http1", feature = "runtime"))]
h1_header_read_timeout: self.h1_header_read_timeout,
h1_writev: self.h1_writev,
#[cfg(feature = "http2")]
h2_builder: self.h2_builder,
mode: self.mode,
max_buf_size: self.max_buf_size,
pipeline_flush: self.pipeline_flush,
}
}
/// Bind a connection together with a [`Service`](crate::service::Service).
///
/// This returns a Future that must be polled in order for HTTP to be
/// driven on the connection.
///
/// # Example
///
/// ```
/// # use hyper::{Body, Request, Response};
/// # use hyper::service::Service;
/// # use hyper::server::conn::Http;
/// # use tokio::io::{AsyncRead, AsyncWrite};
/// # async fn run<I, S>(some_io: I, some_service: S)
/// # where
/// # I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
/// # S: Service<hyper::Request<Body>, Response=hyper::Response<Body>> + Send + 'static,
/// # S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
/// # S::Future: Send,
/// # {
/// let http = Http::new();
/// let conn = http.serve_connection(some_io, some_service);
///
/// if let Err(e) = conn.await {
/// eprintln!("server connection error: {}", e);
/// }
/// # }
/// # fn main() {}
/// ```
pub fn serve_connection<S, I, Bd>(&self, io: I, service: S) -> Connection<I, S, E>
where
S: HttpService<Body, ResBody = Bd>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
Bd: HttpBody + 'static,
Bd::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin,
E: ConnStreamExec<S::Future, Bd>,
{
#[cfg(feature = "http1")]
macro_rules! h1 {
() => {{
let mut conn = proto::Conn::new(io);
if !self.h1_keep_alive {
conn.disable_keep_alive();
}
if self.h1_half_close {
conn.set_allow_half_close();
}
if self.h1_title_case_headers {
conn.set_title_case_headers();
}
if self.h1_preserve_header_case {
conn.set_preserve_header_case();
}
#[cfg(all(feature = "http1", feature = "runtime"))]
if let Some(header_read_timeout) = self.h1_header_read_timeout {
conn.set_http1_header_read_timeout(header_read_timeout);
}
if let Some(writev) = self.h1_writev {
if writev {
conn.set_write_strategy_queue();
} else {
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);
ProtoServer::H1 {
h1: proto::h1::Dispatcher::new(sd, conn),
}
}};
}
let proto = match self.mode {
#[cfg(feature = "http1")]
#[cfg(not(feature = "http2"))]
ConnectionMode::H1Only => h1!(),
#[cfg(feature = "http2")]
#[cfg(feature = "http1")]
ConnectionMode::H1Only | ConnectionMode::Fallback => h1!(),
#[cfg(feature = "http2")]
ConnectionMode::H2Only => {
let rewind_io = Rewind::new(io);
let h2 =
proto::h2::Server::new(rewind_io, service, &self.h2_builder, self.exec.clone());
ProtoServer::H2 { h2 }
}
};
Connection {
conn: Some(proto),
#[cfg(all(feature = "http1", feature = "http2"))]
fallback: if self.mode == ConnectionMode::Fallback {
Fallback::ToHttp2(self.h2_builder.clone(), self.exec.clone())
} else {
Fallback::Http1Only
},
#[cfg(not(all(feature = "http1", feature = "http2")))]
fallback: PhantomData,
}
}
}
// ===== impl Connection =====
#[cfg(any(feature = "http1", feature = "http2"))]
impl<I, B, S, E> Connection<I, S, E>
where
S: HttpService<Body, ResBody = B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin,
B: HttpBody + 'static,
B::Error: Into<Box<dyn StdError + Send + Sync>>,
E: ConnStreamExec<S::Future, B>,
{
/// Start a graceful shutdown process for this connection.
///
/// This `Connection` should continue to be polled until shutdown
/// can finish.
///
/// # Note
///
/// 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(mut self: Pin<&mut Self>) {
match self.conn {
#[cfg(feature = "http1")]
Some(ProtoServer::H1 { ref mut h1, .. }) => {
h1.disable_keep_alive();
}
#[cfg(feature = "http2")]
Some(ProtoServer::H2 { ref mut h2 }) => {
h2.graceful_shutdown();
}
None => (),
#[cfg(not(feature = "http1"))]
Some(ProtoServer::H1 { ref mut h1, .. }) => match h1.0 {},
#[cfg(not(feature = "http2"))]
Some(ProtoServer::H2 { ref mut h2 }) => match h2.0 {},
}
}
/// Return the inner IO object, and additional information.
///
/// If the IO object has been "rewound" the io will not contain those bytes rewound.
/// This should only be called after `poll_without_shutdown` signals
/// that the connection is "done". Otherwise, it may not have finished
/// flushing all necessary HTTP bytes.
///
/// # Panics
/// This method will panic if this connection is using an h2 protocol.
pub fn into_parts(self) -> Parts<I, S> {
self.try_into_parts()
.unwrap_or_else(|| panic!("h2 cannot into_inner"))
}
/// Return the inner IO object, and additional information, if available.
///
/// This method will return a `None` if this connection is using an h2 protocol.
pub fn try_into_parts(self) -> Option<Parts<I, S>> {
match self.conn.unwrap() {
#[cfg(feature = "http1")]
ProtoServer::H1 { h1, .. } => {
let (io, read_buf, dispatch) = h1.into_inner();
Some(Parts {
io,
read_buf,
service: dispatch.into_service(),
_inner: (),
})
}
ProtoServer::H2 { .. } => None,
#[cfg(not(feature = "http1"))]
ProtoServer::H1 { h1, .. } => match h1.0 {},
}
}
/// Poll the connection for completion, but without calling `shutdown`
/// on the underlying IO.
///
/// This is useful to allow running a connection while doing an HTTP
/// upgrade. Once the upgrade is completed, the connection would be "done",
/// but it is not desired to actually shutdown the IO object. Instead you
/// would take it back using `into_parts`.
pub fn poll_without_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>>
where
S: Unpin,
S::Future: Unpin,
B: Unpin,
{
loop {
match *self.conn.as_mut().unwrap() {
#[cfg(feature = "http1")]
ProtoServer::H1 { ref mut h1, .. } => match ready!(h1.poll_without_shutdown(cx)) {
Ok(()) => return Poll::Ready(Ok(())),
Err(e) => {
#[cfg(feature = "http2")]
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.fallback.to_h2() => {
self.upgrade_h2();
continue;
}
_ => (),
}
return Poll::Ready(Err(e));
}
},
#[cfg(feature = "http2")]
ProtoServer::H2 { ref mut h2 } => return Pin::new(h2).poll(cx).map_ok(|_| ()),
#[cfg(not(feature = "http1"))]
ProtoServer::H1 { ref mut h1, .. } => match h1.0 {},
#[cfg(not(feature = "http2"))]
ProtoServer::H2 { ref mut h2 } => match h2.0 {},
};
}
}
/// 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`.
///
/// # Error
///
/// This errors if the underlying connection protocol is not HTTP/1.
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(conn.take().unwrap().try_into_parts().ok_or_else(crate::Error::new_without_shutdown_not_h1))
})
}
#[cfg(all(feature = "http1", feature = "http2"))]
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() {
ProtoServer::H1 { h1, .. } => h1.into_inner(),
ProtoServer::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(ProtoServer::H2 { h2 });
}
/// Enable this connection to support higher-level HTTP upgrades.
///
/// See [the `upgrade` module](crate::upgrade) for more.
pub fn with_upgrades(self) -> UpgradeableConnection<I, S, E>
where
I: Send,
{
UpgradeableConnection { inner: self }
}
}
#[cfg(any(feature = "http1", feature = "http2"))]
impl<I, B, S, E> Future for Connection<I, S, E>
where
S: HttpService<Body, ResBody = B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin + 'static,
B: HttpBody + 'static,
B::Error: Into<Box<dyn StdError + Send + Sync>>,
E: ConnStreamExec<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) => {
match done {
proto::Dispatched::Shutdown => {}
#[cfg(feature = "http1")]
proto::Dispatched::Upgrade(pending) => {
// With no `Send` bound on `I`, we can't try to do
// upgrades here. In case a user was trying to use
// `Body::on_upgrade` with this API, send a special
// error letting them know about that.
pending.manual();
}
};
return Poll::Ready(Ok(()));
}
Err(e) => {
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.fallback.to_h2() => {
self.upgrade_h2();
continue;
}
_ => (),
}
return Poll::Ready(Err(e));
}
}
}
}
}
#[cfg(any(feature = "http1", feature = "http2"))]
impl<I, S> fmt::Debug for Connection<I, S>
where
S: HttpService<Body>,
{
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Connection").finish()
}
}
// ===== impl ConnectionMode =====
#[cfg(any(feature = "http1", feature = "http2"))]
impl Default for ConnectionMode {
#[cfg(all(feature = "http1", feature = "http2"))]
fn default() -> ConnectionMode {
ConnectionMode::Fallback
}
#[cfg(all(feature = "http1", not(feature = "http2")))]
fn default() -> ConnectionMode {
ConnectionMode::H1Only
}
#[cfg(all(not(feature = "http1"), feature = "http2"))]
fn default() -> ConnectionMode {
ConnectionMode::H2Only
}
}
// ===== impl ProtoServer =====
#[cfg(any(feature = "http1", feature = "http2"))]
impl<T, B, S, E> Future for ProtoServer<T, B, S, E>
where
T: AsyncRead + AsyncWrite + Unpin,
S: HttpService<Body, ResBody = B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
B: HttpBody + 'static,
B::Error: Into<Box<dyn StdError + Send + Sync>>,
E: ConnStreamExec<S::Future, B>,
{
type Output = crate::Result<proto::Dispatched>;
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match self.project() {
#[cfg(feature = "http1")]
ProtoServerProj::H1 { h1, .. } => h1.poll(cx),
#[cfg(feature = "http2")]
ProtoServerProj::H2 { h2 } => h2.poll(cx),
#[cfg(not(feature = "http1"))]
ProtoServerProj::H1 { h1, .. } => match h1.0 {},
#[cfg(not(feature = "http2"))]
ProtoServerProj::H2 { h2 } => match h2.0 {},
}
}
}
#[cfg(any(feature = "http1", feature = "http2"))]
mod upgrades {
use super::*;
// A future binding a connection with a Service with Upgrade support.
//
// This type is unnameable outside the crate, and so basically just an
// `impl Future`, without requiring Rust 1.26.
#[must_use = "futures do nothing unless polled"]
#[allow(missing_debug_implementations)]
pub struct UpgradeableConnection<T, S, E>
where
S: HttpService<Body>,
{
pub(super) inner: Connection<T, S, E>,
}
impl<I, B, S, E> UpgradeableConnection<I, S, E>
where
S: HttpService<Body, ResBody = B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin,
B: HttpBody + 'static,
B::Error: Into<Box<dyn StdError + Send + Sync>>,
E: ConnStreamExec<S::Future, B>,
{
/// Start a graceful shutdown process for this connection.
///
/// This `Connection` should continue to be polled until shutdown
/// can finish.
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: HttpService<Body, ResBody = B>,
S::Error: Into<Box<dyn StdError + Send + Sync>>,
I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
B: HttpBody + 'static,
B::Error: Into<Box<dyn StdError + Send + Sync>>,
E: ConnStreamExec<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(())),
#[cfg(feature = "http1")]
Ok(proto::Dispatched::Upgrade(pending)) => {
match self.inner.conn.take() {
Some(ProtoServer::H1 { h1, .. }) => {
let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(io, buf));
return Poll::Ready(Ok(()));
}
_ => {
drop(pending);
unreachable!("Upgrade expects h1")
}
};
}
Err(e) => {
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.inner.fallback.to_h2() => {
self.inner.upgrade_h2();
continue;
}
_ => (),
}
return Poll::Ready(Err(e));
}
}
}
}
}
}