fetch-happen 0.5.1

A comfortable wrapper for the javascript 'fetch' api, with a native reqwest fallback.
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

#![cfg(not(target_arch = "wasm32"))]
use fetch_happen::{AbortController, Client, Error};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::TcpListener,
    sync::oneshot,
    time::{timeout, Duration},
};

#[tokio::test]
async fn precancelled_request_never_connects() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let controller = AbortController::default();
    controller.abort();
    let result = Client
        .get(format!("http://{}", listener.local_addr().unwrap()))
        .abort_signal(controller.signal())
        .send()
        .await;
    assert!(matches!(result, Err(Error::Aborted)));
    assert!(timeout(Duration::from_millis(30), listener.accept())
        .await
        .is_err());
}

async fn interrupt(after_headers: bool) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let (ready, started) = oneshot::channel();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        while !request.ends_with(b"\r\n\r\n") {
            request.push(stream.read_u8().await.unwrap());
        }
        if after_headers {
            stream
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1000000\r\n\r\nx")
                .await
                .unwrap();
        }
        ready.send(()).unwrap();
        let mut byte = [0];
        // Cancelling drops reqwest's pending request/body, releasing the socket.
        let result = timeout(Duration::from_secs(2), stream.read(&mut byte))
            .await
            .unwrap();
        assert!(matches!(result, Ok(0) | Err(_)));
    });
    let controller = AbortController::default();
    let request = async {
        let response = Client
            .get(url)
            .abort_signal(controller.signal())
            .send()
            .await?;
        // Headers arrived; the body is what stalls.
        response.bytes().await
    };
    let cancel = async {
        started.await.unwrap();
        // Allow the client to consume the headers before interrupting the body case.
        tokio::time::sleep(Duration::from_millis(20)).await;
        controller.abort();
    };
    let (result, ()) = timeout(Duration::from_secs(3), async {
        tokio::join!(request, cancel)
    })
    .await
    .unwrap();
    assert!(matches!(result, Err(Error::Aborted)), "{result:?}");
    server.await.unwrap();
}

/// `send()` must resolve on headers, and chunks must arrive as the server
/// writes them rather than after the whole body lands.
#[tokio::test]
async fn body_streams_after_headers() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let (release, held) = oneshot::channel::<()>();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        while !request.ends_with(b"\r\n\r\n") {
            request.push(stream.read_u8().await.unwrap());
        }
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nhello")
            .await
            .unwrap();
        held.await.unwrap();
        stream.write_all(b"world").await.unwrap();
    });
    let response = timeout(Duration::from_secs(2), Client.get(url).send())
        .await
        .expect("send() should resolve on headers, before the body completes")
        .unwrap();
    assert_eq!(response.status(), 200);
    let reader = response.stream_reader().unwrap();
    let first = timeout(Duration::from_secs(2), reader.read_chunk())
        .await
        .expect("first chunk should arrive before the server sends the rest")
        .unwrap()
        .unwrap();
    assert_eq!(first, b"hello");
    release.send(()).unwrap();
    let mut rest = Vec::new();
    while let Some(chunk) = reader.read_chunk().await.unwrap() {
        rest.extend(chunk);
    }
    assert_eq!(rest, b"world");
    assert!(reader.read_chunk().await.unwrap().is_none());
    assert!(
        matches!(response.bytes().await, Err(Error::Transport(_))),
        "a body handed to a stream reader can't be read again"
    );
    server.await.unwrap();
}
#[tokio::test]
async fn aborts_pending_headers() {
    interrupt(false).await;
}
#[tokio::test]
async fn aborts_pending_body() {
    interrupt(true).await;
}