//! 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!"))) //! } //! # } //! ```
/// 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"))))] pubstruct 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"))))] pubstruct Connection<T, S, E = Exec> where
S: HttpService<Body>,
{ pub(super) conn: Option<ProtoServer<T, S::ResBody, S, E>>,
fallback: 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"))))] pubstruct 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. pubfn 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,
}
}
}
/// 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")))] pubfn http1_half_close(&mutself, val: bool) -> &mutSelf { self.h1_half_close = 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")))] pubfn http1_title_case_headers(&mutself, enabled: bool) -> &style='color:red'>mutSelf { 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")))] pubfn http1_preserve_header_case(&mutself, enabled: bool) -> &n style='color:red'>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"))))] pubfn http1_header_read_timeout(&mutself, read_timeout: Duration) -> &mutSelf { 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")))] pubfn http1_writev(&mutself, val: bool) -> &mutSelf { self.h1_writev = Some(val); 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. /// /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_INITIAL_WINDOW_SIZE #[cfg(feature = "http2")] #[cfg_attr(docsrs, doc(cfg(feature = "http2")))] pubfn http2_initial_stream_window_size(&mutself, sz: impl Into<Option<u32>>) -> &mutSelf { iflet 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")))] pubfn http2_initial_connection_window_size(
&mutself,
sz: impl Into<Option<u32>>,
) -> &mutSelf { iflet 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")))] pubfn http2_adaptive_window(&mutself, enabled: bool) -> &le='color:red'>mutSelf { use proto::h2::SPEC_WINDOW_SIZE;
/// 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")))] pubfn http2_max_frame_size(&mutself, sz: impl Into<Option<u32>>) -> &pan style='color:red'>mut Self { iflet 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. /// /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_MAX_CONCURRENT_STREAMS #[cfg(feature = "http2")] #[cfg_attr(docsrs, doc(cfg(feature = "http2")))] pubfn http2_max_concurrent_streams(&mutself, max: impl Into<Option<u32>>) -> &mutSelf { 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")))] pubfn http2_keep_alive_interval(
&mutself,
interval: impl Into<Option<Duration>>,
) -> &mutSelf { 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")))] pubfn http2_keep_alive_timeout(&mutself, timeout: Duration) -> &pan style='color:red'>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")))] pubfn http2_max_send_buf_size(&mutself, max: usize) -> &e='color:red'>mutSelf {
assert!(max <= std::u32::MAX as usize); self.h2_builder.max_send_buffer_size = max; 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")))] pubfn http2_max_header_list_size(&mutself, max: u32) -> &le='color:red'>mutSelf { 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")))] pubfn max_buf_size(&mutself, max: usize) -> &mutSelf {
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. pubfn pipeline_flush(&mutself, enabled: bool) -> &mutSelf { self.pipeline_flush = enabled; self
}
/// 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. pubfn 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. pubfn try_into_parts(self) -> Option<Parts<I, S>> { matchself.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,
/// 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`. pubfn poll_without_shutdown(&mutself, 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 { refmut 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) ifself.fallback.to_h2() => { self.upgrade_h2(); continue;
}
_ => (),
}
/// 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. pubfn without_shutdown(self) -> impl Future<Output = crate::Result<Parts<I, S>>> where
S: Unpin,
S::Future: Unpin,
B: Unpin,
{ letmut 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(&mutself) {
trace!("Trying to upgrade connection to h2"); let conn = self.conn.take();
/// Enable this connection to support higher-level HTTP upgrades. /// /// See [the `upgrade` module](crate::upgrade) for more. pubfn with_upgrades(self) -> UpgradeableConnection<I, S, E> where
I: Send,
{
UpgradeableConnection { inner: self }
}
}
fn poll(mutself: Pin<&mutSelf>, 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) ifself.fallback.to_h2() => { self.upgrade_h2(); continue;
}
_ => (),
}
#[cfg(any(feature = "http1", feature = "http2"))] mod upgrades { usesuper::*;
// 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)] pubstruct 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. pubfn graceful_shutdown(mutself: Pin<&mutSelf>) {
Pin::new(&mutself.inner).graceful_shutdown()
}
}
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.
