leyline-http 0.1.1

HTTP client that mimics browsers on the wire
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204

use super::super::Session;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

async fn read_request_head(sock: &mut tokio::net::TcpStream) -> bool {
    let mut buf = Vec::new();
    let mut tmp = [0u8; 1024];
    loop {
        let n = sock.read(&mut tmp).await.unwrap();
        if n == 0 {
            return !buf.is_empty();
        }
        buf.extend_from_slice(&tmp[..n]);
        if buf.windows(4).any(|w| w == b"\r\n\r\n") {
            return true;
        }
    }
}

#[tokio::test]
async fn streamed_chunked_response_reassembles() {
    use futures_util::StreamExt;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        read_request_head(&mut sock).await;
        sock.write_all(
            b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\
              7\r\nHello, \r\nA\r\nstreaming \r\n6\r\nworld!\r\n0\r\n\r\n",
        )
        .await
        .unwrap();
    });

    let resp = Session::browser(crate::Browser::default())
        .request(http::Method::GET, format!("http://{addr}/x"))
        .stream()
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let mut body = Vec::new();
    let mut stream = resp.into_stream().unwrap();
    while let Some(chunk) = stream.next().await {
        body.extend_from_slice(&chunk.unwrap());
    }
    assert_eq!(body, b"Hello, streaming world!");
    server.await.unwrap();
}

#[tokio::test]
async fn streamed_fixed_length_response_reassembles() {
    use futures_util::StreamExt;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let payload = b"the quick brown fox ".repeat(500);
    let payload_srv = payload.clone();
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        read_request_head(&mut sock).await;
        let head = format!(
            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n",
            payload_srv.len()
        );
        sock.write_all(head.as_bytes()).await.unwrap();
        sock.write_all(&payload_srv).await.unwrap();
    });

    let resp = Session::browser(crate::Browser::default())
        .request(http::Method::GET, format!("http://{addr}/x"))
        .stream()
        .send()
        .await
        .unwrap();
    let mut body = Vec::new();
    let mut stream = resp.into_stream().unwrap();
    while let Some(chunk) = stream.next().await {
        body.extend_from_slice(&chunk.unwrap());
    }
    assert_eq!(body, payload);
    server.await.unwrap();
}

#[tokio::test]
async fn streamed_connection_is_reused_after_full_drain() {
    use futures_util::StreamExt;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        for body in [b"first".as_slice(), b"second".as_slice()] {
            assert!(read_request_head(&mut sock).await, "expected a request");
            let head = format!(
                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: keep-alive\r\n\r\n",
                body.len()
            );
            sock.write_all(head.as_bytes()).await.unwrap();
            sock.write_all(body).await.unwrap();
        }
    });

    let session = Session::browser(crate::Browser::default());
    let r1 = session
        .request(http::Method::GET, format!("http://{addr}/a"))
        .stream()
        .send()
        .await
        .unwrap();
    let mut b1 = Vec::new();
    let mut s1 = r1.into_stream().unwrap();
    while let Some(chunk) = s1.next().await {
        b1.extend_from_slice(&chunk.unwrap());
    }
    assert_eq!(b1, b"first");

    let r2 = session.get(format!("http://{addr}/b")).await.unwrap();
    assert_eq!(r2.text().await.unwrap(), "second");
    server.await.unwrap();
}

#[tokio::test]
async fn streamed_read_timeout_fires_on_stall() {
    use futures_util::StreamExt;
    use std::time::Duration;

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let server = tokio::spawn(async move {
        let (mut sock, _) = listener.accept().await.unwrap();
        read_request_head(&mut sock).await;
        sock.write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n")
            .await
            .unwrap();
        tokio::time::sleep(Duration::from_secs(5)).await;
    });

    let session = Session::builder()
        .timeout(crate::TimeoutConfig::new().read(Duration::from_millis(200)))
        .build()
        .unwrap();
    let resp = session
        .request(http::Method::GET, format!("http://{addr}/x"))
        .stream()
        .send()
        .await
        .unwrap();
    let mut stream = resp.into_stream().unwrap();
    assert_eq!(&stream.next().await.unwrap().unwrap()[..], b"hello");
    let err = stream.next().await.unwrap().unwrap_err();
    assert_eq!(err.kind(), std::io::ErrorKind::TimedOut, "{err}");
    server.abort();
}

#[tokio::test]
async fn streamed_connection_dropped_when_consumer_drops_early() {
    use futures_util::StreamExt;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let accepts = Arc::new(AtomicUsize::new(0));
    let accepts_srv = accepts.clone();
    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            accepts_srv.fetch_add(1, Ordering::SeqCst);
            tokio::spawn(async move {
                while read_request_head(&mut sock).await {
                    let mut resp =
                        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n"
                            .to_vec();
                    for _ in 0..50 {
                        resp.extend_from_slice(b"A\r\n0123456789\r\n");
                    }
                    resp.extend_from_slice(b"0\r\n\r\n");
                    if sock.write_all(&resp).await.is_err() {
                        break;
                    }
                }
            });
        }
    });

    let session = Session::browser(crate::Browser::default());
    let r1 = session
        .request(http::Method::GET, format!("http://{addr}/a"))
        .stream()
        .send()
        .await
        .unwrap();
    let mut s1 = r1.into_stream().unwrap();
    let _first = s1.next().await.unwrap().unwrap();
    drop(s1);
    let r2 = session.get(format!("http://{addr}/b")).await.unwrap();
    assert_eq!(r2.status(), 200);
    assert_eq!(
        accepts.load(Ordering::SeqCst),
        2,
        "an early-dropped stream's connection must not be reused"
    );
}