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

feat(http1): Make HTTP/1 support an optional feature

Browse Source 
cc #2251

BREAKING CHANGE: This puts all HTTP/1 methods and support behind an
  `http1` cargo feature, which will not be enabled by default. To use
  HTTP/1, add `features = ["http1"]` to the hyper dependency in your
  `Cargo.toml`.
This commit is contained in:
Sean McArthur
2020-11-16 15:39:10 -08:00
parent 2f2ceb2426
commit 2a19ab74ed
31 changed files with 459 additions and 239 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
Cargo.toml
View File

@@ -1,4 +1,3 @@
[package]
name = "hyper"
version = "0.14.0-dev" # don't forget to update html_root_url
@@ -76,11 +75,11 @@ default = [
"runtime",
"stream",
#"http1",
"http1",
"http2",
]
full = [
#"http1",
"http1",
"http2",
"stream",
"runtime",
@@ -97,7 +96,7 @@ tcp = [
]
# HTTP versions
#http1 = []
http1 = []
http2 = ["h2"]
# `impl Stream` for things

22
src/body/body.rs
View File

@@ -4,19 +4,23 @@ use std::error::Error as StdError;
use std::fmt;
use bytes::Bytes;
use futures_channel::{mpsc, oneshot};
use futures_channel::mpsc;
#[cfg(any(feature = "http1", feature = "http2"))]
use futures_channel::oneshot;
use futures_core::Stream; // for mpsc::Receiver
#[cfg(feature = "stream")]
use futures_util::TryStreamExt;
use http::HeaderMap;
use http_body::{Body as HttpBody, SizeHint};
use super::DecodedLength;
#[cfg(feature = "stream")]
use crate::common::sync_wrapper::SyncWrapper;
use crate::common::{task, watch, Future, Never, Pin, Poll};
use crate::common::{task, watch, Pin, Poll};
#[cfg(any(feature = "http1", feature = "http2"))]
use crate::common::{Future, Never};
#[cfg(feature = "http2")]
use crate::proto::h2::ping;
use crate::proto::DecodedLength;
use crate::upgrade::OnUpgrade;
type BodySender = mpsc::Sender<Result<Bytes, crate::Error>>;
@@ -67,14 +71,17 @@ struct Extra {
on_upgrade: OnUpgrade,
}
#[cfg(any(feature = "http1", feature = "http2"))]
type DelayEofUntil = oneshot::Receiver<Never>;
enum DelayEof {
/// Initial state, stream hasn't seen EOF yet.
#[cfg(any(feature = "http1", feature = "http2"))]
NotEof(DelayEofUntil),
/// Transitions to this state once we've seen `poll` try to
/// return EOF (`None`). This future is then polled, and
/// when it completes, the Body finally returns EOF (`None`).
#[cfg(any(feature = "http1", feature = "http2"))]
Eof(DelayEofUntil),
}
@@ -203,6 +210,7 @@ impl Body {
body
}
#[cfg(feature = "http1")]
pub(crate) fn set_on_upgrade(&mut self, upgrade: OnUpgrade) {
debug_assert!(!upgrade.is_none(), "set_on_upgrade with empty upgrade");
let extra = self.extra_mut();
@@ -210,6 +218,7 @@ impl Body {
extra.on_upgrade = upgrade;
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn delayed_eof(&mut self, fut: DelayEofUntil) {
self.extra_mut().delayed_eof = Some(DelayEof::NotEof(fut));
}
@@ -220,6 +229,7 @@ impl Body {
.and_then(|extra| extra.delayed_eof.take())
}
#[cfg(any(feature = "http1", feature = "http2"))]
fn extra_mut(&mut self) -> &mut Extra {
self.extra.get_or_insert_with(|| {
Box::new(Extra {
@@ -231,6 +241,7 @@ impl Body {
fn poll_eof(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<crate::Result<Bytes>>> {
match self.take_delayed_eof() {
#[cfg(any(feature = "http1", feature = "http2"))]
Some(DelayEof::NotEof(mut delay)) => match self.poll_inner(cx) {
ok @ Poll::Ready(Some(Ok(..))) | ok @ Poll::Pending => {
self.extra_mut().delayed_eof = Some(DelayEof::NotEof(delay));
@@ -246,6 +257,7 @@ impl Body {
},
Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
},
#[cfg(any(feature = "http1", feature = "http2"))]
Some(DelayEof::Eof(mut delay)) => match Pin::new(&mut delay).poll(cx) {
Poll::Ready(Ok(never)) => match never {},
Poll::Pending => {
@@ -254,6 +266,8 @@ impl Body {
}
Poll::Ready(Err(_done)) => Poll::Ready(None),
},
#[cfg(not(any(feature = "http1", feature = "http2")))]
Some(delay_eof) => match delay_eof {},
None => self.poll_inner(cx),
}
}
@@ -300,6 +314,7 @@ impl Body {
}
}
#[cfg(feature = "http1")]
pub(super) fn take_full_data(&mut self) -> Option<Bytes> {
if let Kind::Once(ref mut chunk) = self.kind {
chunk.take()
@@ -549,6 +564,7 @@ impl Sender {
.try_send(Err(crate::Error::new_body_write_aborted()));
}
#[cfg(feature = "http1")]
pub(crate) fn send_error(&mut self, err: crate::Error) {
let _ = self.tx.try_send(Err(err));
}

100
src/body/length.rs Normal file
View File

@@ -0,0 +1,100 @@
use std::fmt;
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) struct DecodedLength(u64);
#[cfg(any(feature = "http1", feature = "http2", test))]
const MAX_LEN: u64 = std::u64::MAX - 2;
impl DecodedLength {
pub(crate) const CLOSE_DELIMITED: DecodedLength = DecodedLength(::std::u64::MAX);
pub(crate) const CHUNKED: DecodedLength = DecodedLength(::std::u64::MAX - 1);
pub(crate) const ZERO: DecodedLength = DecodedLength(0);
#[cfg(test)]
pub(crate) fn new(len: u64) -> Self {
debug_assert!(len <= MAX_LEN);
DecodedLength(len)
}
/// Takes the length as a content-length without other checks.
///
/// Should only be called if previously confirmed this isn't
/// CLOSE_DELIMITED or CHUNKED.
#[inline]
#[cfg(feature = "http1")]
pub(crate) fn danger_len(self) -> u64 {
debug_assert!(self.0 < Self::CHUNKED.0);
self.0
}
/// Converts to an Option<u64> representing a Known or Unknown length.
pub(crate) fn into_opt(self) -> Option<u64> {
match self {
DecodedLength::CHUNKED | DecodedLength::CLOSE_DELIMITED => None,
DecodedLength(known) => Some(known),
}
}
/// Checks the `u64` is within the maximum allowed for content-length.
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn checked_new(len: u64) -> Result<Self, crate::error::Parse> {
if len <= MAX_LEN {
Ok(DecodedLength(len))
} else {
warn!("content-length bigger than maximum: {} > {}", len, MAX_LEN);
Err(crate::error::Parse::TooLarge)
}
}
pub(crate) fn sub_if(&mut self, amt: u64) {
match *self {
DecodedLength::CHUNKED | DecodedLength::CLOSE_DELIMITED => (),
DecodedLength(ref mut known) => {
*known -= amt;
}
}
}
}
impl fmt::Debug for DecodedLength {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
DecodedLength::CLOSE_DELIMITED => f.write_str("CLOSE_DELIMITED"),
DecodedLength::CHUNKED => f.write_str("CHUNKED"),
DecodedLength(n) => f.debug_tuple("DecodedLength").field(&n).finish(),
}
}
}
impl fmt::Display for DecodedLength {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
DecodedLength::CLOSE_DELIMITED => f.write_str("close-delimited"),
DecodedLength::CHUNKED => f.write_str("chunked encoding"),
DecodedLength::ZERO => f.write_str("empty"),
DecodedLength(n) => write!(f, "content-length ({} bytes)", n),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sub_if_known() {
let mut len = DecodedLength::new(30);
len.sub_if(20);
assert_eq!(len.0, 10);
}
#[test]
fn sub_if_chunked() {
let mut len = DecodedLength::CHUNKED;
len.sub_if(20);
assert_eq!(len, DecodedLength::CHUNKED);
}
}

3
src/body/mod.rs
View File

@@ -20,15 +20,18 @@ pub use http_body::Body as HttpBody;
pub use self::aggregate::aggregate;
pub use self::body::{Body, Sender};
pub(crate) use self::length::DecodedLength;
pub use self::to_bytes::to_bytes;
mod aggregate;
mod body;
mod length;
mod to_bytes;
/// An optimization to try to take a full body if immediately available.
///
/// This is currently limited to *only* `hyper::Body`s.
#[cfg(feature = "http1")]
pub(crate) fn take_full_data<T: HttpBody + 'static>(body: &mut T) -> Option<T::Data> {
use std::any::{Any, TypeId};

34
src/cfg.rs
View File

@@ -1,9 +1,37 @@
macro_rules! cfg_http2 {
macro_rules! cfg_any_http {
($($item:item)*) => {
$(
#[cfg(feature = "http2")]
//#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
#[cfg(any(
feature = "http1",
feature = "http2",
))]
#[cfg_attr(docsrs, doc(cfg(any(
feature = "http1",
feature = "http2",
))))]
$item
)*
}
}
cfg_any_http! {
macro_rules! cfg_http1 {
($($item:item)*) => {
$(
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
$item
)*
}
}
macro_rules! cfg_http2 {
($($item:item)*) => {
$(
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
$item
)*
}
}
}

71
src/client/conn.rs
View File

@@ -10,7 +10,8 @@
use std::error::Error as StdError;
use std::fmt;
use std::mem;
#[cfg(feature = "http2")]
use std::marker::PhantomData;
use std::sync::Arc;
#[cfg(feature = "runtime")]
#[cfg(feature = "http2")]
@@ -24,11 +25,14 @@ use tower_service::Service;
use super::dispatch;
use crate::body::HttpBody;
use crate::common::{task, BoxSendFuture, Exec, Executor, Future, Pin, Poll};
use crate::common::{task, BoxSendFuture, Exec, Future, Pin, Poll};
use crate::proto;
use crate::rt::Executor;
#[cfg(feature = "http1")]
use crate::upgrade::Upgraded;
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)]
@@ -36,9 +40,10 @@ 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>),
H2(#[pin] proto::h2::ClientTask<B>, PhantomData<fn(T)>),
}
/// Returns a handshake future over some IO.
@@ -88,6 +93,7 @@ pub struct Builder {
#[derive(Clone, Debug)]
enum Proto {
#[cfg(feature = "http1")]
Http1,
#[cfg(feature = "http2")]
Http2,
@@ -353,18 +359,20 @@ where
///
/// Only works for HTTP/1 connections. HTTP/2 connections will panic.
pub fn into_parts(self) -> Parts<T> {
let (io, read_buf, _) = match self.inner.expect("already upgraded") {
ProtoClient::H1(h1) => h1.into_inner(),
match self.inner.expect("already upgraded") {
#[cfg(feature = "http1")]
ProtoClient::H1(h1) => {
let (io, read_buf, _) = h1.into_inner();
Parts {
io,
read_buf,
_inner: (),
}
}
#[cfg(feature = "http2")]
ProtoClient::H2(_h2) => {
ProtoClient::H2(..) => {
panic!("http2 cannot into_inner");
}
};
Parts {
io,
read_buf,
_inner: (),
}
}
@@ -381,9 +389,10 @@ where
/// to work with this function; or use the `without_shutdown` wrapper.
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),
#[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(|_| ()),
}
}
@@ -410,16 +419,18 @@ where
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
match ready!(Pin::new(self.inner.as_mut().unwrap()).poll(cx))? {
proto::Dispatched::Shutdown => Poll::Ready(Ok(())),
proto::Dispatched::Upgrade(pending) => {
let h1 = match mem::replace(&mut self.inner, None) {
Some(ProtoClient::H1(h1)) => h1,
_ => unreachable!("Upgrade expects h1"),
};
let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(io, buf));
Poll::Ready(Ok(()))
}
#[cfg(feature = "http1")]
proto::Dispatched::Upgrade(pending) => match self.inner.take() {
Some(ProtoClient::H1(h1)) => {
let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(io, buf));
Poll::Ready(Ok(()))
}
_ => {
drop(pending);
unreachable!("Upgrade expects h1");
}
},
}
}
}
@@ -448,7 +459,10 @@ impl Builder {
h1_max_buf_size: None,
#[cfg(feature = "http2")]
h2_builder: Default::default(),
#[cfg(feature = "http1")]
version: Proto::Http1,
#[cfg(not(feature = "http1"))]
version: Proto::Http2,
}
}
@@ -477,6 +491,7 @@ impl Builder {
self
}
#[cfg(feature = "http1")]
pub(super) fn h1_max_buf_size(&mut self, max: usize) -> &mut Self {
assert!(
max >= proto::h1::MINIMUM_MAX_BUFFER_SIZE,
@@ -494,7 +509,9 @@ impl Builder {
#[cfg(feature = "http2")]
#[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
pub fn http2_only(&mut self, enabled: bool) -> &mut Builder {
self.version = if enabled { Proto::Http2 } else { Proto::Http1 };
if enabled {
self.version = Proto::Http2
}
self
}
@@ -643,6 +660,7 @@ impl Builder {
let (tx, rx) = dispatch::channel();
let proto = match opts.version {
#[cfg(feature = "http1")]
Proto::Http1 => {
let mut conn = proto::Conn::new(io);
if let Some(writev) = opts.h1_writev {
@@ -670,7 +688,7 @@ impl Builder {
let h2 =
proto::h2::client::handshake(io, rx, &opts.h2_builder, opts.exec.clone())
.await?;
ProtoClient::H2(h2)
ProtoClient::H2(h2, PhantomData)
}
};
@@ -723,9 +741,10 @@ 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),
#[cfg(feature = "http2")]
ProtoClientProj::H2(c) => c.poll(cx),
ProtoClientProj::H2(c, _) => c.poll(cx),
}
}
}

2
src/client/dispatch.rs
View File

@@ -162,11 +162,13 @@ impl<T, U> Receiver<T, U> {
}
}
#[cfg(feature = "http1")]
pub(crate) fn close(&mut self) {
self.taker.cancel();
self.inner.close();
}
#[cfg(feature = "http1")]
pub(crate) fn try_recv(&mut self) -> Option<(T, Callback<T, U>)> {
match self.inner.try_recv() {
Ok(mut env) => env.0.take(),

5
src/client/mod.rs
View File

@@ -62,7 +62,8 @@ use http::{Method, Request, Response, Uri, Version};
use self::connect::{sealed::Connect, Alpn, Connected, Connection};
use self::pool::{Key as PoolKey, Pool, Poolable, Pooled, Reservation};
use crate::body::{Body, HttpBody};
use crate::common::{lazy as hyper_lazy, task, BoxSendFuture, Executor, Future, Lazy, Pin, Poll};
use crate::common::{lazy as hyper_lazy, task, BoxSendFuture, Future, Lazy, Pin, Poll};
use crate::rt::Executor;
#[cfg(feature = "tcp")]
pub use self::connect::HttpConnector;
@@ -1022,6 +1023,8 @@ impl Builder {
/// # 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 http1_max_buf_size(&mut self, max: usize) -> &mut Self {
self.conn_builder.h1_max_buf_size(max);
self

1
src/common/buf.rs
View File

@@ -21,6 +21,7 @@ impl<T: Buf> BufList<T> {
}
#[inline]
#[cfg(feature = "http1")]
pub(crate) fn bufs_cnt(&self) -> usize {
self.bufs.len()
}

2
src/proto/h1/date.rs → src/common/date.rs
View File

@@ -10,12 +10,14 @@ use httpdate::HttpDate;
// "Sun, 06 Nov 1994 08:49:37 GMT".len()
pub const DATE_VALUE_LENGTH: usize = 29;
#[cfg(feature = "http1")]
pub fn extend(dst: &mut Vec<u8>) {
CACHED.with(|cache| {
dst.extend_from_slice(cache.borrow().buffer());
})
}
#[cfg(feature = "http1")]
pub fn update() {
CACHED.with(|cache| {
cache.borrow_mut().check();

7
src/common/exec.rs
View File

@@ -6,15 +6,10 @@ use std::sync::Arc;
use crate::body::{Body, HttpBody};
#[cfg(feature = "http2")]
use crate::proto::h2::server::H2Stream;
use crate::rt::Executor;
use crate::server::conn::spawn_all::{NewSvcTask, Watcher};
use crate::service::HttpService;
/// An executor of futures.
pub trait Executor<Fut> {
/// Place the future into the executor to be run.
fn execute(&self, fut: Fut);
}
pub trait ConnStreamExec<F, B: HttpBody>: Clone {
fn execute_h2stream(&mut self, fut: H2Stream<F, B>);
}

2
src/common/io/rewind.rs
View File

@@ -29,7 +29,7 @@ impl<T> Rewind<T> {
}
}
#[cfg(any(feature = "http2", test))]
#[cfg(any(all(feature = "http1", feature = "http2"), test))]
pub(crate) fn rewind(&mut self, bs: Bytes) {
debug_assert!(self.pre.is_none());
self.pre = Some(bs);

12
src/common/mod.rs
View File

@@ -8,9 +8,14 @@ macro_rules! ready {
}
pub(crate) mod buf;
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) mod date;
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) mod drain;
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) mod exec;
pub(crate) mod io;
#[cfg(any(feature = "http1", feature = "http2"))]
mod lazy;
mod never;
#[cfg(feature = "stream")]
@@ -18,11 +23,14 @@ pub(crate) mod sync_wrapper;
pub(crate) mod task;
pub(crate) mod watch;
pub use self::exec::Executor;
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) use self::exec::{BoxSendFuture, Exec};
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) use self::lazy::{lazy, Started as Lazy};
pub use self::never::Never;
pub(crate) use self::task::Poll;
// group up types normally needed for `Future`
pub(crate) use std::{future::Future, marker::Unpin, pin::Pin};
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) use std::marker::Unpin;
pub(crate) use std::{future::Future, pin::Pin};

2
src/common/task.rs
View File

@@ -1,9 +1,11 @@
#[cfg(feature = "http1")]
use super::Never;
pub(crate) use std::task::{Context, Poll};
/// A function to help "yield" a future, such that it is re-scheduled immediately.
///
/// Useful for spin counts, so a future doesn't hog too much time.
#[cfg(feature = "http1")]
pub(crate) fn yield_now(cx: &mut Context<'_>) -> Poll<Never> {
cx.waker().wake_by_ref();
Poll::Pending

63
src/error.rs
View File

@@ -1,7 +1,6 @@
//! Error and Result module.
use std::error::Error as StdError;
use std::fmt;
use std::io;
/// Result type often returned from methods that can have hyper `Error`s.
pub type Result<T> = std::result::Result<T, Error>;
@@ -25,27 +24,33 @@ pub(crate) enum Kind {
/// A message reached EOF, but is not complete.
IncompleteMessage,
/// A connection received a message (or bytes) when not waiting for one.
#[cfg(feature = "http1")]
UnexpectedMessage,
/// A pending item was dropped before ever being processed.
Canceled,
/// Indicates a channel (client or body sender) is closed.
ChannelClosed,
/// An `io::Error` that occurred while trying to read or write to a network stream.
#[cfg(any(feature = "http1", feature = "http2"))]
Io,
/// Error occurred while connecting.
Connect,
/// Error creating a TcpListener.
#[cfg(feature = "tcp")]
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "tcp"))]
Listen,
/// Error accepting on an Incoming stream.
#[cfg(any(feature = "http1", feature = "http2"))]
Accept,
/// Error while reading a body from connection.
#[cfg(any(feature = "http1", feature = "http2"))]
Body,
/// Error while writing a body to connection.
#[cfg(any(feature = "http1", feature = "http2"))]
BodyWrite,
/// The body write was aborted.
BodyWriteAborted,
/// Error calling AsyncWrite::shutdown()
#[cfg(feature = "http1")]
Shutdown,
/// A general error from h2.
@@ -57,6 +62,7 @@ pub(crate) enum Kind {
pub(crate) enum Parse {
Method,
Version,
#[cfg(feature = "http1")]
VersionH2,
Uri,
Header,
@@ -67,28 +73,37 @@ pub(crate) enum Parse {
#[derive(Debug, PartialEq)]
pub(crate) enum User {
/// Error calling user's HttpBody::poll_data().
#[cfg(any(feature = "http1", feature = "http2"))]
Body,
/// Error calling user's MakeService.
#[cfg(any(feature = "http1", feature = "http2"))]
MakeService,
/// Error from future of user's Service.
#[cfg(any(feature = "http1", feature = "http2"))]
Service,
/// User tried to send a certain header in an unexpected context.
///
/// For example, sending both `content-length` and `transfer-encoding`.
#[cfg(feature = "http1")]
UnexpectedHeader,
/// User tried to create a Request with bad version.
#[cfg(any(feature = "http1", feature = "http2"))]
UnsupportedVersion,
/// User tried to create a CONNECT Request with the Client.
#[cfg(any(feature = "http1", feature = "http2"))]
UnsupportedRequestMethod,
/// User tried to respond with a 1xx (not 101) response code.
#[cfg(feature = "http1")]
UnsupportedStatusCode,
/// User tried to send a Request with Client with non-absolute URI.
#[cfg(any(feature = "http1", feature = "http2"))]
AbsoluteUriRequired,
/// User tried polling for an upgrade that doesn't exist.
NoUpgrade,
/// User polled for an upgrade, but low-level API is not using upgrades.
#[cfg(feature = "http1")]
ManualUpgrade,
}
@@ -159,6 +174,7 @@ impl Error {
self
}
#[cfg(feature = "http1")]
pub(crate) fn kind(&self) -> &Kind {
&self.inner.kind
}
@@ -189,35 +205,42 @@ impl Error {
Error::new(Kind::Canceled)
}
#[cfg(feature = "http1")]
pub(crate) fn new_incomplete() -> Error {
Error::new(Kind::IncompleteMessage)
}
#[cfg(feature = "http1")]
pub(crate) fn new_too_large() -> Error {
Error::new(Kind::Parse(Parse::TooLarge))
}
#[cfg(feature = "http1")]
pub(crate) fn new_version_h2() -> Error {
Error::new(Kind::Parse(Parse::VersionH2))
}
#[cfg(feature = "http1")]
pub(crate) fn new_unexpected_message() -> Error {
Error::new(Kind::UnexpectedMessage)
}
pub(crate) fn new_io(cause: io::Error) -> Error {
#[cfg(feature = "http1")]
pub(crate) fn new_io(cause: std::io::Error) -> Error {
Error::new(Kind::Io).with(cause)
}
#[cfg(feature = "tcp")]
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "tcp"))]
pub(crate) fn new_listen<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Listen).with(cause)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_accept<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Accept).with(cause)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_connect<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Connect).with(cause)
}
@@ -226,10 +249,12 @@ impl Error {
Error::new(Kind::ChannelClosed)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_body<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::Body).with(cause)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_body_write<E: Into<Cause>>(cause: E) -> Error {
Error::new(Kind::BodyWrite).with(cause)
}
@@ -242,22 +267,27 @@ impl Error {
Error::new(Kind::User(user))
}
#[cfg(feature = "http1")]
pub(crate) fn new_user_header() -> Error {
Error::new_user(User::UnexpectedHeader)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_user_unsupported_version() -> Error {
Error::new_user(User::UnsupportedVersion)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_user_unsupported_request_method() -> Error {
Error::new_user(User::UnsupportedRequestMethod)
}
#[cfg(feature = "http1")]
pub(crate) fn new_user_unsupported_status_code() -> Error {
Error::new_user(User::UnsupportedStatusCode)
}
#[cfg(feature = "http1")]
pub(crate) fn new_user_absolute_uri_required() -> Error {
Error::new_user(User::AbsoluteUriRequired)
}
@@ -266,23 +296,28 @@ impl Error {
Error::new_user(User::NoUpgrade)
}
#[cfg(feature = "http1")]
pub(crate) fn new_user_manual_upgrade() -> Error {
Error::new_user(User::ManualUpgrade)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_user_make_service<E: Into<Cause>>(cause: E) -> Error {
Error::new_user(User::MakeService).with(cause)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_user_service<E: Into<Cause>>(cause: E) -> Error {
Error::new_user(User::Service).with(cause)
}
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) fn new_user_body<E: Into<Cause>>(cause: E) -> Error {
Error::new_user(User::Body).with(cause)
}
pub(crate) fn new_shutdown(cause: io::Error) -> Error {
#[cfg(feature = "http1")]
pub(crate) fn new_shutdown(cause: std::io::Error) -> Error {
Error::new(Kind::Shutdown).with(cause)
}
@@ -299,38 +334,54 @@ impl Error {
match self.inner.kind {
Kind::Parse(Parse::Method) => "invalid HTTP method parsed",
Kind::Parse(Parse::Version) => "invalid HTTP version parsed",
#[cfg(feature = "http1")]
Kind::Parse(Parse::VersionH2) => "invalid HTTP version parsed (found HTTP2 preface)",
Kind::Parse(Parse::Uri) => "invalid URI",
Kind::Parse(Parse::Header) => "invalid HTTP header parsed",
Kind::Parse(Parse::TooLarge) => "message head is too large",
Kind::Parse(Parse::Status) => "invalid HTTP status-code parsed",
Kind::IncompleteMessage => "connection closed before message completed",
#[cfg(feature = "http1")]
Kind::UnexpectedMessage => "received unexpected message from connection",
Kind::ChannelClosed => "channel closed",
Kind::Connect => "error trying to connect",
Kind::Canceled => "operation was canceled",
#[cfg(feature = "tcp")]
#[cfg(all(any(feature = "http1", feature = "http2"), feature = "tcp"))]
Kind::Listen => "error creating server listener",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::Accept => "error accepting connection",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::Body => "error reading a body from connection",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::BodyWrite => "error writing a body to connection",
Kind::BodyWriteAborted => "body write aborted",
#[cfg(feature = "http1")]
Kind::Shutdown => "error shutting down connection",
#[cfg(feature = "http2")]
Kind::Http2 => "http2 error",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::Io => "connection error",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::User(User::Body) => "error from user's HttpBody stream",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::User(User::MakeService) => "error from user's MakeService",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::User(User::Service) => "error from user's Service",
#[cfg(feature = "http1")]
Kind::User(User::UnexpectedHeader) => "user sent unexpected header",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::User(User::UnsupportedVersion) => "request has unsupported HTTP version",
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::User(User::UnsupportedRequestMethod) => "request has unsupported HTTP method",
#[cfg(feature = "http1")]
Kind::User(User::UnsupportedStatusCode) => {
"response has 1xx status code, not supported by server"
}
#[cfg(any(feature = "http1", feature = "http2"))]
Kind::User(User::AbsoluteUriRequired) => "client requires absolute-form URIs",
Kind::User(User::NoUpgrade) => "no upgrade available",
#[cfg(feature = "http1")]
Kind::User(User::ManualUpgrade) => "upgrade expected but low level API in use",
}
}

17
src/headers.rs
View File

@@ -1,18 +1,22 @@
#[cfg(feature = "http1")]
use bytes::BytesMut;
use http::header::{HeaderValue, OccupiedEntry, ValueIter};
use http::header::{CONTENT_LENGTH, TRANSFER_ENCODING};
use http::header::CONTENT_LENGTH;
use http::header::{HeaderValue, ValueIter};
#[cfg(feature = "http2")]
use http::method::Method;
use http::HeaderMap;
#[cfg(feature = "http1")]
pub fn connection_keep_alive(value: &HeaderValue) -> bool {
connection_has(value, "keep-alive")
}
#[cfg(feature = "http1")]
pub fn connection_close(value: &HeaderValue) -> bool {
connection_has(value, "close")
}
#[cfg(feature = "http1")]
fn connection_has(value: &HeaderValue, needle: &str) -> bool {
if let Ok(s) = value.to_str() {
for val in s.split(',') {
@@ -24,6 +28,7 @@ fn connection_has(value: &HeaderValue, needle: &str) -> bool {
false
}
#[cfg(feature = "http1")]
pub fn content_length_parse(value: &HeaderValue) -> Option<u64> {
value.to_str().ok().and_then(|s| s.parse().ok())
}
@@ -74,10 +79,12 @@ pub fn set_content_length_if_missing(headers: &mut HeaderMap, len: u64) {
.or_insert_with(|| HeaderValue::from(len));
}
#[cfg(feature = "http1")]
pub fn transfer_encoding_is_chunked(headers: &HeaderMap) -> bool {
is_chunked(headers.get_all(TRANSFER_ENCODING).into_iter())
is_chunked(headers.get_all(http::header::TRANSFER_ENCODING).into_iter())
}
#[cfg(feature = "http1")]
pub fn is_chunked(mut encodings: ValueIter<'_, HeaderValue>) -> bool {
// chunked must always be the last encoding, according to spec
if let Some(line) = encodings.next_back() {
@@ -87,6 +94,7 @@ pub fn is_chunked(mut encodings: ValueIter<'_, HeaderValue>) -> bool {
false
}
#[cfg(feature = "http1")]
pub fn is_chunked_(value: &HeaderValue) -> bool {
// chunked must always be the last encoding, according to spec
if let Ok(s) = value.to_str() {
@@ -98,7 +106,8 @@ pub fn is_chunked_(value: &HeaderValue) -> bool {
false
}
pub fn add_chunked(mut entry: OccupiedEntry<'_, HeaderValue>) {
#[cfg(feature = "http1")]
pub fn add_chunked(mut entry: http::header::OccupiedEntry<'_, HeaderValue>) {
const CHUNKED: &str = "chunked";
if let Some(line) = entry.iter_mut().next_back() {

18
src/lib.rs
View File

@@ -42,6 +42,7 @@
#[doc(hidden)]
pub use http;
#[cfg(any(feature = "http1", feature = "http2"))]
#[macro_use]
extern crate tracing;
@@ -51,23 +52,28 @@ extern crate test;
pub use http::{header, HeaderMap, Method, Request, Response, StatusCode, Uri, Version};
pub use crate::body::Body;
pub use crate::client::Client;
pub use crate::error::{Error, Result};
pub use crate::server::Server;
#[macro_use]
mod cfg;
#[macro_use]
mod common;
pub mod body;
pub mod client;
#[doc(hidden)] // Mistakenly public...
pub mod error;
mod headers;
#[cfg(test)]
mod mock;
mod proto;
#[cfg(any(feature = "http1", feature = "http2",))]
pub mod rt;
pub mod server;
pub mod service;
pub mod upgrade;
cfg_any_http! {
mod headers;
mod proto;
pub mod client;
pub mod server;
pub use crate::client::Client;
pub use crate::server::Server;
}

6
src/proto/h1/conn.rs
View File

@@ -9,10 +9,10 @@ use tokio::io::{AsyncRead, AsyncWrite};
use super::io::Buffered;
use super::{Decoder, Encode, EncodedBuf, Encoder, Http1Transaction, ParseContext, Wants};
use crate::body::DecodedLength;
use crate::common::{task, Pin, Poll, Unpin};
use crate::headers::connection_keep_alive;
use crate::proto::{BodyLength, DecodedLength, MessageHead};
use crate::Result;
use crate::proto::{BodyLength, MessageHead};
const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
@@ -589,7 +589,7 @@ where
self.state.writing = state;
}
pub fn end_body(&mut self) -> Result<()> {
pub fn end_body(&mut self) -> crate::Result<()> {
debug_assert!(self.can_write_body());
let mut res = Ok(());

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

@@ -5,10 +5,10 @@ use http::{Request, Response, StatusCode};
use tokio::io::{AsyncRead, AsyncWrite};
use super::{Http1Transaction, Wants};
use crate::body::{Body, HttpBody};
use crate::body::{Body, DecodedLength, HttpBody};
use crate::common::{task, Future, Never, Pin, Poll, Unpin};
use crate::proto::{
BodyLength, Conn, DecodedLength, Dispatched, MessageHead, RequestHead, RequestLine,
BodyLength, Conn, Dispatched, MessageHead, RequestHead, RequestLine,
ResponseHead,
};
use crate::service::HttpService;

3
src/proto/h1/io.rs
View File

@@ -2,13 +2,14 @@ use std::cell::Cell;
use std::cmp;
use std::fmt;
use std::io::{self, IoSlice};
use std::marker::Unpin;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use super::{Http1Transaction, ParseContext, ParsedMessage};
use crate::common::buf::BufList;
use crate::common::{task, Pin, Poll, Unpin};
use crate::common::{task, Pin, Poll};
/// The initial buffer size allocated before trying to read from IO.
pub(crate) const INIT_BUFFER_SIZE: usize = 8192;

4
src/proto/h1/mod.rs
View File

@@ -1,7 +1,8 @@
use bytes::BytesMut;
use http::{HeaderMap, Method};
use crate::proto::{BodyLength, DecodedLength, MessageHead};
use crate::body::DecodedLength;
use crate::proto::{BodyLength, MessageHead};
pub(crate) use self::conn::Conn;
pub use self::decode::Decoder;
@@ -11,7 +12,6 @@ pub use self::io::Cursor; //TODO: move out of h1::io
pub use self::io::MINIMUM_MAX_BUFFER_SIZE;
mod conn;
pub(super) mod date;
mod decode;
pub(crate) mod dispatch;
mod encode;

6
src/proto/h1/role.rs
View File

@@ -9,12 +9,14 @@ use bytes::BytesMut;
use http::header::{self, Entry, HeaderName, HeaderValue};
use http::{HeaderMap, Method, StatusCode, Version};
use crate::body::DecodedLength;
use crate::common::date;
use crate::error::Parse;
use crate::headers;
use crate::proto::h1::{
date, Encode, Encoder, Http1Transaction, ParseContext, ParseResult, ParsedMessage,
Encode, Encoder, Http1Transaction, ParseContext, ParseResult, ParsedMessage,
};
use crate::proto::{BodyLength, DecodedLength, MessageHead, RequestHead, RequestLine};
use crate::proto::{BodyLength, MessageHead, RequestHead, RequestLine};
const MAX_HEADERS: usize = 100;
const AVERAGE_HEADER_SIZE: usize = 30; // totally scientific

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

@@ -9,8 +9,7 @@ use pin_project::pin_project;
use std::error::Error as StdError;
use std::io::IoSlice;
use super::DecodedLength;
use crate::body::HttpBody;
use crate::body::{DecodedLength, HttpBody};
use crate::common::{task, Future, Pin, Poll};
use crate::headers::content_length_parse_all;

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

@@ -11,7 +11,7 @@ use tokio::io::{AsyncRead, AsyncWrite};
use super::{decode_content_length, ping, PipeToSendStream, SendBuf};
use crate::body::HttpBody;
use crate::common::exec::ConnStreamExec;
use crate::common::{task, Future, Pin, Poll};
use crate::common::{date, task, Future, Pin, Poll};
use crate::headers;
use crate::proto::Dispatched;
use crate::service::HttpService;
@@ -400,7 +400,7 @@ where
// set Date header if it isn't already set...
res.headers_mut()
.entry(::http::header::DATE)
.or_insert_with(crate::proto::h1::date::update_and_header_value);
.or_insert_with(date::update_and_header_value);
// automatically set Content-Length from body...
if let Some(len) = body.size_hint().exact() {

123
src/proto/mod.rs
View File

@@ -1,10 +1,11 @@
//! Pieces pertaining to the HTTP message protocol.
use http::{HeaderMap, Method, StatusCode, Uri, Version};
pub(crate) use self::body_length::DecodedLength;
pub(crate) use self::h1::{dispatch, Conn, ServerTransaction};
cfg_http1! {
pub(crate) mod h1;
pub(crate) use self::h1::{dispatch, Conn, ServerTransaction};
}
pub(crate) mod h1;
cfg_http2! {
pub(crate) mod h2;
}
@@ -13,23 +14,27 @@ cfg_http2! {
#[derive(Clone, Debug, Default, PartialEq)]
pub struct MessageHead<S> {
/// HTTP version of the message.
pub version: Version,
pub version: http::Version,
/// Subject (request line or status line) of Incoming message.
pub subject: S,
/// Headers of the Incoming message.
pub headers: HeaderMap,
pub headers: http::HeaderMap,
}
/// An incoming request message.
#[cfg(feature = "http1")]
pub type RequestHead = MessageHead<RequestLine>;
#[derive(Debug, Default, PartialEq)]
pub struct RequestLine(pub Method, pub Uri);
#[cfg(feature = "http1")]
pub struct RequestLine(pub http::Method, pub http::Uri);
/// An incoming response message.
pub type ResponseHead = MessageHead<StatusCode>;
#[cfg(feature = "http1")]
pub type ResponseHead = MessageHead<http::StatusCode>;
#[derive(Debug)]
#[cfg(feature = "http1")]
pub enum BodyLength {
/// Content-Length
Known(u64),
@@ -42,106 +47,6 @@ pub(crate) enum Dispatched {
/// Dispatcher completely shutdown connection.
Shutdown,
/// Dispatcher has pending upgrade, and so did not shutdown.
#[cfg(feature = "http1")]
Upgrade(crate::upgrade::Pending),
}
/// A separate module to encapsulate the invariants of the DecodedLength type.
mod body_length {
use std::fmt;
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) struct DecodedLength(u64);
const MAX_LEN: u64 = std::u64::MAX - 2;
impl DecodedLength {
pub(crate) const CLOSE_DELIMITED: DecodedLength = DecodedLength(::std::u64::MAX);
pub(crate) const CHUNKED: DecodedLength = DecodedLength(::std::u64::MAX - 1);
pub(crate) const ZERO: DecodedLength = DecodedLength(0);
#[cfg(test)]
pub(crate) fn new(len: u64) -> Self {
debug_assert!(len <= MAX_LEN);
DecodedLength(len)
}
/// Takes the length as a content-length without other checks.
///
/// Should only be called if previously confirmed this isn't
/// CLOSE_DELIMITED or CHUNKED.
#[inline]
pub(crate) fn danger_len(self) -> u64 {
debug_assert!(self.0 < Self::CHUNKED.0);
self.0
}
/// Converts to an Option<u64> representing a Known or Unknown length.
pub(crate) fn into_opt(self) -> Option<u64> {
match self {
DecodedLength::CHUNKED | DecodedLength::CLOSE_DELIMITED => None,
DecodedLength(known) => Some(known),
}
}
/// Checks the `u64` is within the maximum allowed for content-length.
pub(crate) fn checked_new(len: u64) -> Result<Self, crate::error::Parse> {
if len <= MAX_LEN {
Ok(DecodedLength(len))
} else {
warn!("content-length bigger than maximum: {} > {}", len, MAX_LEN);
Err(crate::error::Parse::TooLarge)
}
}
pub(crate) fn sub_if(&mut self, amt: u64) {
match *self {
DecodedLength::CHUNKED | DecodedLength::CLOSE_DELIMITED => (),
DecodedLength(ref mut known) => {
*known -= amt;
}
}
}
}
impl fmt::Debug for DecodedLength {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
DecodedLength::CLOSE_DELIMITED => f.write_str("CLOSE_DELIMITED"),
DecodedLength::CHUNKED => f.write_str("CHUNKED"),
DecodedLength(n) => f.debug_tuple("DecodedLength").field(&n).finish(),
}
}
}
impl fmt::Display for DecodedLength {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match *self {
DecodedLength::CLOSE_DELIMITED => f.write_str("close-delimited"),
DecodedLength::CHUNKED => f.write_str("chunked encoding"),
DecodedLength::ZERO => f.write_str("empty"),
DecodedLength(n) => write!(f, "content-length ({} bytes)", n),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sub_if_known() {
let mut len = DecodedLength::new(30);
len.sub_if(20);
assert_eq!(len.0, 10);
}
#[test]
fn sub_if_chunked() {
let mut len = DecodedLength::CHUNKED;
len.sub_if(20);
assert_eq!(len, DecodedLength::CHUNKED);
}
}
}

6
src/rt.rs
View File

@@ -5,4 +5,8 @@
//! If the `runtime` feature is disabled, the types in this module can be used
//! to plug in other runtimes.
pub use crate::common::Executor;
/// An executor of futures.
pub trait Executor<Fut> {
/// Place the future into the executor to be run.
fn execute(&self, fut: Fut);
}

128
src/server/conn.rs
View File

@@ -11,7 +11,7 @@
//! ## Example
//! A simple example that uses the `Http` struct to talk HTTP over a Tokio TCP stream
//! ```no_run
//! # #[cfg(feature = "runtime")]
//! # #[cfg(all(feature = "http1", feature = "runtime"))]
//! # mod rt {
//! use http::{Request, Response, StatusCode};
//! use hyper::{server::conn::Http, service::service_fn, Body};
@@ -28,7 +28,7 @@
//! tokio::task::spawn(async move {
//! if let Err(http_err) = Http::new()
//! .http1_only(true)
//! .keep_alive(true)
//! .http1_keep_alive(true)
//! .serve_connection(tcp_stream, service_fn(hello))
//! .await {
//! eprintln!("Error while serving HTTP connection: {}", http_err);
@@ -45,8 +45,8 @@
use std::error::Error as StdError;
use std::fmt;
#[cfg(feature = "http1")]
use std::marker::PhantomData;
use std::mem;
#[cfg(feature = "tcp")]
use std::net::SocketAddr;
#[cfg(feature = "runtime")]
@@ -63,10 +63,12 @@ use crate::common::exec::{ConnStreamExec, Exec, NewSvcExec};
#[cfg(feature = "http2")]
use crate::common::io::Rewind;
use crate::common::{task, Future, Pin, Poll, Unpin};
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
use crate::error::{Kind, Parse};
use crate::proto;
use crate::service::{HttpService, MakeServiceRef};
#[cfg(feature = "http1")]
use crate::upgrade::Upgraded;
use self::spawn_all::NewSvcTask;
@@ -100,11 +102,13 @@ pub struct Http<E = Exec> {
#[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(feature = "http1")]
#[cfg(feature = "http2")]
Fallback,
}
@@ -157,6 +161,7 @@ where
S: HttpService<Body>,
{
pub(super) conn: Option<ProtoServer<T, S::ResBody, S, E>>,
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
fallback: Fallback<E>,
}
@@ -167,6 +172,7 @@ where
S: HttpService<Body>,
B: HttpBody,
{
#[cfg(feature = "http1")]
H1(
#[pin]
proto::h1::Dispatcher<
@@ -181,6 +187,7 @@ where
H2(#[pin] proto::h2::Server<Rewind<T>, S, B, E>),
}
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
#[derive(Clone, Debug)]
enum Fallback<E> {
@@ -188,6 +195,7 @@ enum Fallback<E> {
Http1Only,
}
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
impl<E> Fallback<E> {
fn to_h2(&self) -> bool {
@@ -198,6 +206,7 @@ impl<E> Fallback<E> {
}
}
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
impl<E> Unpin for Fallback<E> {}
@@ -247,6 +256,8 @@ 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;
@@ -267,6 +278,8 @@ impl<E> Http<E> {
/// 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
@@ -275,18 +288,13 @@ impl<E> Http<E> {
/// 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
}
// renamed due different semantics of http2 keep alive
#[doc(hidden)]
#[deprecated(note = "renamed to `http1_keep_alive`")]
pub fn keep_alive(&mut self, val: bool) -> &mut Self {
self.http1_keep_alive(val)
}
/// Set whether HTTP/1 connections should try to use vectored writes,
/// or always flatten into a single buffer.
///
@@ -300,6 +308,8 @@ impl<E> Http<E> {
/// 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
@@ -314,7 +324,10 @@ impl<E> Http<E> {
if val {
self.mode = ConnectionMode::H2Only;
} else {
self.mode = ConnectionMode::Fallback;
#[cfg(feature = "http1")]
{
self.mode = ConnectionMode::Fallback;
}
}
self
}
@@ -446,6 +459,8 @@ impl<E> Http<E> {
/// # 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,
@@ -519,6 +534,7 @@ impl<E> Http<E> {
I: AsyncRead + AsyncWrite + Unpin,
E: ConnStreamExec<S::Future, Bd>,
{
#[cfg(feature = "http1")]
macro_rules! h1 {
() => {{
let mut conn = proto::Conn::new(io);
@@ -545,9 +561,11 @@ impl<E> Http<E> {
}
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 => {
@@ -560,6 +578,7 @@ impl<E> Http<E> {
Connection {
conn: Some(proto),
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
fallback: if self.mode == ConnectionMode::Fallback {
Fallback::ToHttp2(self.h2_builder.clone(), self.exec.clone())
@@ -610,6 +629,7 @@ where
/// nothing.
pub fn graceful_shutdown(self: Pin<&mut Self>) {
match self.project().conn {
#[cfg(feature = "http1")]
Some(ProtoServer::H1(ref mut h1, _)) => {
h1.disable_keep_alive();
}
@@ -640,6 +660,7 @@ where
/// 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 {
@@ -672,26 +693,26 @@ where
B: Unpin,
{
loop {
let polled = match *self.conn.as_mut().unwrap() {
ProtoServer::H1(ref mut h1, _) => h1.poll_without_shutdown(cx),
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(|_| ()),
};
match ready!(polled) {
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));
}
}
}
}
@@ -710,6 +731,7 @@ where
})
}
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
fn upgrade_h2(&mut self) {
trace!("Trying to upgrade connection to h2");
@@ -759,16 +781,21 @@ where
loop {
match ready!(Pin::new(self.conn.as_mut().unwrap()).poll(cx)) {
Ok(done) => {
if let proto::Dispatched::Upgrade(pending) = done {
// 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();
}
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() => {
@@ -796,18 +823,21 @@ where
// ===== impl ConnectionMode =====
impl ConnectionMode {}
impl Default for ConnectionMode {
#[cfg(feature = "http2")]
#[cfg(all(feature = "http1", feature = "http2"))]
fn default() -> ConnectionMode {
ConnectionMode::Fallback
}
#[cfg(not(feature = "http2"))]
#[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 Serve =====
@@ -955,6 +985,7 @@ 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),
#[cfg(feature = "http2")]
ProtoServerProj::H2(s) => s.poll(cx),
@@ -1137,17 +1168,22 @@ mod upgrades {
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)) => {
let h1 = match mem::replace(&mut self.inner.conn, None) {
Some(ProtoServer::H1(h1, _)) => h1,
_ => unreachable!("Upgrade expects h1"),
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")
}
};
let (io, buf, _) = h1.into_inner();
pending.fulfill(Upgraded::new(io, buf));
return Poll::Ready(Ok(()));
}
Err(e) => {
#[cfg(feature = "http1")]
#[cfg(feature = "http2")]
match *e.kind() {
Kind::Parse(Parse::VersionH2) if self.inner.fallback.to_h2() => {

13
src/server/mod.rs
View File

@@ -241,6 +241,8 @@ impl<I, E> Builder<I, E> {
/// Sets whether to use keep-alive for HTTP/1 connections.
///
/// Default is `true`.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_keepalive(mut self, val: bool) -> Self {
self.protocol.http1_keep_alive(val);
self
@@ -254,6 +256,8 @@ impl<I, E> Builder<I, E> {
/// 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) -> Self {
self.protocol.http1_half_close(val);
self
@@ -262,6 +266,8 @@ impl<I, E> Builder<I, E> {
/// Set the maximum buffer size.
///
/// Default is ~ 400kb.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_max_buf_size(mut self, val: usize) -> Self {
self.protocol.max_buf_size(val);
self
@@ -272,6 +278,7 @@ impl<I, E> Builder<I, E> {
// This isn't really desirable in most cases, only really being useful in
// silly pipeline benchmarks.
#[doc(hidden)]
#[cfg(feature = "http1")]
pub fn http1_pipeline_flush(mut self, val: bool) -> Self {
self.protocol.pipeline_flush(val);
self
@@ -290,7 +297,9 @@ impl<I, E> Builder<I, E> {
/// which may eliminate unnecessary cloning on some TLS backends
///
/// Default is `auto`. In this mode hyper will try to guess which
/// mode to use
/// mode to use.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_writev(mut self, val: bool) -> Self {
self.protocol.http1_writev(val);
self
@@ -299,6 +308,8 @@ impl<I, E> Builder<I, E> {
/// Sets whether HTTP/1 is required.
///
/// Default is `false`.
#[cfg(feature = "http1")]
#[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
pub fn http1_only(mut self, val: bool) -> Self {
self.protocol.http1_only(val);
self

3
src/service/mod.rs
View File

@@ -39,11 +39,14 @@ pub use tower_service::Service;
mod http;
mod make;
#[cfg(any(feature = "http1", feature = "http2"))]
mod oneshot;
mod util;
pub(crate) use self::http::HttpService;
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) use self::make::{MakeConnection, MakeServiceRef};
#[cfg(any(feature = "http1", feature = "http2"))]
pub(crate) use self::oneshot::{oneshot, Oneshot};
pub use self::make::make_service_fn;

5
src/upgrade.rs
View File

@@ -57,6 +57,7 @@ pub struct Parts<T> {
_inner: (),
}
#[cfg(feature = "http1")]
pub(crate) struct Pending {
tx: oneshot::Sender<crate::Result<Upgraded>>,
}
@@ -68,6 +69,7 @@ pub(crate) struct Pending {
#[derive(Debug)]
struct UpgradeExpected(());
#[cfg(feature = "http1")]
pub(crate) fn pending() -> (Pending, OnUpgrade) {
let (tx, rx) = oneshot::channel();
(Pending { tx }, OnUpgrade { rx: Some(rx) })
@@ -76,6 +78,7 @@ pub(crate) fn pending() -> (Pending, OnUpgrade) {
// ===== impl Upgraded =====
impl Upgraded {
#[cfg(any(feature = "http1", test))]
pub(crate) fn new<T>(io: T, read_buf: Bytes) -> Self
where
T: AsyncRead + AsyncWrite + Unpin + Send + 'static,
@@ -145,6 +148,7 @@ impl OnUpgrade {
OnUpgrade { rx: None }
}
#[cfg(feature = "http1")]
pub(crate) fn is_none(&self) -> bool {
self.rx.is_none()
}
@@ -175,6 +179,7 @@ impl fmt::Debug for OnUpgrade {
// ===== impl Pending =====
#[cfg(feature = "http1")]
impl Pending {
pub(crate) fn fulfill(self, upgraded: Upgraded) {
trace!("pending upgrade fulfill");

22
tests/server.rs
View File

@@ -1078,6 +1078,7 @@ async fn nonempty_parse_eof_returns_error() {
.expect_err("partial parse eof is error");
}
#[cfg(feature = "http1")]
#[tokio::test]
async fn http1_allow_half_close() {
let _ = pretty_env_logger::try_init();
@@ -1111,6 +1112,7 @@ async fn http1_allow_half_close() {
t1.join().expect("client thread");
}
#[cfg(feature = "http1")]
#[tokio::test]
async fn disconnect_after_reading_request_before_responding() {
let _ = pretty_env_logger::try_init();
@@ -1524,18 +1526,22 @@ async fn illegal_request_length_returns_400_response() {
.expect_err("illegal Content-Length should error");
}
#[cfg(feature = "http1")]
#[test]
#[should_panic]
fn max_buf_size_panic_too_small() {
const MAX: usize = 8191;
Http::new().max_buf_size(MAX);
}
#[cfg(feature = "http1")]
#[test]
fn max_buf_size_no_panic() {
const MAX: usize = 8193;
Http::new().max_buf_size(MAX);
}
#[cfg(feature = "http1")]
#[tokio::test]
async fn max_buf_size() {
let _ = pretty_env_logger::try_init();
@@ -2277,7 +2283,7 @@ impl ServeOptions {
fn serve(self) -> Serve {
let _ = pretty_env_logger::try_init();
let options = self;
let _options = self;
let (addr_tx, addr_rx) = mpsc::channel();
let (msg_tx, msg_rx) = mpsc::channel();
@@ -2306,11 +2312,15 @@ impl ServeOptions {
})
});
let server = Server::bind(&addr)
.http1_only(options.http1_only)
.http1_keepalive(options.keep_alive)
.http1_pipeline_flush(options.pipeline)
.serve(service);
let builder = Server::bind(&addr);
#[cfg(feature = "http1")]
let builder = builder
.http1_only(_options.http1_only)
.http1_keepalive(_options.keep_alive)
.http1_pipeline_flush(_options.pipeline);
let server = builder.serve(service);
addr_tx.send(server.local_addr()).expect("server addr tx");