io_http/rfc9112/
send.rs

1//! I/O-free coroutine sending an HTTP/1.1 request and receiving its
2//! response ([RFC 9112]).
3//!
4//! Serialises the request, drives the read/parse cycle through
5//! [`Http11HeadersRead`] for the head and selects a body-reading
6//! strategy from the response headers:
7//!
8//! | Strategy     | Trigger                      |
9//! |--------------|------------------------------|
10//! | Chunked      | `Transfer-Encoding: chunked` |
11//! | Fixed-length | `Content-Length: <n>`        |
12//! | Read-to-EOF  | Neither header present       |
13//!
14//! # Example
15//!
16//! ```rust,no_run
17//! use std::{io::{Read, Write}, net::TcpStream};
18//!
19//! use io_http::{
20//!     coroutine::*,
21//!     rfc9110::{request::HttpRequest, send::HttpSendYield},
22//!     rfc9112::send::Http11Send,
23//! };
24//! use url::Url;
25//!
26//! let url = Url::parse("http://example.com/").unwrap();
27//! let request = HttpRequest::get(url.clone())
28//!     .header("Host", url.host_str().unwrap())
29//!     .header("Connection", "close");
30//!
31//! let mut stream = TcpStream::connect("example.com:80").unwrap();
32//! let mut send = Http11Send::new(request);
33//! let mut arg: Option<&[u8]> = None;
34//! let mut buf = [0u8; 4096];
35//!
36//! let (response, keep_alive) = loop {
37//!     match send.resume(arg.take()) {
38//!         HttpCoroutineState::Complete(Ok(out)) => break (out.response, out.keep_alive),
39//!         HttpCoroutineState::Complete(Err(err)) => panic!("{err}"),
40//!         HttpCoroutineState::Yielded(HttpSendYield::WantsRead) => {
41//!             let n = stream.read(&mut buf).unwrap();
42//!             arg = Some(&buf[..n]);
43//!         }
44//!         HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes)) => {
45//!             stream.write_all(&bytes).unwrap();
46//!         }
47//!         HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect { url, .. }) => {
48//!             panic!("redirect to {url}");
49//!         }
50//!     }
51//! };
52//!
53//! println!("{}", *response.status);
54//! # let _ = keep_alive;
55//! ```
56//!
57//! [RFC 9112]: https://www.rfc-editor.org/rfc/rfc9112
58
59use core::mem;
60
61use alloc::{borrow::ToOwned, string::String, vec::Vec};
62
63use httparse::Error as HttparseError;
64use log::{debug, trace};
65use thiserror::Error;
66use url::Url;
67
68use crate::{
69    coroutine::*,
70    http_try,
71    rfc1945::version::HTTP_10,
72    rfc9110::{
73        headers::{HTTP_CONTENT_LENGTH, HTTP_LOCATION, HTTP_TRANSFER_ENCODING},
74        request::HttpRequest,
75        response::HttpResponse,
76        send::{HttpSendOutput, HttpSendYield},
77    },
78    rfc9112::{
79        chunk::{Http11ChunksRead, Http11ChunksReadError},
80        read_headers::{Http11HeadersRead, Http11HeadersReadError},
81    },
82};
83
84/// Failure causes during the HTTP/1.1 send flow.
85#[derive(Debug, Error)]
86pub enum Http11SendError {
87    /// The stream reached EOF before the response was complete.
88    #[error("HTTP/1.1 send failed: reached unexpected EOF")]
89    Eof,
90    /// The response head could not be parsed.
91    #[error("HTTP/1.1 send failed: parse response headers: {0}")]
92    ParseResponseHeaders(HttparseError),
93    /// The content length header value is not a valid integer.
94    #[error("HTTP/1.1 send failed: invalid content length `{0}`")]
95    InvalidContentLength(String),
96    /// The chunked-body decoder failed.
97    #[error("HTTP/1.1 send failed: {0}")]
98    ReadChunks(#[from] Http11ChunksReadError),
99}
100
101impl From<Http11HeadersReadError> for Http11SendError {
102    fn from(err: Http11HeadersReadError) -> Self {
103        match err {
104            Http11HeadersReadError::Eof => Self::Eof,
105            Http11HeadersReadError::ParseResponseHeaders(e) => Self::ParseResponseHeaders(e),
106        }
107    }
108}
109
110#[derive(Debug)]
111enum State {
112    ReadHeaders(Http11HeadersRead),
113    BodyChunks(Http11ChunksRead),
114    BodyLength(usize),
115    BodyEof,
116}
117
118/// I/O-free coroutine to send an HTTP/1.1 request and receive its response.
119#[derive(Debug)]
120pub struct Http11Send {
121    request_url: Url,
122    state: State,
123    wants_write: Option<Vec<u8>>,
124    is_conn_closed: bool,
125    response: Option<HttpResponse>,
126    buf: Vec<u8>,
127}
128
129impl Http11Send {
130    /// Creates a new coroutine that will send the given request and
131    /// receive its response.
132    pub fn new(req: HttpRequest) -> Self {
133        debug!("prepare request to send");
134        trace!("{req:?}");
135
136        let request_url = req.url.clone();
137        let bytes = req.to_http_11_vec();
138
139        Self {
140            request_url,
141            state: State::ReadHeaders(Http11HeadersRead::default()),
142            wants_write: Some(bytes),
143            is_conn_closed: false,
144            response: None,
145            buf: Vec::new(),
146        }
147    }
148
149    fn finish(
150        &self,
151        response: HttpResponse,
152        remaining: Vec<u8>,
153    ) -> HttpCoroutineState<HttpSendYield, Result<HttpSendOutput, Http11SendError>> {
154        let keep_alive = !self.is_conn_closed;
155
156        if response.status.is_redirection() {
157            if let Some(location) = response.header(HTTP_LOCATION) {
158                if let Ok(url) = self.request_url.join(location) {
159                    let same_scheme = self.request_url.scheme() == url.scheme();
160                    let same_host = self.request_url.host() == url.host()
161                        && self.request_url.port() == url.port();
162                    let same_origin = same_scheme && same_host;
163
164                    return HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
165                        url,
166                        response,
167                        keep_alive,
168                        same_origin,
169                    });
170                }
171            }
172        }
173
174        HttpCoroutineState::Complete(Ok(HttpSendOutput {
175            response,
176            remaining,
177            keep_alive,
178        }))
179    }
180}
181
182impl HttpCoroutine for Http11Send {
183    type Yield = HttpSendYield;
184    type Return = Result<HttpSendOutput, Http11SendError>;
185
186    fn resume(&mut self, mut arg: Option<&[u8]>) -> HttpCoroutineState<Self::Yield, Self::Return> {
187        loop {
188            if let Some(bytes) = self.wants_write.take() {
189                return HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes));
190            }
191
192            match &mut self.state {
193                State::ReadHeaders(rh) => match rh.resume(arg.take()) {
194                    HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
195                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
196                    }
197                    HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
198                        unreachable!("Http11HeadersRead never writes");
199                    }
200                    HttpCoroutineState::Complete(Err(err)) => {
201                        return HttpCoroutineState::Complete(Err(err.into()));
202                    }
203                    HttpCoroutineState::Complete(Ok(out)) => {
204                        let mut response = out.response;
205                        let is_http10 = response.version == HTTP_10;
206                        self.is_conn_closed = !out.keep_alive;
207                        let status = *response.status;
208
209                        if status == 204 || status == 304 {
210                            return self.finish(response, out.remaining);
211                        }
212
213                        if !is_http10 {
214                            let chunked = response
215                                .header(HTTP_TRANSFER_ENCODING)
216                                .is_some_and(|enc| enc.eq_ignore_ascii_case("chunked"));
217                            if chunked {
218                                let mut chunks = Http11ChunksRead::default();
219                                match chunks.resume(Some(&out.remaining)) {
220                                    HttpCoroutineState::Complete(Ok(chunk_out)) => {
221                                        response.body = chunk_out.body;
222                                        return self.finish(response, chunk_out.remaining);
223                                    }
224                                    HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
225                                        self.response = Some(response);
226                                        debug!("reading chunked body");
227                                        self.state = State::BodyChunks(chunks);
228                                        return HttpCoroutineState::Yielded(
229                                            HttpSendYield::WantsRead,
230                                        );
231                                    }
232                                    HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
233                                        unreachable!("Http11ChunksRead never writes");
234                                    }
235                                    HttpCoroutineState::Complete(Err(err)) => {
236                                        return HttpCoroutineState::Complete(Err(err.into()));
237                                    }
238                                }
239                            }
240                        }
241
242                        if let Some(len_str) = response.header(HTTP_CONTENT_LENGTH) {
243                            let len_str = len_str.trim();
244                            let Ok(len) = len_str.parse::<usize>() else {
245                                let err = Http11SendError::InvalidContentLength(len_str.to_owned());
246                                return HttpCoroutineState::Complete(Err(err));
247                            };
248                            self.buf = out.remaining;
249                            self.response = Some(response);
250                            debug!("reading body of length {len}");
251                            self.state = State::BodyLength(len);
252                            continue;
253                        }
254
255                        self.buf = out.remaining;
256                        self.response = Some(response);
257                        debug!("reading body until connection closes");
258                        self.state = State::BodyEof;
259                    }
260                },
261                State::BodyChunks(chunks) => {
262                    let chunk_out = http_try!(chunks, arg.take());
263                    let mut response = self.response.take().expect("response missing");
264                    response.body = chunk_out.body;
265                    return self.finish(response, chunk_out.remaining);
266                }
267                State::BodyLength(len) => {
268                    if let Some(data) = arg.take() {
269                        self.buf.extend_from_slice(data);
270                    }
271
272                    if *len > self.buf.len() {
273                        trace!("received incomplete body {len}/{}", self.buf.len());
274                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
275                    }
276
277                    let body = self.buf.drain(..*len).collect();
278                    let remaining = mem::take(&mut self.buf);
279                    let mut response = self.response.take().expect("response missing");
280                    response.body = body;
281                    return self.finish(response, remaining);
282                }
283                State::BodyEof => match arg.take() {
284                    Some(&[]) => {
285                        let buf = mem::take(&mut self.buf);
286                        let mut response = self.response.take().expect("response missing");
287                        response.body = buf;
288                        return self.finish(response, Vec::new());
289                    }
290                    Some(data) => {
291                        self.buf.extend_from_slice(data);
292                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
293                    }
294                    None => {
295                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
296                    }
297                },
298            }
299        }
300    }
301}
302
303#[cfg(test)]
304mod tests {
305    use crate::{coroutine::*, rfc9112::send::*};
306
307    #[test]
308    fn body_chunks_completes() {
309        let req = HttpRequest::get("https://example.com".try_into().unwrap());
310        let mut coroutine = Http11Send::new(req);
311
312        let bytes = expect_wants_write(&mut coroutine, None);
313        assert_eq!(
314            bytes,
315            b"GET / HTTP/1.1\r\nhost: example.com\r\ncontent-length: 0\r\n\r\n"
316        );
317
318        expect_wants_read(&mut coroutine, None);
319
320        let reply = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n";
321        let out = expect_complete_ok(&mut coroutine, Some(reply));
322        assert_eq!(out.response.version, "HTTP/1.1");
323        assert_eq!(*out.response.status, 200);
324        assert_eq!(out.response.body, b"hello world");
325        assert!(out.remaining.is_empty());
326        assert!(out.keep_alive);
327    }
328
329    #[test]
330    fn body_length_completes() {
331        let req = HttpRequest::get("https://example.com".try_into().unwrap());
332        let mut coroutine = Http11Send::new(req);
333
334        expect_wants_write(&mut coroutine, None);
335        expect_wants_read(&mut coroutine, None);
336
337        let reply = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello";
338        let out = expect_complete_ok(&mut coroutine, Some(reply));
339        assert_eq!(*out.response.status, 200);
340        assert_eq!(out.response.body, b"hello");
341        assert!(out.keep_alive);
342    }
343
344    #[test]
345    fn body_eof_completes() {
346        let req = HttpRequest::get("https://example.com".try_into().unwrap());
347        let mut coroutine = Http11Send::new(req);
348
349        expect_wants_write(&mut coroutine, None);
350        expect_wants_read(&mut coroutine, None);
351        expect_wants_read(&mut coroutine, Some(b"HTTP/1.1 200 OK\r\n\r\nhello "));
352        expect_wants_read(&mut coroutine, Some(b"world"));
353
354        let out = expect_complete_ok(&mut coroutine, Some(b""));
355        assert_eq!(out.response.body, b"hello world");
356    }
357
358    #[test]
359    fn invalid_content_length_errors() {
360        let req = HttpRequest::get("https://example.com".try_into().unwrap());
361        let mut coroutine = Http11Send::new(req);
362
363        expect_wants_write(&mut coroutine, None);
364        expect_wants_read(&mut coroutine, None);
365
366        let reply = b"HTTP/1.1 200 OK\r\nContent-Length: notanumber\r\n\r\n";
367        let err = expect_complete_err(&mut coroutine, Some(reply));
368        let Http11SendError::InvalidContentLength(s) = err else {
369            panic!("expected InvalidContentLength, got {err:?}");
370        };
371        assert_eq!(s, "notanumber");
372    }
373
374    #[test]
375    fn redirect_yields_wants_redirect() {
376        let req = HttpRequest::get("http://example.com/old".try_into().unwrap());
377        let mut coroutine = Http11Send::new(req);
378
379        expect_wants_write(&mut coroutine, None);
380        expect_wants_read(&mut coroutine, None);
381
382        let reply =
383            b"HTTP/1.1 301 Moved Permanently\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n";
384        match coroutine.resume(Some(reply)) {
385            HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
386                url,
387                same_origin,
388                keep_alive,
389                ..
390            }) => {
391                assert_eq!(url.path(), "/new");
392                assert!(same_origin);
393                assert!(keep_alive);
394            }
395            state => panic!("expected WantsRedirect, got {state:?}"),
396        }
397    }
398
399    fn expect_wants_write(cor: &mut Http11Send, arg: Option<&[u8]>) -> Vec<u8> {
400        match cor.resume(arg) {
401            HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes)) => bytes,
402            state => panic!("expected WantsWrite, got {state:?}"),
403        }
404    }
405
406    fn expect_wants_read(cor: &mut Http11Send, arg: Option<&[u8]>) {
407        match cor.resume(arg) {
408            HttpCoroutineState::Yielded(HttpSendYield::WantsRead) => {}
409            state => panic!("expected WantsRead, got {state:?}"),
410        }
411    }
412
413    fn expect_complete_ok(cor: &mut Http11Send, arg: Option<&[u8]>) -> HttpSendOutput {
414        match cor.resume(arg) {
415            HttpCoroutineState::Complete(Ok(out)) => out,
416            state => panic!("expected Complete(Ok), got {state:?}"),
417        }
418    }
419
420    fn expect_complete_err(cor: &mut Http11Send, arg: Option<&[u8]>) -> Http11SendError {
421        match cor.resume(arg) {
422            HttpCoroutineState::Complete(Err(err)) => err,
423            state => panic!("expected Complete(Err), got {state:?}"),
424        }
425    }
426}
io_http/rfc9112/send.rs

io_http/rfc9112/
send.rs
