camber 0.1.1

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
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

use crate::resource::{HealthState, Resource};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

/// Shut down all registered resources in parallel.
/// Errors are logged but do not prevent others.
pub(crate) fn shutdown_resources(resources: &[Box<dyn Resource>]) {
    std::thread::scope(|scope| spawn_shutdown_tasks(scope, resources));
}

fn spawn_shutdown_tasks<'scope, 'env>(
    scope: &'scope std::thread::Scope<'scope, 'env>,
    resources: &'env [Box<dyn Resource>],
) {
    for resource in resources.iter() {
        let resource = resource.as_ref();
        scope.spawn(move || shutdown_one(resource));
    }
}

fn shutdown_one(resource: &dyn Resource) {
    if let Err(e) = resource.shutdown() {
        tracing::error!(resource = resource.name(), error = %e, "resource shutdown failed");
    }
}

/// Log a health check failure at warn level. Successes are silent.
fn log_health_result(name: &str, result: &Result<(), crate::RuntimeError>) {
    if let Err(e) = result {
        tracing::warn!(resource = name, error = %e, "health check failed");
    }
}

/// Spawn one background task per resource that periodically runs health checks.
/// Each task updates the corresponding AtomicBool in the health state array.
/// Tasks exit when shutdown is notified or when aborted.
pub(crate) fn spawn_health_tasks(
    resources: &Arc<[Box<dyn Resource>]>,
    health_state: &Option<HealthState>,
    interval: Duration,
    shutdown_notify: &Arc<tokio::sync::Notify>,
) -> Box<[tokio::task::JoinHandle<()>]> {
    let hs = match health_state {
        Some(hs) => hs,
        None => return Vec::new().into_boxed_slice(),
    };

    resources
        .iter()
        .enumerate()
        .map(|(idx, _)| {
            spawn_one_health_task(
                Arc::clone(resources),
                Arc::clone(hs),
                Arc::clone(shutdown_notify),
                interval,
                idx,
            )
        })
        .collect()
}

pub(crate) async fn run_initial_health_checks(
    resources: &Arc<[Box<dyn Resource>]>,
    health_state: &HealthState,
) {
    let mut join_set = tokio::task::JoinSet::new();
    for idx in 0..resources.len() {
        spawn_initial_health_check(
            &mut join_set,
            Arc::clone(resources),
            Arc::clone(health_state),
            idx,
        );
    }
    while join_set.join_next().await.is_some() {}
}

fn spawn_initial_health_check(
    join_set: &mut tokio::task::JoinSet<()>,
    resources: Arc<[Box<dyn Resource>]>,
    health_state: HealthState,
    idx: usize,
) {
    join_set.spawn_blocking(move || update_resource_health(resources.as_ref(), &health_state, idx));
}

fn spawn_one_health_task(
    resources: Arc<[Box<dyn Resource>]>,
    health_state: HealthState,
    shutdown_notify: Arc<tokio::sync::Notify>,
    interval: Duration,
    idx: usize,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            tokio::select! {
                () = tokio::time::sleep(interval) => {}
                () = shutdown_notify.notified() => return,
            }
            tokio::task::block_in_place(|| {
                update_resource_health(resources.as_ref(), &health_state, idx)
            });
        }
    })
}

fn update_resource_health(resources: &[Box<dyn Resource>], health_state: &HealthState, idx: usize) {
    let result = resources[idx].health_check();
    log_health_result(resources[idx].name(), &result);
    health_state[idx].1.store(result.is_ok(), Ordering::Release);
}