http server: more complex streaming example

pchickey · pchickey · commit 85c2ab71d44b · 2025-10-16T16:20:36.000-07:00
diff --git a/examples/http_server.rs b/examples/http_server.rs
@@ -1,5 +1,5 @@
 use anyhow::{Context, Result};
-use futures_lite::stream::once_future;
+use futures_lite::stream::{once_future, unfold};
 use http_body_util::{BodyExt, StreamBody};
 use std::convert::Infallible;
 use wstd::http::body::{Body, Bytes, Frame};
@@ -14,6 +14,7 @@ async fn main(request: Request<Body>) -> Result<Response<Body>, Error> {
         "/" => http_home(request).await,
         "/wait-response" => http_wait_response(request).await,
         "/wait-body" => http_wait_body(request).await,
+        "/stream-body" => http_stream_body(request).await,
         "/echo" => http_echo(request).await,
         "/echo-headers" => http_echo_headers(request).await,
         "/echo-trailers" => http_echo_trailers(request).await,
@@ -62,6 +63,30 @@ async fn http_wait_body(_request: Request<Body>) -> Result<Response<Body>> {
     Ok(Response::new(Body::from_try_stream(once_future(body))))
 }
 
+async fn http_stream_body(_request: Request<Body>) -> Result<Response<Body>> {
+    // Get the time now
+    let start = Instant::now();
+
+    let body = move |iters: usize| async move {
+        if iters == 0 {
+            return None;
+        }
+        // Sleep for 0.1 second.
+        wstd::task::sleep(Duration::from_millis(100)).await;
+
+        // Compute how long we slept for.
+        let elapsed = Instant::now().duration_since(start).as_millis();
+        Some((
+            Ok::<_, Infallible>(Bytes::from(format!(
+                "stream started {elapsed} millis ago\n"
+            ))),
+            iters - 1,
+        ))
+    };
+
+    Ok(Response::new(Body::from_try_stream(unfold(5, body))))
+}
+
 async fn http_echo(request: Request<Body>) -> Result<Response<Body>> {
     let (_parts, body) = request.into_parts();
     Ok(Response::new(body))
diff --git a/src/http/body.rs b/src/http/body.rs
@@ -40,8 +40,19 @@ pub mod util {
 ///   `Stream` of `Into<Bytes>`
 ///
 /// Consume this HTTP body using:
-///
-///
+/// * `Body::into_boxed_body` converts it to an `UnsyncBoxBody<Bytes, Error>`.
+///    This is a boxed representation of `http_body::Body` that is `Send` but not
+/// `  Sync`. The Unsync variant is required for compatibility with the `axum`
+///   crate.
+/// * `async fn Body::contents(&mut self) -> Result<&[u8], Error>` is ready
+///   when all contents of the body have been collected, and gives them as a
+///   byte slice.
+/// * `async fn Body::str_contents(&mut self) -> Result<&str, Error>` is ready
+///   when all contents of the body have been collected, and gives them as a str
+///   slice.
+/// * `async fn Body::json(&mut self) -> Result<T, Error>` gathers body
+///   contents and then uses `T: serde::Deserialize` to deserialize to json
+///   (requires feature `json`).
 #[derive(Debug)]
 pub struct Body(BodyInner);
 
diff --git a/test-programs/artifacts/tests/http_server.rs b/test-programs/artifacts/tests/http_server.rs
@@ -81,6 +81,31 @@ fn http_server() -> Result<()> {
     );
     assert!(duration >= Duration::from_secs(1));
 
+    // TEST /stream-body http_stream_body
+    // Sends response status and headers, then unfolds 5 iterations of a
+    // stream that sleeps for 100ms and then prints the time since stream
+    // started.
+    // With ureq we can't tell that the response status and headers were sent
+    // with a delay in the body. Additionally, the implementation MAY buffer up the
+    // entire response and body before sending it, though wasmtime does not.
+    let start = Instant::now();
+    let body: String = ureq::get("http://127.0.0.1:8081/stream-body")
+        .call()?
+        .into_string()?;
+    let duration = start.elapsed();
+    assert_eq!(body.lines().count(), 5, "body has 5 lines");
+    for (iter, line) in body.lines().enumerate() {
+        let sleep_report = line
+            .split(' ')
+            .find_map(|s| s.parse::<usize>().ok())
+            .expect("body should print 'stream started Nxx millis ago'");
+        assert!(
+            sleep_report >= (iter * 100),
+            "should have slept for {iter} * 100 or more millis, got {sleep_report}"
+        );
+    }
+    assert!(duration >= Duration::from_millis(500));
+
     // TEST /echo htto_echo
     // Send a request body, see that we got the same back in response body.
     const MESSAGE: &[u8] = b"hello, echoserver!\n";
