From 6bb3c5d52c27fc1c605bed34987ebf1ac90bede4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dominykas=20Mak=C5=ABnas?= Date: Sat, 12 Sep 2026 10:14:37 +0300 Subject: [PATCH] feat(client): add Expect: 100-continue body wrapper --- Cargo.toml | 4 + examples/expect_continue.rs | 59 +++++++++++ src/client/expect_continue.rs | 189 ++++++++++++++++++++++++++++++++++ src/client/mod.rs | 3 + 4 files changed, 255 insertions(+) create mode 100644 examples/expect_continue.rs create mode 100644 src/client/expect_continue.rs diff --git a/Cargo.toml b/Cargo.toml index fad41db0..0b3fe5e7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,3 +108,7 @@ required-features = ["server", "http1", "tokio"] [[example]] name = "server_graceful" required-features = ["tokio", "server-graceful", "server-auto"] + +[[example]] +name = "expect_continue" +required-features = ["client", "server", "http1", "tokio"] diff --git a/examples/expect_continue.rs b/examples/expect_continue.rs new file mode 100644 index 00000000..033ab30e --- /dev/null +++ b/examples/expect_continue.rs @@ -0,0 +1,59 @@ +//! This example demonstrates request -> response flow of the `Expect: 100-continue` header + +use std::{convert::Infallible, error::Error}; + +use bytes::Bytes; +use http::{Method, Request, Response}; +use http_body_util::{BodyExt, Full}; +use hyper::{client::conn::http1::handshake, server::conn::http1::Builder, service::service_fn}; +use hyper_util::{client::expect_continue::wrap, rt::TokioIo}; +use tokio::net::{TcpListener, TcpStream}; + +async fn echo(req: Request) -> Result>, Infallible> { + let body = req.into_body().collect().await.unwrap().to_bytes(); + + println!( + "server read {} body bytes {:?}", + body.len(), + String::from_utf8_lossy(&body) + ); + + Ok(Response::new(Full::new(body))) +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Box> { + let addr = "127.0.0.1:3000"; + + let listener = TcpListener::bind(addr).await?; + tokio::spawn(async move { + let (tcp, _) = listener.accept().await.unwrap(); + let io = TokioIo::new(tcp); + + if let Err(e) = Builder::new().serve_connection(io, service_fn(echo)).await { + eprintln!("server error {e:?}"); + } + }); + + let stream = TcpStream::connect(addr).await?; + let io = TokioIo::new(stream); + let (mut sender, conn) = handshake(io).await?; + tokio::spawn(async move { + if let Err(e) = conn.await { + eprintln!("client connection error: {e:?}"); + } + }); + + let req = Request::builder() + .method(Method::POST) + .uri("/") + .header(hyper::header::HOST, addr) + .body(Full::new(Bytes::from("hi")))?; + + let resp = sender.send_request(wrap(req, None)).await?; + + println!("{:?} {:?}", resp.version(), resp.status()); + println!("{:#?}", resp.headers()); + + Ok(()) +} diff --git a/src/client/expect_continue.rs b/src/client/expect_continue.rs new file mode 100644 index 00000000..3f05d03f --- /dev/null +++ b/src/client/expect_continue.rs @@ -0,0 +1,189 @@ +//! Client side `Expect: 100-Continue` support +//! +//! This module contains the `ExpectContinueBody` request body wrapper +//! and the `wrap` convenience helper for building it. + +use std::{ + pin::Pin, + task::{Context, Poll}, +}; + +use futures_channel::oneshot; +use http::{HeaderValue, Request, StatusCode, header}; +use http_body::{Body, Frame}; +use hyper::rt::Sleep; +use pin_project_lite::pin_project; + +pin_project! { + /// ExpectContinueBody is a request body wrapper that withholds + /// its data until the client is cleared to send. + /// + /// The body is cleared to send when a `100-continue` is received + /// (delivered via hyper's `on_informational` hook) or the optional timeout elapses. + /// + /// HTTP/1.1 only. Use the `wrap` helper for building this conveniently. + pub struct ExpectContinueBody { + #[pin] + inner: B, + signal: Option>, + sleep: Option>>, + released: bool, + } +} + +impl ExpectContinueBody { + pub(crate) fn new( + inner: B, + signal: oneshot::Receiver<()>, + sleep: Option>>, + ) -> Self { + Self { + inner, + signal: Some(signal), + sleep, + released: false, + } + } +} + +impl Body for ExpectContinueBody { + type Data = B::Data; + type Error = B::Error; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, B::Error>>> { + let this = self.project(); + if !*this.released { + if let Some(rx) = this.signal.as_mut() { + if Pin::new(rx).poll(cx).is_ready() { + *this.released = true; + } + } + + if !*this.released { + if let Some(sleep) = this.sleep.as_mut() { + if sleep.as_mut().poll(cx).is_ready() { + *this.released = true; + } + } + } + + if !*this.released { + return Poll::Pending; + } + } + + this.inner.poll_frame(cx) + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> http_body::SizeHint { + self.inner.size_hint() + } +} + +/// wrap returns a request whose body is withheld until 100/timeout +/// +/// It adds `Expect: 100-continue` header to the request if absent. +/// Wraps the body in `ExpectContinueBody`. +/// Registers an `on_informational` hook that releases the body on a 100. +pub fn wrap( + req: Request, + sleep: Option>>, +) -> Request> { + let (tx, rx) = oneshot::channel(); + + let (mut parts, body) = req.into_parts(); + parts + .headers + .entry(header::EXPECT) + .or_insert(HeaderValue::from_static("100-continue")); + + let mut req = Request::from_parts(parts, ExpectContinueBody::new(body, rx, sleep)); + + let tx = std::sync::Mutex::new(Some(tx)); + hyper::ext::on_informational(&mut req, move |res| { + if res.status() == StatusCode::CONTINUE { + if let Some(tx) = tx.lock().unwrap().take() { + let _ = tx.send(()); + } + } + }); + + req +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use bytes::Bytes; + use futures_util::future::poll_fn; + use http_body_util::{BodyExt, Full}; + + use super::*; + + use crate::rt::TokioTimer; + use hyper::rt::Timer; + + #[tokio::test] + async fn withholds_body_until_signalled() { + let (tx, rx) = oneshot::channel::<()>(); + let mut body = std::pin::pin!(ExpectContinueBody::new( + Full::new(Bytes::from("hi")), + rx, + None, + )); + + let first = poll_fn(|cx| Poll::Ready(body.as_mut().poll_frame(cx))).await; + assert!(first.is_pending()); + + tx.send(()).unwrap(); + + let frame = body.frame().await.unwrap().unwrap(); + assert_eq!(frame.into_data().unwrap(), Bytes::from("hi")); + } + + #[tokio::test(start_paused = true)] + async fn release_body_with_timeout() { + let (_tx, rx) = oneshot::channel::<()>(); + + let sleep = TokioTimer.sleep(Duration::from_millis(100)); + let mut body = std::pin::pin!(ExpectContinueBody::new( + Full::new(Bytes::from("hi")), + rx, + Some(sleep), + )); + + let first = poll_fn(|cx| Poll::Ready(body.as_mut().poll_frame(cx))).await; + assert!(first.is_pending()); + + tokio::time::advance(Duration::from_millis(100)).await; + + let frame = body.frame().await.unwrap().unwrap(); + assert_eq!(frame.into_data().unwrap(), Bytes::from("hi")); + } + + #[tokio::test] + async fn release_on_signal_cancel() { + let (tx, rx) = oneshot::channel::<()>(); + let mut body = std::pin::pin!(ExpectContinueBody::new( + Full::new(Bytes::from("hi")), + rx, + None, + )); + + let first = poll_fn(|cx| Poll::Ready(body.as_mut().poll_frame(cx))).await; + assert!(first.is_pending()); + + drop(tx); + + let frame = body.frame().await.unwrap().unwrap(); + assert_eq!(frame.into_data().unwrap(), Bytes::from("hi")); + } +} diff --git a/src/client/mod.rs b/src/client/mod.rs index 9e3ee139..584795c2 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -10,6 +10,9 @@ pub mod pool; #[cfg(feature = "client-proxy")] pub mod proxy; +#[cfg(all(feature = "client", feature = "http1"))] +pub mod expect_continue; + #[cfg(any(feature = "client-legacy", feature = "client-proxy"))] fn strip_ipv6_brackets(host: &str) -> &str { host.strip_prefix('[')