liburlx 0.2.2

A memory-safe URL transfer library — idiomatic Rust reimplementation of libcurl
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

//! Integration tests for connection pooling (HTTP keep-alive).

#![allow(clippy::unwrap_used, unused_results, clippy::significant_drop_tightening)]

use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;

use http_body_util::Full;
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::Response;
use tokio::net::TcpListener;
use tokio::sync::oneshot;

/// A test HTTP server that tracks the number of TCP connections accepted.
struct TrackingServer {
    addr: SocketAddr,
    connection_count: Arc<AtomicU32>,
    shutdown: Option<oneshot::Sender<()>>,
}

impl TrackingServer {
    async fn start() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let connection_count = Arc::new(AtomicU32::new(0));
        let count = connection_count.clone();
        let (tx, mut rx) = oneshot::channel();

        tokio::spawn(async move {
            loop {
                tokio::select! {
                    accept_result = listener.accept() => {
                        if let Ok((stream, _)) = accept_result {
                            count.fetch_add(1, Ordering::SeqCst);
                            let io = hyper_util::rt::TokioIo::new(stream);
                            tokio::spawn(async move {
                                let _ = http1::Builder::new()
                                    .keep_alive(true)
                                    .serve_connection(
                                        io,
                                        service_fn(|_req| async {
                                            Ok::<_, Infallible>(
                                                Response::builder()
                                                    .status(200)
                                                    .body(Full::new(Bytes::from("ok")))
                                                    .unwrap(),
                                            )
                                        }),
                                    )
                                    .await;
                            });
                        }
                    }
                    _ = &mut rx => {
                        break;
                    }
                }
            }
        });

        Self { addr, connection_count, shutdown: Some(tx) }
    }

    fn url(&self, path: &str) -> String {
        format!("http://127.0.0.1:{}{path}", self.addr.port())
    }

    fn connections(&self) -> u32 {
        self.connection_count.load(Ordering::SeqCst)
    }
}

impl Drop for TrackingServer {
    fn drop(&mut self) {
        if let Some(tx) = self.shutdown.take() {
            let _ = tx.send(());
        }
    }
}

#[tokio::test]
async fn connection_reused_for_sequential_requests() {
    let server = TrackingServer::start().await;

    let mut easy = liburlx::Easy::new();

    // First request
    easy.url(&server.url("/first")).unwrap();
    let resp = easy.perform_async().await.unwrap();
    assert_eq!(resp.status(), 200);

    // Second request to same host
    easy.url(&server.url("/second")).unwrap();
    let resp = easy.perform_async().await.unwrap();
    assert_eq!(resp.status(), 200);

    // Should have reused the connection (only 1 TCP accept)
    assert_eq!(server.connections(), 1, "should reuse connection");
}

#[tokio::test]
async fn new_connection_for_different_ports() {
    let server1 = TrackingServer::start().await;
    let server2 = TrackingServer::start().await;

    let mut easy = liburlx::Easy::new();

    easy.url(&server1.url("/")).unwrap();
    let resp = easy.perform_async().await.unwrap();
    assert_eq!(resp.status(), 200);

    easy.url(&server2.url("/")).unwrap();
    let resp = easy.perform_async().await.unwrap();
    assert_eq!(resp.status(), 200);

    // Different ports → different connections
    assert_eq!(server1.connections(), 1);
    assert_eq!(server2.connections(), 1);
}

#[tokio::test]
async fn three_requests_one_connection() {
    let server = TrackingServer::start().await;

    let mut easy = liburlx::Easy::new();

    for i in 0..3 {
        easy.url(&server.url(&format!("/req{i}"))).unwrap();
        let resp = easy.perform_async().await.unwrap();
        assert_eq!(resp.status(), 200);
    }

    assert_eq!(server.connections(), 1, "should reuse connection for all 3 requests");
}