Skip to content
This repository was archived by the owner on Sep 8, 2025. It is now read-only.

Commit feb9d10

Browse files
rvolosatovsrvolosatovs
committed
feat(p3/http): handle content-length
Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>
1 parent beed238 commit feb9d10

16 files changed

Lines changed: 622 additions & 370 deletions

File tree

‎crates/test-programs/src/bin/http_0_3_outbound_request_content_length.rs‎

Lines changed: 84 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
1+
use anyhow::Context as _;
2+
use futures::join;
3+
use test_programs::p3::wasi::http::handler;
14
use test_programs::p3::wasi::http::types::{ErrorCode, Headers, Method, Request, Scheme, Trailers};
25
use test_programs::p3::{wit_future, wit_stream};
6+
use wit_bindgen::FutureReader;
37
use wit_bindgen_rt::async_support::{FutureWriter, StreamWriter};
48

59
struct Component;
@@ -10,10 +14,11 @@ fn make_request() -> (
1014
Request,
1115
StreamWriter<u8>,
1216
FutureWriter<Result<Option<Trailers>, ErrorCode>>,
17+
FutureReader<Result<(), ErrorCode>>,
1318
) {
1419
let (contents_tx, contents_rx) = wit_stream::new();
1520
let (trailers_tx, trailers_rx) = wit_future::new();
16-
let (request, _) = Request::new(
21+
let (request, transmit) = Request::new(
1722
Headers::from_list(&[("Content-Length".to_string(), b"11".to_vec())]).unwrap(),
1823
Some(contents_rx),
1924
trailers_rx,
@@ -35,72 +40,95 @@ fn make_request() -> (
3540
.set_path_with_query(Some("/"))
3641
.expect("setting path with query");
3742

38-
(request, contents_tx, trailers_tx)
43+
(request, contents_tx, trailers_tx, transmit)
3944
}
4045

4146
impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
4247
async fn run() -> Result<(), ()> {
4348
{
44-
println!("writing enough");
45-
let (_, mut contents_tx, trailers_tx) = make_request();
46-
let remaining = contents_tx.write_all(b"long enough".to_vec()).await;
47-
assert!(remaining.is_empty());
48-
drop(contents_tx);
49-
trailers_tx.write(Ok(None)).await.unwrap();
49+
let (request, mut contents_tx, trailers_tx, transmit) = make_request();
50+
let (transmit, handle) = join!(async { transmit.await }, async {
51+
let res = handler::handle(request)
52+
.await
53+
.context("failed to send request")?;
54+
println!("writing enough");
55+
let remaining = contents_tx.write_all(b"long enough".to_vec()).await;
56+
assert!(
57+
remaining.is_empty(),
58+
"{}",
59+
String::from_utf8_lossy(&remaining)
60+
);
61+
drop(contents_tx);
62+
trailers_tx
63+
.write(Ok(None))
64+
.await
65+
.context("failed to finish body")?;
66+
anyhow::Ok(res)
67+
});
68+
let res = handle.unwrap();
69+
drop(res);
70+
transmit
71+
.expect("transmit sender dropped")
72+
.expect("failed to transmit request");
5073
}
5174

5275
{
53-
println!("writing too little");
54-
let (_, mut contents_tx, trailers_tx) = make_request();
55-
let remaining = contents_tx.write_all(b"msg".to_vec()).await;
56-
assert!(remaining.is_empty());
57-
drop(contents_tx);
58-
trailers_tx.write(Ok(None)).await.unwrap();
59-
60-
// handle()
61-
62-
// TODO: Figure out how/if to represent this in wasip3
63-
//let e = OutgoingBody::finish(outgoing_body, None)
64-
// .expect_err("finish should fail");
65-
66-
//assert!(
67-
// matches!(&e, ErrorCode::HttpRequestBodySize(Some(3))),
68-
// "unexpected error: {e:#?}"
69-
//);
76+
let (request, mut contents_tx, trailers_tx, transmit) = make_request();
77+
let (transmit, handle) = join!(async { transmit.await }, async {
78+
let res = handler::handle(request)
79+
.await
80+
.context("failed to send request")?;
81+
println!("writing too little");
82+
let remaining = contents_tx.write_all(b"msg".to_vec()).await;
83+
assert!(
84+
remaining.is_empty(),
85+
"{}",
86+
String::from_utf8_lossy(&remaining)
87+
);
88+
drop(contents_tx);
89+
trailers_tx
90+
.write(Ok(None))
91+
.await
92+
.context("failed to finish body")?;
93+
anyhow::Ok(res)
94+
});
95+
let res = handle.unwrap();
96+
drop(res);
97+
let err = transmit
98+
.expect("transmit sender dropped")
99+
.expect_err("request transmission should have failed");
100+
assert!(
101+
matches!(err, ErrorCode::HttpRequestBodySize(Some(3))),
102+
"unexpected error: {err:#?}"
103+
);
70104
}
71105

72106
{
73-
println!("writing too much");
74-
let (_, mut contents_tx, trailers_tx) = make_request();
75-
let remaining = contents_tx.write_all(b"more than 11 bytes".to_vec()).await;
76-
assert!(remaining.is_empty());
77-
drop(contents_tx);
78-
trailers_tx.write(Ok(None)).await.unwrap();
79-
80-
// TODO: Figure out how/if to represent this in wasip3
81-
//let e = request_body
82-
// .blocking_write_and_flush("more than 11 bytes".as_bytes())
83-
// .expect_err("write should fail");
84-
//let e = match e {
85-
// test_programs::wasi::io::streams::StreamError::LastOperationFailed(e) => {
86-
// http_error_code(&e)
87-
// }
88-
// test_programs::wasi::io::streams::StreamError::Closed => panic!("request closed"),
89-
//};
90-
//assert!(
91-
// matches!(
92-
// e,
93-
// Some(ErrorCode::HttpRequestBodySize(Some(18)))
94-
// ),
95-
// "unexpected error {e:?}"
96-
//);
97-
//let e = OutgoingBody::finish(outgoing_body, None)
98-
// .expect_err("finish should fail");
99-
100-
//assert!(
101-
// matches!(&e, ErrorCode::HttpRequestBodySize(Some(18))),
102-
// "unexpected error: {e:#?}"
103-
//);
107+
let (request, mut contents_tx, trailers_tx, transmit) = make_request();
108+
let (transmit, handle) = join!(async { transmit.await }, async {
109+
let res = handler::handle(request)
110+
.await
111+
.context("failed to send request")?;
112+
println!("writing too much");
113+
let remaining = contents_tx.write_all(b"more than 11 bytes".to_vec()).await;
114+
assert!(
115+
remaining.is_empty(),
116+
"{}",
117+
String::from_utf8_lossy(&remaining)
118+
);
119+
drop(contents_tx);
120+
_ = trailers_tx.write(Ok(None)).await;
121+
anyhow::Ok(res)
122+
});
123+
let res = handle.unwrap();
124+
drop(res);
125+
let err = transmit
126+
.expect("transmit sender dropped")
127+
.expect_err("request transmission should have failed");
128+
assert!(
129+
matches!(err, ErrorCode::HttpRequestBodySize(Some(18))),
130+
"unexpected error: {err:#?}"
131+
);
104132
}
105133
Ok(())
106134
}

‎crates/test-programs/src/bin/http_0_3_outbound_request_post.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ test_programs::p3::export!(Component);
77

88
impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
99
async fn run() -> Result<(), ()> {
10+
const BODY: &[u8] = b"{\"foo\": \"bar\"}";
1011
let addr = test_programs::p3::wasi::cli::environment::get_environment()
1112
.into_iter()
1213
.find_map(|(k, v)| k.eq("HTTP_SERVER").then_some(v))
@@ -16,7 +17,7 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
1617
Scheme::Http,
1718
&addr,
1819
"/post",
19-
Some(b"{\"foo\": \"bar\"}"),
20+
Some(BODY),
2021
None,
2122
None,
2223
None,
@@ -32,7 +33,7 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
3233
assert_eq!(std::str::from_utf8(method).unwrap(), "POST");
3334
let uri = res.header("x-wasmtime-test-uri").unwrap();
3435
assert_eq!(std::str::from_utf8(uri).unwrap(), format!("/post"));
35-
assert_eq!(res.body, b"{\"foo\": \"bar\"}", "invalid body returned");
36+
assert_eq!(res.body, BODY, "invalid body returned");
3637
Ok(())
3738
}
3839
}

‎crates/test-programs/src/p3/http.rs‎

Lines changed: 34 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use anyhow::{anyhow, Context as _, Result};
22
use core::fmt;
3-
use futures::try_join;
3+
use futures::join;
44

55
use crate::p3::wasi::http::{handler, types};
66
use crate::p3::{wit_future, wit_stream};
@@ -88,42 +88,47 @@ pub async fn request(
8888
.set_path_with_query(Some(&path_with_query))
8989
.map_err(|()| anyhow!("failed to set path_with_query"))?;
9090

91-
let ((), (), response) = try_join!(
92-
async {
93-
if let Some(buf) = body {
94-
let remaining = contents_tx.write_all(buf.into()).await;
95-
assert!(remaining.is_empty());
96-
}
97-
drop(contents_tx);
98-
trailers_tx.write(Ok(None)).await.unwrap();
99-
anyhow::Ok(())
100-
},
91+
let (transmit, handle) = join!(
10192
async {
10293
transmit
10394
.await
104-
.expect("transmit sender dropped")
105-
.context("failed to transmit request")?;
106-
Ok(())
95+
.context("transmit sender dropped")?
96+
.context("failed to transmit request")
10797
},
10898
async {
10999
let response = handler::handle(request).await?;
110100
let status = response.status_code();
111101
let headers = response.headers().entries();
112-
113-
let (body, trailers) = response.body().expect("failed to get response body");
114-
let body = body.collect().await;
115-
let trailers = trailers
116-
.await
117-
.expect("trailers sender dropped")
118-
.context("failed to read body")?;
119-
let trailers = trailers.map(|trailers| trailers.entries());
120-
Ok(Response {
121-
status,
122-
headers,
123-
body,
124-
trailers,
125-
})
102+
let (body_rx, trailers_rx) = response.body().expect("failed to get response body");
103+
let ((), rx) = join!(
104+
async {
105+
if let Some(buf) = body {
106+
let remaining = contents_tx.write_all(buf.into()).await;
107+
assert!(remaining.is_empty());
108+
}
109+
drop(contents_tx);
110+
// This can fail in HTTP/1.1, since the connection might already be closed
111+
_ = trailers_tx.write(Ok(None)).await;
112+
},
113+
async {
114+
let body = body_rx.collect().await;
115+
let trailers = trailers_rx
116+
.await
117+
.context("trailers sender dropped")?
118+
.context("failed to read body")?;
119+
let trailers = trailers.map(|trailers| trailers.entries());
120+
anyhow::Ok(Response {
121+
status,
122+
headers,
123+
body,
124+
trailers,
125+
})
126+
}
127+
);
128+
rx
126129
},
127-
)?;
130+
);
131+
let response = handle?;
132+
transmit?;
128133
Ok(response)
129134
}

‎crates/wasi-http/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ workspace = true
1515
anyhow = { workspace = true }
1616
async-trait = { workspace = true }
1717
bytes = { workspace = true }
18-
futures = { workspace = true, default-features = false }
18+
futures = { workspace = true, default-features = false, features = ["async-await"] }
1919
hyper = { workspace = true, features = ["full"] }
2020
tokio = { workspace = true, features = [
2121
"net",

0 commit comments

Comments
 (0)