From ca1d6d49bbf6a5a3086ec50c9a59d74c60299357 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dominykas=20Mak=C5=ABnas?= Date: Thu, 3 Sep 2026 21:38:53 +0300 Subject: [PATCH 1/2] feat(http1): support expect: 100-continue on the client --- src/headers.rs | 6 ++++ src/proto/h1/conn.rs | 44 +++++++++++++++++++++--- src/proto/h1/dispatch.rs | 4 +++ src/proto/h1/io.rs | 4 +++ src/proto/h1/mod.rs | 2 ++ src/proto/h1/role.rs | 46 +++++++++++++++++++++++++ tests/client.rs | 74 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 175 insertions(+), 5 deletions(-) diff --git a/src/headers.rs b/src/headers.rs index caace71f2a..db084fa153 100644 --- a/src/headers.rs +++ b/src/headers.rs @@ -42,6 +42,12 @@ fn connection_has(value: &HeaderValue, needle: &str) -> bool { false } +pub(super) fn expect_continue(headers: &http::HeaderMap) -> bool { + headers.get(http::header::EXPECT).map_or(false, |value| { + value.as_bytes().eq_ignore_ascii_case(b"100-continue") + }) +} + #[cfg(feature = "http1")] pub(super) fn te_is_trailers(headers: &http::HeaderMap) -> bool { header_value_list_has(headers.get_all(http::header::TE).into_iter(), "trailers") diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs index 593d3d9eda..99406a19df 100644 --- a/src/proto/h1/conn.rs +++ b/src/proto/h1/conn.rs @@ -234,7 +234,10 @@ where } } - let msg = match self.io.parse::( + #[cfg(feature = "client")] + let mut seen_continue = false; + + let parse_result = self.io.parse::( cx, ParseContext { cached_headers: &mut self.state.cached_headers, @@ -247,8 +250,17 @@ where h09_responses: self.state.h09_responses, #[cfg(feature = "client")] on_informational: &mut self.state.on_informational, + #[cfg(feature = "client")] + seen_continue: &mut seen_continue, }, - ) { + ); + + #[cfg(feature = "client")] + if seen_continue { + self.release_continue(); + } + + let msg = match parse_result { Poll::Ready(Ok(msg)) => msg, Poll::Ready(Err(e)) => return self.on_read_head_error(e), Poll::Pending => { @@ -529,7 +541,7 @@ where match self.state.writing { Writing::Body(..) => return, - Writing::Init | Writing::KeepAlive | Writing::Closed => (), + Writing::Init | Writing::Continue(..) | Writing::KeepAlive | Writing::Closed => (), } if !self.io.is_read_blocked() { @@ -580,7 +592,7 @@ where pub(crate) fn can_write_body(&self) -> bool { match self.state.writing { Writing::Body(..) => true, - Writing::Init | Writing::KeepAlive | Writing::Closed => false, + Writing::Init | Writing::Continue(..) | Writing::KeepAlive | Writing::Closed => false, } } @@ -593,10 +605,30 @@ where self.io.has_buffered_write() } + pub(crate) fn is_awaiting_continue(&self) -> bool { + matches!(self.state.writing, Writing::Continue(..)) + } + + #[cfg(feature = "client")] + fn release_continue(&mut self) { + self.state.writing = match std::mem::replace(&mut self.state.writing, Writing::Init) { + Writing::Continue(enc) => Writing::Body(enc), + other => other, + }; + } + pub(crate) fn write_head(&mut self, head: MessageHead, body: Option) { + let expect_continue = !T::should_read_first() + && head.version.gt(&Version::HTTP_10) + && headers::expect_continue(&head.headers); + if let Some(encoder) = self.encode_head(head, body) { self.state.writing = if !encoder.is_eof() { - Writing::Body(encoder) + if expect_continue { + Writing::Continue(encoder) + } else { + Writing::Body(encoder) + } } else if encoder.is_last() { Writing::Closed } else { @@ -979,6 +1011,7 @@ enum Reading { enum Writing { Init, + Continue(Encoder), Body(Encoder), KeepAlive, Closed, @@ -1011,6 +1044,7 @@ impl fmt::Debug for Writing { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Writing::Init => f.write_str("Init"), + Writing::Continue(enc) => f.debug_tuple("Continue").field(enc).finish(), Writing::Body(enc) => f.debug_tuple("Body").field(enc).finish(), Writing::KeepAlive => f.write_str("KeepAlive"), Writing::Closed => f.write_str("Closed"), diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs index d98aebcf1f..e98c63181c 100644 --- a/src/proto/h1/dispatch.rs +++ b/src/proto/h1/dispatch.rs @@ -390,6 +390,10 @@ where { debug_assert!(!*clear_body, "opt guard defaults to keeping body"); if !self.conn.can_write_body() { + if self.conn.is_awaiting_continue() { + return Poll::Pending; + } + trace!( "no more write body allowed, user body is_end_stream = {}", body.is_end_stream(), diff --git a/src/proto/h1/io.rs b/src/proto/h1/io.rs index b49e48e5c8..916435b039 100644 --- a/src/proto/h1/io.rs +++ b/src/proto/h1/io.rs @@ -194,6 +194,8 @@ where h09_responses: parse_ctx.h09_responses, #[cfg(feature = "client")] on_informational: parse_ctx.on_informational, + #[cfg(feature = "client")] + seen_continue: parse_ctx.seen_continue, }, )? { debug!("parsed {} headers", msg.head.headers.len()); @@ -708,6 +710,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; assert!(buffered .parse::(cx, parse_ctx) diff --git a/src/proto/h1/mod.rs b/src/proto/h1/mod.rs index a17dbae83c..ae7862632c 100644 --- a/src/proto/h1/mod.rs +++ b/src/proto/h1/mod.rs @@ -79,6 +79,8 @@ pub(crate) struct ParseContext<'ctx> { h09_responses: bool, #[cfg(feature = "client")] on_informational: &'ctx mut Option, + #[cfg(feature = "client")] + seen_continue: &'ctx mut bool, } /// Passed to `Http1Transaction::encode`. diff --git a/src/proto/h1/role.rs b/src/proto/h1/role.rs index d083d2a912..d2fa2dfdd4 100644 --- a/src/proto/h1/role.rs +++ b/src/proto/h1/role.rs @@ -1182,6 +1182,10 @@ impl Http1Transaction for Client { } if head.subject.is_informational() { + if head.subject.as_u16() == 100 { + *ctx.seen_continue = true; + } + if let Some(callback) = ctx.on_informational { callback.call(head.into_response(())); } @@ -1698,6 +1702,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .unwrap() @@ -1726,6 +1732,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; let msg = Client::parse(&mut raw, ctx).unwrap().unwrap(); assert_eq!(raw.len(), 0); @@ -1750,6 +1758,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; Server::parse(&mut raw, ctx).unwrap_err(); } @@ -1771,6 +1781,8 @@ mod tests { h09_responses: true, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; let msg = Client::parse(&mut raw, ctx).unwrap().unwrap(); assert_eq!(raw, H09_RESPONSE); @@ -1794,6 +1806,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; Client::parse(&mut raw, ctx).unwrap_err(); assert_eq!(raw, H09_RESPONSE); @@ -1821,6 +1835,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; let msg = Client::parse(&mut raw, ctx).unwrap().unwrap(); assert_eq!(raw.len(), 0); @@ -1845,6 +1861,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; Client::parse(&mut raw, ctx).unwrap_err(); } @@ -1873,6 +1891,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; let msg = Server::parse(&mut raw, ctx).unwrap().unwrap(); assert_eq!(raw.len(), 0); @@ -1900,6 +1920,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; Server::parse(&mut raw, ctx).unwrap_err(); } @@ -1920,6 +1942,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }; let parsed_message = Server::parse(&mut raw, ctx).unwrap().unwrap(); let orig_headers = parsed_message @@ -1959,6 +1983,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .expect("parse ok") @@ -1980,6 +2006,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .expect_err(comment) @@ -2220,6 +2248,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, } ) .expect("parse ok") @@ -2241,6 +2271,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .expect("parse ok") @@ -2262,6 +2294,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .expect_err("parse should err") @@ -2832,6 +2866,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .expect("parse ok") @@ -2876,6 +2912,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ); if should_success { @@ -2900,6 +2938,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ); if should_success { @@ -3020,6 +3060,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .expect("parse ok") @@ -3103,6 +3145,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .unwrap() @@ -3148,6 +3192,8 @@ mod tests { h09_responses: false, #[cfg(feature = "client")] on_informational: &mut None, + #[cfg(feature = "client")] + seen_continue: &mut false, }, ) .unwrap() diff --git a/tests/client.rs b/tests/client.rs index b512260cc5..1912ecc461 100644 --- a/tests/client.rs +++ b/tests/client.rs @@ -3315,6 +3315,80 @@ mod conn { drop(tx_a); let _ = tokio::time::timeout(Duration::from_secs(5), a_handle).await; } + + #[tokio::test] + async fn body_after_100() { + let io = tokio_test::io::Builder::new() + .write(b"POST /a HTTP/1.1\r\nexpect: 100-continue\r\ncontent-length: 5\r\n\r\n") + .read(b"HTTP/1.1 100 Continue\r\n\r\n") + .write(b"hello") + .read(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") + .build(); + + let (mut client, conn) = conn::http1::handshake(TokioIo::new(io)).await.unwrap(); + + tokio::spawn(async move { + let _ = conn.await; + }); + + let req = Request::builder() + .method(Method::POST) + .uri("/a") + .header("expect", "100-continue") + .body(Full::new(Bytes::from("hello"))) + .unwrap(); + + let res = client.send_request(req).await.expect("send_request"); + assert_eq!(res.status(), 200); + } + + #[tokio::test] + async fn final_response_before_100() { + let io = tokio_test::io::Builder::new() + .write(b"POST /a HTTP/1.1\r\nexpect: 100-continue\r\ncontent-length: 5\r\n\r\n") + .read(b"HTTP/1.1 417 Expectation Failed\r\ncontent-length: 0\r\n\r\n") + .build(); + + let (mut client, conn) = conn::http1::handshake(TokioIo::new(io)).await.unwrap(); + + tokio::spawn(async move { + let _ = conn.await; + }); + + let req = Request::builder() + .method(Method::POST) + .uri("/a") + .header("expect", "100-continue") + .body(Full::new(Bytes::from("hello"))) + .unwrap(); + + let res = client.send_request(req).await.expect("send_request"); + assert_eq!(res.status(), 417); + } + + #[tokio::test] + async fn expect_ignored_with_empty_body() { + let io = tokio_test::io::Builder::new() + .write(b"POST /a HTTP/1.1\r\nexpect: 100-continue\r\n\r\n") + .read(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") + .build(); + + let (mut client, conn) = conn::http1::handshake(TokioIo::new(io)).await.unwrap(); + + tokio::spawn(async move { + let _ = conn.await; + }); + + let req = Request::builder() + .method(Method::POST) + .uri("/a") + .header("expect", "100-continue") + .body(Empty::::new()) + .unwrap(); + + let res = client.send_request(req).await.expect("send_request"); + assert_eq!(res.status(), 200); + } } trait FutureHyperExt: TryFuture { From e3c9de1a278aa7cafd594dbd6e799eba813a905a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dominykas=20Mak=C5=ABnas?= Date: Thu, 3 Sep 2026 21:39:12 +0300 Subject: [PATCH 2/2] feat(http1): add a timeout for the client's 100-continue wait --- src/client/conn/http1.rs | 41 ++++++++++++++++++++++++- src/common/mod.rs | 6 ++-- src/common/time.rs | 13 ++++---- src/headers.rs | 1 + src/proto/h1/conn.rs | 64 +++++++++++++++++++++++++++++++++++----- src/proto/h1/dispatch.rs | 7 ++++- tests/client.rs | 30 +++++++++++++++++++ 7 files changed, 144 insertions(+), 18 deletions(-) diff --git a/src/client/conn/http1.rs b/src/client/conn/http1.rs index bdb03171d4..9d5e0d6861 100644 --- a/src/client/conn/http1.rs +++ b/src/client/conn/http1.rs @@ -4,9 +4,12 @@ use std::error::Error as StdError; use std::fmt; use std::future::Future; use std::pin::Pin; +use std::sync::Arc; use std::task::{Context, Poll}; +use std::time::Duration; -use crate::rt::{Read, Write}; +use crate::common::time::{Dur, Time}; +use crate::rt::{Read, Timer, Write}; use bytes::Bytes; use futures_core::ready; use http::{Request, Response}; @@ -127,10 +130,12 @@ where pub struct Builder { h09_responses: bool, h1_parser_config: ParserConfig, + timer: Time, h1_writev: Option, h1_title_case_headers: bool, h1_preserve_header_case: bool, h1_max_headers: Option, + h1_continue_timeout: Dur, #[cfg(feature = "ffi")] h1_preserve_header_order: bool, h1_read_buf_exact_size: Option, @@ -349,9 +354,11 @@ impl Builder { h1_writev: None, h1_read_buf_exact_size: None, h1_parser_config: ParserConfig::default(), + timer: Time::Empty, h1_title_case_headers: false, h1_preserve_header_case: false, h1_max_headers: None, + h1_continue_timeout: Dur::Default(Some(Duration::from_secs(30))), #[cfg(feature = "ffi")] h1_preserve_header_order: false, h1_max_buf_size: None, @@ -544,6 +551,31 @@ impl Builder { self } + /// Set the timer used in background tasks. + pub fn timer(&mut self, timer: M) -> &mut Self + where + M: Timer + Send + Sync + 'static, + { + self.timer = Time::Timer(Arc::new(timer)); + self + } + + /// Set how long the client waits for a `100 Continue` response before + /// sending the request body anyway. + /// + /// Only applies to requests with an `Expect: 100-continue` header and a body. + /// + /// Requires a [`Timer`] set by [`Builder::timer`] to take effect. Panics if + /// `expect_100_timeout` is configured without a [`Timer`]. + /// + /// Pass `None` to disable. + /// + /// Default is 30 seconds. + pub fn expect_100_timeout(&mut self, timeout: impl Into>) -> &mut Self { + self.h1_continue_timeout = Dur::Configured(timeout.into()); + self + } + /// Constructs a connection with the configured options and IO. /// See [`client::conn`](crate::client::conn) for more. /// @@ -571,6 +603,13 @@ impl Builder { let (tx, rx) = dispatch::channel(); let mut conn = proto::Conn::new(io); conn.set_h1_parser_config(opts.h1_parser_config); + conn.set_timer(opts.timer.clone()); + if let Some(dur) = opts + .timer + .check(opts.h1_continue_timeout, "continue_timeout") + { + conn.set_http1_continue_timeout(dur); + } if let Some(writev) = opts.h1_writev { if writev { conn.set_write_strategy_queue(); diff --git a/src/common/mod.rs b/src/common/mod.rs index 5be740b000..317bdea11b 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -13,9 +13,9 @@ pub(crate) mod io; pub(crate) mod lock; #[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))] pub(crate) mod task; -#[cfg(any( - all(feature = "server", feature = "http1"), - all(any(feature = "client", feature = "server"), feature = "http2"), +#[cfg(all( + any(feature = "client", feature = "server"), + any(feature = "http1", feature = "http2") ))] pub(crate) mod time; #[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))] diff --git a/src/common/time.rs b/src/common/time.rs index b3534f1580..fa803e6bf3 100644 --- a/src/common/time.rs +++ b/src/common/time.rs @@ -1,6 +1,6 @@ -#[cfg(any( - all(any(feature = "client", feature = "server"), feature = "http2"), - all(feature = "server", feature = "http1"), +#[cfg(all( + any(feature = "client", feature = "server"), + any(feature = "http1", feature = "http2") ))] use std::time::Duration; use std::{fmt, sync::Arc}; @@ -16,7 +16,7 @@ pub(crate) enum Time { Empty, } -#[cfg(all(feature = "server", feature = "http1"))] +#[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))] #[derive(Clone, Copy, Debug)] pub(crate) enum Dur { Default(Option), @@ -40,7 +40,7 @@ impl Time { } } - #[cfg(all(feature = "server", feature = "http1"))] + #[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))] pub(crate) fn sleep_until(&self, deadline: Instant) -> Pin> { match &self { Time::Empty => { @@ -57,6 +57,7 @@ impl Time { } } + #[cfg(any(all(feature = "server", feature = "http1"), feature = "http2"))] pub(crate) fn reset(&self, sleep: &mut Pin>, new_deadline: Instant) { match &self { Time::Empty => { @@ -66,7 +67,7 @@ impl Time { } } - #[cfg(all(feature = "server", feature = "http1"))] + #[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))] pub(crate) fn check(&self, dur: Dur, name: &'static str) -> Option { match dur { Dur::Default(Some(dur)) => match self { diff --git a/src/headers.rs b/src/headers.rs index db084fa153..8a5fd12038 100644 --- a/src/headers.rs +++ b/src/headers.rs @@ -42,6 +42,7 @@ fn connection_has(value: &HeaderValue, needle: &str) -> bool { false } +#[cfg(feature = "http1")] pub(super) fn expect_continue(headers: &http::HeaderMap) -> bool { headers.get(http::header::EXPECT).map_or(false, |value| { value.as_bytes().eq_ignore_ascii_case(b"100-continue") diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs index 99406a19df..45bec8d15f 100644 --- a/src/proto/h1/conn.rs +++ b/src/proto/h1/conn.rs @@ -1,11 +1,11 @@ use std::fmt; -#[cfg(feature = "server")] +#[cfg(any(feature = "server", feature = "client"))] use std::future::Future; use std::io; use std::marker::{PhantomData, Unpin}; use std::pin::Pin; use std::task::{Context, Poll}; -#[cfg(feature = "server")] +#[cfg(any(feature = "server", feature = "client"))] use std::time::Duration; use crate::rt::{Read, Write}; @@ -19,11 +19,11 @@ use httparse::ParserConfig; use super::io::Buffered; use super::{Decoder, Encode, EncodedBuf, Encoder, Http1Transaction, ParseContext, Wants}; use crate::body::DecodedLength; -#[cfg(feature = "server")] +#[cfg(any(feature = "server", feature = "client"))] use crate::common::time::Time; use crate::headers; use crate::proto::{BodyLength, MessageHead}; -#[cfg(feature = "server")] +#[cfg(any(feature = "server", feature = "client"))] use crate::rt::Sleep; const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; @@ -66,7 +66,7 @@ where h1_header_read_timeout_running: false, #[cfg(feature = "server")] date_header: true, - #[cfg(feature = "server")] + #[cfg(any(feature = "server", feature = "client"))] timer: Time::Empty, preserve_header_case: false, #[cfg(feature = "ffi")] @@ -75,6 +75,12 @@ where h09_responses: false, #[cfg(feature = "client")] on_informational: None, + #[cfg(feature = "client")] + h1_continue_timeout: None, + #[cfg(feature = "client")] + h1_continue_timeout_fut: None, + #[cfg(feature = "client")] + h1_continue_timeout_running: false, notify_read: false, reading: Reading::Init, writing: Writing::Init, @@ -88,7 +94,7 @@ where } } - #[cfg(feature = "server")] + #[cfg(any(feature = "server", feature = "client"))] pub(crate) fn set_timer(&mut self, timer: Time) { self.state.timer = timer; } @@ -146,6 +152,11 @@ where self.state.h1_header_read_timeout = Some(val); } + #[cfg(feature = "client")] + pub(crate) fn set_http1_continue_timeout(&mut self, val: Duration) { + self.state.h1_continue_timeout = Some(val); + } + #[cfg(feature = "server")] pub(crate) fn set_allow_half_close(&mut self) { self.state.allow_half_close = true; @@ -605,6 +616,7 @@ where self.io.has_buffered_write() } + #[cfg(feature = "client")] pub(crate) fn is_awaiting_continue(&self) -> bool { matches!(self.state.writing, Writing::Continue(..)) } @@ -615,6 +627,38 @@ where Writing::Continue(enc) => Writing::Body(enc), other => other, }; + + // Reset so a reused connection re-arms instead of firing a stale deadline. + self.state.h1_continue_timeout_running = false; + self.state.h1_continue_timeout_fut = None; + } + + #[cfg(feature = "client")] + pub(crate) fn poll_continue_timeout(&mut self, cx: &mut Context<'_>) -> Poll<()> { + if !matches!(self.state.writing, Writing::Continue(..)) { + return Poll::Pending; + } + + let timeout = match self.state.h1_continue_timeout { + Some(t) => t, + None => return Poll::Pending, + }; + + if !self.state.h1_continue_timeout_running { + let deadline = self.state.timer.now() + timeout; + self.state.h1_continue_timeout_running = true; + self.state.h1_continue_timeout_fut = Some(self.state.timer.sleep_until(deadline)); + } + + if let Some(fut) = &mut self.state.h1_continue_timeout_fut { + if Pin::new(fut).poll(cx).is_ready() { + trace!("expect-continue timeout elapsed; sending body anyway"); + self.release_continue(); + return Poll::Ready(()); + } + } + + Poll::Pending } pub(crate) fn write_head(&mut self, head: MessageHead, body: Option) { @@ -973,7 +1017,7 @@ struct State { h1_header_read_timeout_running: bool, #[cfg(feature = "server")] date_header: bool, - #[cfg(feature = "server")] + #[cfg(any(feature = "server", feature = "client"))] timer: Time, preserve_header_case: bool, #[cfg(feature = "ffi")] @@ -985,6 +1029,12 @@ struct State { /// received. #[cfg(feature = "client")] on_informational: Option, + #[cfg(feature = "client")] + h1_continue_timeout: Option, + #[cfg(feature = "client")] + h1_continue_timeout_fut: Option>>, + #[cfg(feature = "client")] + h1_continue_timeout_running: bool, /// Set to true when the Dispatcher should poll read operations /// again. See the `maybe_notify` method for more. notify_read: bool, diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs index e98c63181c..3edf306704 100644 --- a/src/proto/h1/dispatch.rs +++ b/src/proto/h1/dispatch.rs @@ -390,8 +390,13 @@ where { debug_assert!(!*clear_body, "opt guard defaults to keeping body"); if !self.conn.can_write_body() { + #[cfg(feature = "client")] if self.conn.is_awaiting_continue() { - return Poll::Pending; + if self.conn.poll_continue_timeout(cx).is_pending() { + return Poll::Pending; + } + + continue; } trace!( diff --git a/tests/client.rs b/tests/client.rs index 1912ecc461..a1c5fb7f2a 100644 --- a/tests/client.rs +++ b/tests/client.rs @@ -3389,6 +3389,36 @@ mod conn { let res = client.send_request(req).await.expect("send_request"); assert_eq!(res.status(), 200); } + + #[tokio::test] + async fn timeout_send_body() { + let io = tokio_test::io::Builder::new() + .write(b"POST /a HTTP/1.1\r\nexpect: 100-continue\r\ncontent-length: 5\r\n\r\n") + .write(b"hello") + .read(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") + .build(); + + let (mut client, conn) = conn::http1::Builder::new() + .timer(TokioTimer::new()) + .expect_100_timeout(Duration::from_millis(50)) + .handshake::<_, Full>(TokioIo::new(io)) + .await + .unwrap(); + + tokio::spawn(async move { + let _ = conn.await; + }); + + let req = Request::builder() + .method(Method::POST) + .uri("/a") + .header("expect", "100-continue") + .body(Full::new(Bytes::from("hello"))) + .unwrap(); + + let res = client.send_request(req).await.expect("send_request"); + assert_eq!(res.status(), 200); + } } trait FutureHyperExt: TryFuture {