feat(http1): support expect: 100-continue on the client

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<bool>,
     h1_title_case_headers: bool,
     h1_preserve_header_case: bool,
     h1_max_headers: Option<usize>,
+    h1_continue_timeout: Dur,
     #[cfg(feature = "ffi")]
     h1_preserve_header_order: bool,
     h1_read_buf_exact_size: Option<usize>,
@@ -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<M>(&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<Option<Duration>>) -> &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();

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"))]

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<Duration>),
@@ -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<Box<dyn Sleep>> {
         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<Box<dyn Sleep>>, 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<Duration> {
         match dur {
             Dur::Default(Some(dur)) => match self {

src/headers.rs

@@ -42,6 +42,13 @@ 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")
+    })
+}
+
 #[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")

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;
@@ -234,7 +245,10 @@ where
             }
         }
 
-        let msg = match self.io.parse::<T>(
+        #[cfg(feature = "client")]
+        let mut seen_continue = false;
+
+        let parse_result = self.io.parse::<T>(
             cx,
             ParseContext {
                 cached_headers: &mut self.state.cached_headers,
@@ -247,8 +261,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 +552,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 +603,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 +616,63 @@ where
         self.io.has_buffered_write()
     }
 
+    #[cfg(feature = "client")]
+    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,
+        };
+
+        // 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<T::Outgoing>, body: Option<BodyLength>) {
+        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 {
@@ -941,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")]
@@ -953,6 +1029,12 @@ struct State {
     /// received.
     #[cfg(feature = "client")]
     on_informational: Option<crate::ext::OnInformational>,
+    #[cfg(feature = "client")]
+    h1_continue_timeout: Option<Duration>,
+    #[cfg(feature = "client")]
+    h1_continue_timeout_fut: Option<Pin<Box<dyn Sleep>>>,
+    #[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,
@@ -979,6 +1061,7 @@ enum Reading {
 
 enum Writing {
     Init,
+    Continue(Encoder),
     Body(Encoder),
     KeepAlive,
     Closed,
@@ -1011,6 +1094,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"),

src/proto/h1/dispatch.rs

@@ -390,6 +390,15 @@ 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() {
+                            if self.conn.poll_continue_timeout(cx).is_pending() {
+                                return Poll::Pending;
+                            }
+
+                            continue;
+                        }
+
                         trace!(
                             "no more write body allowed, user body is_end_stream = {}",
                             body.is_end_stream(),