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//! [`Http11ReadHeaders`] 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::trace;
65use thiserror::Error;
66use url::Url;
67
68use crate::{
69    coroutine::*,
70    http_try,
71    rfc1945::version::HTTP_10,
72    rfc9110::{
73        headers::{CONTENT_LENGTH, LOCATION, TRANSFER_ENCODING},
74        request::HttpRequest,
75        response::HttpResponse,
76        send::{HttpSendOutput, HttpSendYield},
77    },
78    rfc9112::{
79        chunk::{Http11ReadChunks, Http11ReadChunksError},
80        read_headers::{Http11ReadHeaders, Http11ReadHeadersError},
81    },
82};
83
84/// Failure causes during the HTTP/1.1 send flow.
85#[derive(Debug, Error)]
86pub enum Http11SendError {
87    #[error("HTTP/1.1 send failed: reached unexpected EOF")]
88    Eof,
89    #[error("HTTP/1.1 send failed: parse response headers: {0}")]
90    ParseResponseHeaders(HttparseError),
91    #[error("HTTP/1.1 send failed: invalid content length `{0}`")]
92    InvalidContentLength(String),
93    #[error("HTTP/1.1 send failed: {0}")]
94    ReadChunks(#[from] Http11ReadChunksError),
95}
96
97impl From<Http11ReadHeadersError> for Http11SendError {
98    fn from(err: Http11ReadHeadersError) -> Self {
99        match err {
100            Http11ReadHeadersError::Eof => Self::Eof,
101            Http11ReadHeadersError::ParseResponseHeaders(e) => Self::ParseResponseHeaders(e),
102        }
103    }
104}
105
106#[derive(Debug)]
107enum State {
108    ReadHeaders(Http11ReadHeaders),
109    BodyChunks(Http11ReadChunks),
110    BodyLength(usize),
111    BodyEof,
112}
113
114/// I/O-free coroutine to send an HTTP/1.1 request and receive its response.
115#[derive(Debug)]
116pub struct Http11Send {
117    request_url: Url,
118    state: State,
119    wants_write: Option<Vec<u8>>,
120    is_conn_closed: bool,
121    response: Option<HttpResponse>,
122    buf: Vec<u8>,
123}
124
125impl Http11Send {
126    /// Creates a new coroutine that will send the given request and
127    /// receive its response.
128    pub fn new(req: HttpRequest) -> Self {
129        trace!("prepare request to be sent: {req:?}");
130
131        let request_url = req.url.clone();
132        let bytes = req.to_http_11_vec();
133
134        Self {
135            request_url,
136            state: State::ReadHeaders(Http11ReadHeaders::default()),
137            wants_write: Some(bytes),
138            is_conn_closed: false,
139            response: None,
140            buf: Vec::new(),
141        }
142    }
143
144    fn finish(
145        &self,
146        response: HttpResponse,
147        remaining: Vec<u8>,
148    ) -> HttpCoroutineState<HttpSendYield, Result<HttpSendOutput, Http11SendError>> {
149        let keep_alive = !self.is_conn_closed;
150
151        if response.status.is_redirection() {
152            if let Some(location) = response.header(LOCATION) {
153                if let Ok(url) = self.request_url.join(location) {
154                    let same_scheme = self.request_url.scheme() == url.scheme();
155                    let same_host = self.request_url.host() == url.host()
156                        && self.request_url.port() == url.port();
157                    let same_origin = same_scheme && same_host;
158
159                    return HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
160                        url,
161                        response,
162                        keep_alive,
163                        same_origin,
164                    });
165                }
166            }
167        }
168
169        HttpCoroutineState::Complete(Ok(HttpSendOutput {
170            response,
171            remaining,
172            keep_alive,
173        }))
174    }
175}
176
177impl HttpCoroutine for Http11Send {
178    type Yield = HttpSendYield;
179    type Return = Result<HttpSendOutput, Http11SendError>;
180
181    fn resume(&mut self, mut arg: Option<&[u8]>) -> HttpCoroutineState<Self::Yield, Self::Return> {
182        loop {
183            if let Some(bytes) = self.wants_write.take() {
184                return HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes));
185            }
186
187            match &mut self.state {
188                State::ReadHeaders(rh) => match rh.resume(arg.take()) {
189                    HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
190                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
191                    }
192                    HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
193                        unreachable!("Http11ReadHeaders never writes");
194                    }
195                    HttpCoroutineState::Complete(Err(err)) => {
196                        return HttpCoroutineState::Complete(Err(err.into()));
197                    }
198                    HttpCoroutineState::Complete(Ok(out)) => {
199                        let mut response = out.response;
200                        let is_http10 = response.version == HTTP_10;
201                        self.is_conn_closed = !out.keep_alive;
202                        let status = *response.status;
203
204                        if status == 204 || status == 304 {
205                            return self.finish(response, out.remaining);
206                        }
207
208                        if !is_http10 {
209                            let chunked = response
210                                .header(TRANSFER_ENCODING)
211                                .is_some_and(|enc| enc.eq_ignore_ascii_case("chunked"));
212                            if chunked {
213                                let mut chunks = Http11ReadChunks::default();
214                                match chunks.resume(Some(&out.remaining)) {
215                                    HttpCoroutineState::Complete(Ok(chunk_out)) => {
216                                        response.body = chunk_out.body;
217                                        return self.finish(response, chunk_out.remaining);
218                                    }
219                                    HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
220                                        self.response = Some(response);
221                                        trace!("reading chunked body");
222                                        self.state = State::BodyChunks(chunks);
223                                        return HttpCoroutineState::Yielded(
224                                            HttpSendYield::WantsRead,
225                                        );
226                                    }
227                                    HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
228                                        unreachable!("Http11ReadChunks never writes");
229                                    }
230                                    HttpCoroutineState::Complete(Err(err)) => {
231                                        return HttpCoroutineState::Complete(Err(err.into()));
232                                    }
233                                }
234                            }
235                        }
236
237                        if let Some(len_str) = response.header(CONTENT_LENGTH) {
238                            let len_str = len_str.trim();
239                            let Ok(len) = len_str.parse::<usize>() else {
240                                let err = Http11SendError::InvalidContentLength(len_str.to_owned());
241                                return HttpCoroutineState::Complete(Err(err));
242                            };
243                            self.buf = out.remaining;
244                            self.response = Some(response);
245                            trace!("reading body of length {len}");
246                            self.state = State::BodyLength(len);
247                            continue;
248                        }
249
250                        self.buf = out.remaining;
251                        self.response = Some(response);
252                        trace!("reading body until connection closes");
253                        self.state = State::BodyEof;
254                    }
255                },
256                State::BodyChunks(chunks) => {
257                    let chunk_out = http_try!(chunks, arg.take());
258                    let mut response = self.response.take().expect("response missing");
259                    response.body = chunk_out.body;
260                    return self.finish(response, chunk_out.remaining);
261                }
262                State::BodyLength(len) => {
263                    if let Some(data) = arg.take() {
264                        self.buf.extend_from_slice(data);
265                    }
266
267                    if *len > self.buf.len() {
268                        trace!("received incomplete body {len}/{}", self.buf.len());
269                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
270                    }
271
272                    let body = self.buf.drain(..*len).collect();
273                    let remaining = mem::take(&mut self.buf);
274                    let mut response = self.response.take().expect("response missing");
275                    response.body = body;
276                    return self.finish(response, remaining);
277                }
278                State::BodyEof => match arg.take() {
279                    Some(&[]) => {
280                        let buf = mem::take(&mut self.buf);
281                        let mut response = self.response.take().expect("response missing");
282                        response.body = buf;
283                        return self.finish(response, Vec::new());
284                    }
285                    Some(data) => {
286                        self.buf.extend_from_slice(data);
287                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
288                    }
289                    None => {
290                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
291                    }
292                },
293            }
294        }
295    }
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301
302    #[test]
303    fn body_chunks_completes() {
304        let req = HttpRequest::get("https://example.com".try_into().unwrap());
305        let mut coroutine = Http11Send::new(req);
306
307        let bytes = expect_wants_write(&mut coroutine, None);
308        assert_eq!(
309            bytes,
310            b"GET / HTTP/1.1\r\nhost: example.com\r\ncontent-length: 0\r\n\r\n"
311        );
312
313        expect_wants_read(&mut coroutine, None);
314
315        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";
316        let out = expect_complete_ok(&mut coroutine, Some(reply));
317        assert_eq!(out.response.version, "HTTP/1.1");
318        assert_eq!(*out.response.status, 200);
319        assert_eq!(out.response.body, b"hello world");
320        assert!(out.remaining.is_empty());
321        assert!(out.keep_alive);
322    }
323
324    #[test]
325    fn body_length_completes() {
326        let req = HttpRequest::get("https://example.com".try_into().unwrap());
327        let mut coroutine = Http11Send::new(req);
328
329        expect_wants_write(&mut coroutine, None);
330        expect_wants_read(&mut coroutine, None);
331
332        let reply = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello";
333        let out = expect_complete_ok(&mut coroutine, Some(reply));
334        assert_eq!(*out.response.status, 200);
335        assert_eq!(out.response.body, b"hello");
336        assert!(out.keep_alive);
337    }
338
339    #[test]
340    fn body_eof_completes() {
341        let req = HttpRequest::get("https://example.com".try_into().unwrap());
342        let mut coroutine = Http11Send::new(req);
343
344        expect_wants_write(&mut coroutine, None);
345        expect_wants_read(&mut coroutine, None);
346        expect_wants_read(&mut coroutine, Some(b"HTTP/1.1 200 OK\r\n\r\nhello "));
347        expect_wants_read(&mut coroutine, Some(b"world"));
348
349        let out = expect_complete_ok(&mut coroutine, Some(b""));
350        assert_eq!(out.response.body, b"hello world");
351    }
352
353    #[test]
354    fn invalid_content_length_errors() {
355        let req = HttpRequest::get("https://example.com".try_into().unwrap());
356        let mut coroutine = Http11Send::new(req);
357
358        expect_wants_write(&mut coroutine, None);
359        expect_wants_read(&mut coroutine, None);
360
361        let reply = b"HTTP/1.1 200 OK\r\nContent-Length: notanumber\r\n\r\n";
362        let err = expect_complete_err(&mut coroutine, Some(reply));
363        let Http11SendError::InvalidContentLength(s) = err else {
364            panic!("expected InvalidContentLength, got {err:?}");
365        };
366        assert_eq!(s, "notanumber");
367    }
368
369    #[test]
370    fn redirect_yields_wants_redirect() {
371        let req = HttpRequest::get("http://example.com/old".try_into().unwrap());
372        let mut coroutine = Http11Send::new(req);
373
374        expect_wants_write(&mut coroutine, None);
375        expect_wants_read(&mut coroutine, None);
376
377        let reply =
378            b"HTTP/1.1 301 Moved Permanently\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n";
379        match coroutine.resume(Some(reply)) {
380            HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
381                url,
382                same_origin,
383                keep_alive,
384                ..
385            }) => {
386                assert_eq!(url.path(), "/new");
387                assert!(same_origin);
388                assert!(keep_alive);
389            }
390            state => panic!("expected WantsRedirect, got {state:?}"),
391        }
392    }
393
394    // --- utils
395
396    fn expect_wants_write(cor: &mut Http11Send, arg: Option<&[u8]>) -> Vec<u8> {
397        match cor.resume(arg) {
398            HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes)) => bytes,
399            state => panic!("expected WantsWrite, got {state:?}"),
400        }
401    }
402
403    fn expect_wants_read(cor: &mut Http11Send, arg: Option<&[u8]>) {
404        match cor.resume(arg) {
405            HttpCoroutineState::Yielded(HttpSendYield::WantsRead) => {}
406            state => panic!("expected WantsRead, got {state:?}"),
407        }
408    }
409
410    fn expect_complete_ok(cor: &mut Http11Send, arg: Option<&[u8]>) -> HttpSendOutput {
411        match cor.resume(arg) {
412            HttpCoroutineState::Complete(Ok(out)) => out,
413            state => panic!("expected Complete(Ok), got {state:?}"),
414        }
415    }
416
417    fn expect_complete_err(cor: &mut Http11Send, arg: Option<&[u8]>) -> Http11SendError {
418        match cor.resume(arg) {
419            HttpCoroutineState::Complete(Err(err)) => err,
420            state => panic!("expected Complete(Err), got {state:?}"),
421        }
422    }
423}
io_http/rfc9112/send.rs

io_http/rfc9112/
send.rs
