aioduct 0.2.5

Async-native HTTP client built directly on hyper 1.x — no hyper-util, no legacy
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

use super::super::*;
use super::multipart::multipart_body;
use super::support::connected_budget_client_with_h2_server;

use std::convert::Infallible;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Full};

use crate::runtime::TokioRuntime;

fn has_io_error_kind(mut error: &(dyn std::error::Error + 'static), kind: io::ErrorKind) -> bool {
    loop {
        if error
            .downcast_ref::<io::Error>()
            .is_some_and(|error| error.kind() == kind)
        {
            return true;
        }
        let Some(source) = error.source() else {
            return false;
        };
        error = source;
    }
}

#[tokio::test]
async fn hyper_h2_multipart_upload_resumes_after_tls_backpressure() {
    let (mut client_tls, server_tls, control) = connected_budget_client_with_h2_server().await;
    client_tls.tls.set_buffer_limit(Some(1024));

    let boundary = "AioductH2BackpressureBoundary";
    let file: Vec<u8> = (0..4096).map(|i| (i % 251) as u8).collect();
    let multipart = multipart_body(boundary, &file);
    let expected_multipart = multipart.clone();

    let server_task = tokio::spawn(async move {
        hyper::server::conn::http2::Builder::new(crate::runtime::executor::poll_executor::<
            TokioRuntime,
        >())
        .serve_connection(
            server_tls,
            hyper::service::service_fn(move |req: http::Request<hyper::body::Incoming>| {
                let expected_multipart = expected_multipart.clone();
                async move {
                    assert_eq!(req.method(), http::Method::POST);
                    assert_eq!(req.uri().path(), "/api/v2/ocr/jobs");
                    assert_eq!(
                        req.headers().get(http::header::CONTENT_TYPE).unwrap(),
                        &format!("multipart/form-data; boundary={boundary}")
                    );
                    let body = req.into_body().collect().await.unwrap().to_bytes();
                    assert_eq!(body, expected_multipart);
                    Ok::<_, Infallible>(hyper::Response::new(Full::new(Bytes::from_static(
                        b"upload ok",
                    ))))
                }
            }),
        )
        .await
    });

    let (mut sender, connection) = hyper::client::conn::http2::handshake(
        crate::runtime::executor::poll_executor::<TokioRuntime>(),
        client_tls,
    )
    .await
    .unwrap();
    let connection_task = tokio::spawn(connection);
    sender.ready().await.unwrap();

    control.set_write_budget(Some(1));
    let release_control = control.clone();
    let release_task = tokio::spawn(async move {
        tokio::time::timeout(
            Duration::from_secs(2),
            release_control.wait_for_blocked_writes(2),
        )
        .await
        .expect("HTTP/2 TLS transport should become backpressured twice");
        release_control.set_write_budget(None);
    });

    let request = http::Request::builder()
        .method(http::Method::POST)
        .uri("https://localhost/api/v2/ocr/jobs")
        .header(
            http::header::CONTENT_TYPE,
            format!("multipart/form-data; boundary={boundary}"),
        )
        .header(http::header::CONTENT_LENGTH, multipart.len())
        .body(Full::new(Bytes::from(multipart)))
        .unwrap();
    let response = tokio::time::timeout(Duration::from_secs(5), sender.send_request(request))
        .await
        .expect("HTTP/2 request should not hang")
        .expect("HTTP/2 request should survive TLS backpressure");
    assert_eq!(response.status(), http::StatusCode::OK);
    assert_eq!(
        response.into_body().collect().await.unwrap().to_bytes(),
        Bytes::from_static(b"upload ok")
    );

    drop(sender);
    release_task.await.unwrap();
    tokio::time::timeout(Duration::from_secs(2), connection_task)
        .await
        .expect("HTTP/2 client connection should finish")
        .expect("HTTP/2 client connection task should not panic")
        .expect("HTTP/2 client connection should close cleanly");
    let server_result = tokio::time::timeout(Duration::from_secs(2), server_task)
        .await
        .expect("HTTP/2 server connection should finish")
        .expect("HTTP/2 server connection task should not panic");
    if let Err(error) = server_result {
        assert!(
            has_io_error_kind(&error, io::ErrorKind::BrokenPipe),
            "HTTP/2 server must not fail from trailing or corrupt plaintext: {error}"
        );
    }
}