async-h1 2.3.4

Asynchronous HTTP 1.1 parser.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

use std::io::Write;
use std::pin::Pin;
use std::task::{Context, Poll};

use futures_lite::io::{self, AsyncRead as Read, Cursor};
use http_types::headers::{CONTENT_LENGTH, HOST, TRANSFER_ENCODING};
use http_types::{Method, Request};

use crate::body_encoder::BodyEncoder;
use crate::read_to_end;
use crate::EncoderState;

/// An HTTP encoder.
#[doc(hidden)]
#[derive(Debug)]
pub struct Encoder {
    request: Request,
    state: EncoderState,
}

impl Encoder {
    /// build a new client encoder
    pub fn new(request: Request) -> Self {
        Self {
            request,
            state: EncoderState::Start,
        }
    }

    fn finalize_headers(&mut self) -> io::Result<()> {
        if self.request.header(HOST).is_none() {
            let url = self.request.url();
            let host = url
                .host_str()
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Missing hostname"))?
                .to_owned();

            if let Some(port) = url.port() {
                self.request
                    .insert_header(HOST, format!("{}:{}", host, port));
            } else {
                self.request.insert_header(HOST, host);
            };
        }

        // Insert Proxy-Connection header when method is CONNECT
        if self.request.method() == Method::Connect {
            self.request.insert_header("proxy-connection", "keep-alive");
        }

        // If the body isn't streaming, we can set the content-length ahead of time. Else we need to
        // send all items in chunks.
        if let Some(len) = self.request.len() {
            self.request.insert_header(CONTENT_LENGTH, len.to_string());
        } else {
            self.request.insert_header(TRANSFER_ENCODING, "chunked");
        }

        Ok(())
    }

    fn compute_head(&mut self) -> io::Result<Cursor<Vec<u8>>> {
        let mut buf = Vec::with_capacity(128);
        let url = self.request.url();
        let method = self.request.method();
        write!(buf, "{} ", method)?;

        // A client sending a CONNECT request MUST consists of only the host
        // name and port number of the tunnel destination, separated by a colon.
        // See: https://tools.ietf.org/html/rfc7231#section-4.3.6
        if method == Method::Connect {
            let host = url
                .host_str()
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Missing hostname"))?;

            let port = url.port_or_known_default().ok_or_else(|| {
                io::Error::new(
                    io::ErrorKind::InvalidData,
                    "Unexpected scheme with no default port",
                )
            })?;

            write!(buf, "{}:{}", host, port)?;
        } else {
            write!(buf, "{}", url.path())?;
            if let Some(query) = url.query() {
                write!(buf, "?{}", query)?;
            }
        }

        write!(buf, " HTTP/1.1\r\n")?;

        self.finalize_headers()?;
        let mut headers = self.request.iter().collect::<Vec<_>>();
        headers.sort_unstable_by_key(|(h, _)| if **h == HOST { "0" } else { h.as_str() });
        for (header, values) in headers {
            for value in values.iter() {
                write!(buf, "{}: {}\r\n", header, value)?;
            }
        }

        write!(buf, "\r\n")?;
        Ok(Cursor::new(buf))
    }
}

impl Read for Encoder {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<io::Result<usize>> {
        loop {
            self.state = match self.state {
                EncoderState::Start => EncoderState::Head(self.compute_head()?),

                EncoderState::Head(ref mut cursor) => {
                    read_to_end!(Pin::new(cursor).poll_read(cx, buf));
                    EncoderState::Body(BodyEncoder::new(self.request.take_body()))
                }

                EncoderState::Body(ref mut encoder) => {
                    read_to_end!(Pin::new(encoder).poll_read(cx, buf));
                    EncoderState::End
                }

                EncoderState::End => return Poll::Ready(Ok(0)),
            }
        }
    }
}