camber 0.2.0

Opinionated async Rust for IO-bound services on top of Tokio
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

use crate::common;

use camber::http::{self, Request, Response, Router};
use camber::runtime;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

const EVENT_TIMEOUT: Duration = Duration::from_secs(2);

#[test]
fn builder_configures_concurrent_requests() {
    runtime::builder()
        .worker_threads(2)
        .keepalive_timeout(Duration::from_millis(100))
        .shutdown_timeout(Duration::from_secs(2))
        .run(|| {
            let mut router = Router::new();
            router.get("/slow", |_req: &Request| async {
                Response::text(200, "done")
            });

            let server = common::spawn_server_ready(router, EVENT_TIMEOUT).unwrap();
            let addr = server.local_addr();

            let counter = AtomicUsize::new(0);
            let url = format!("http://{addr}/slow");
            let responses = common::block_on(futures_util::future::join_all(
                (0..3).map(|_| http::get(&url)),
            ));
            responses.into_iter().for_each(|response| {
                let resp = response.unwrap();
                assert_eq!(resp.status(), 200);
                counter.fetch_add(1, Ordering::SeqCst);
            });

            assert_eq!(counter.load(Ordering::SeqCst), 3);

            server.shutdown_bounded(EVENT_TIMEOUT).unwrap();
            runtime::request_shutdown();
        })
        .unwrap();
}

#[test]
fn builder_configures_shutdown_timeout() {
    let start = Instant::now();

    let task = runtime::builder()
        .shutdown_timeout(Duration::from_millis(200))
        .run(|| {
            let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
            let task = camber::spawn_async(async move {
                let _ = entered_tx.send(());
                std::future::pending::<()>().await;
            });
            common::block_on(entered_rx).unwrap();
            runtime::request_shutdown();
            task
        })
        .unwrap();

    let elapsed = start.elapsed();

    assert!(
        elapsed < Duration::from_secs(1),
        "expected < 1s (safety-net timeout), got {elapsed:?}"
    );

    let join_runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    let result = join_runtime.block_on(async { tokio::time::timeout(EVENT_TIMEOUT, task).await });
    assert!(
        result.is_ok(),
        "timed-out task handle was not boundedly joined"
    );
    assert!(
        result.unwrap().is_err(),
        "pending task completed successfully"
    );
}