nwep 0.1.0

idiomatic rust bindings for the web/1 protocol over quic.
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

// the managed AsyncStream: pull a streamed body chunk by chunk, fully async, no
// tick loop or fd in the test. the dedicated stream connection + owner thread
// live behind AsyncStream; reaching the end (recv -> None) means the trailer
// signature verified against the peer, so the body is authentic. mirrors the
// nwcurl --stream / nwdrop large-transfer path on the managed surface.
#![cfg(feature = "runtime")]

use nwep::{Address, Identity, Method, Server, Status};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc, Arc, Mutex};
use std::time::{Duration, Instant};

fn now_ms() -> i64 {
    use std::sync::OnceLock;
    static START: OnceLock<Instant> = OnceLock::new();
    START.get_or_init(Instant::now).elapsed().as_millis() as i64
}

#[tokio::test]
async fn async_stream_pulls_a_large_body() {
    // a body well over one chunk, to force chunked streaming + back-pressure.
    let body: Vec<u8> = (0..400_000u32).map(|i| (i % 251) as u8).collect();
    let body_for_thread = body.clone();

    let server_id = Identity::generate().unwrap();
    let server_node = *server_id.node_id();
    let (tx, rx) = mpsc::channel();
    let stop = Arc::new(AtomicBool::new(false));
    let stop_t = stop.clone();

    // a driven server that streams /big across ticks (the embedder owns its loop).
    let server_thread = std::thread::spawn(move || {
        let pending: Arc<Mutex<Vec<(u64, u64)>>> = Arc::new(Mutex::new(Vec::new()));
        let pending_h = pending.clone();
        let server = Server::builder()
            .identity(server_id)
            .bind(Address::loopback(0))
            .on_request(move |req, res| {
                if req.path() == Some("/big") {
                    pending_h
                        .lock()
                        .unwrap()
                        .push((req.conn_id(), req.stream_id()));
                    res.stream(
                        "/big",
                        Status::Ok,
                        &[("content-type", "application/octet-stream")],
                    )
                } else {
                    res.not_found()
                }
            })
            .build()
            .unwrap();
        tx.send(server.local_port()).unwrap();
        let mut active: Vec<(u64, u64, usize)> = Vec::new();
        while !stop_t.load(Ordering::Relaxed) {
            server.tick(now_ms()).unwrap();
            active.extend(pending.lock().unwrap().drain(..).map(|(c, s)| (c, s, 0)));
            active.retain_mut(|(c, s, sent)| {
                while *sent < body_for_thread.len() {
                    let n = server
                        .stream_send(*c, *s, &body_for_thread[*sent..])
                        .unwrap();
                    *sent += n;
                    if n == 0 {
                        return true;
                    }
                }
                server.stream_end(*c, *s).unwrap();
                false
            });
            std::thread::sleep(Duration::from_millis(1));
        }
    });
    let port = rx.recv_timeout(Duration::from_secs(3)).unwrap();

    // open a managed async stream and pull the body chunk by chunk.
    let mut stream = nwep::Client::builder()
        .identity(Identity::generate().unwrap())
        .stream(Method::Read, "/big", &server_node, &Address::loopback(port))
        .await
        .expect("open async stream");
    assert_eq!(stream.status(), Status::Ok);
    assert_eq!(
        stream.header("content-type"),
        Some("application/octet-stream")
    );

    let mut received = Vec::with_capacity(body.len());
    // recv -> Some(chunk) while streaming; None at the verified end.
    while let Some(chunk) = stream.recv().await.expect("stream chunk") {
        received.extend_from_slice(&chunk);
    }
    assert_eq!(received.len(), body.len());
    assert_eq!(received, body, "the streamed body is byte-exact");
    // reaching None means the trailer signature verified against the peer.

    stop.store(true, Ordering::Relaxed);
    server_thread.join().unwrap();
}