continuo 0.12.0

Runtime service composition for multi-service Rust applications: typed component resolution, dependency-ordered lifecycle, reload, runnable tasks, graceful drain, and finalization.
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

use continuo::Gate;
use std::sync::Mutex;

struct ReleaseOnCapacity(Mutex<Option<continuo::Permit>>);
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S>
    for ReleaseOnCapacity
{
    fn on_event(
        &self,
        event: &tracing::Event<'_>,
        _: tracing_subscriber::layer::Context<'_, S>,
    ) {
        if event.metadata().target() == "continuo::gate" {
            drop(self.0.lock().unwrap().take());
        }
    }
}
#[tokio::test]
async fn enter_does_not_lose_capacity_release() {
    use tracing_subscriber::layer::SubscriberExt;
    let gate = Gate::new(Some(1), continuo::gate::default_acquire_timeout());
    let held = gate.try_enter().unwrap();
    let subscriber = tracing_subscriber::registry()
        .with(ReleaseOnCapacity(Mutex::new(Some(held))));
    let _default = tracing::subscriber::set_default(subscriber);
    // The synchronous capacity debug event releases the permit after the
    // failed capacity check but before enter creates its Notified future.
    let result = gate.enter().await;
    assert_eq!(gate.count(), 1);
    assert!(
        result.is_ok(),
        "capacity was released before sleeping, but enter returned {:?}",
        result.err()
    );
}

#[tokio::test]
async fn queued_entrants_recheck_capacity_and_shutdown_interrupts() {
    use std::future::Future;
    use std::task::{Context, Waker};
    let gate = Gate::new(Some(1), continuo::gate::default_acquire_timeout());
    let held = gate.try_enter().unwrap();
    let mut first = Box::pin(gate.enter());
    let mut second = Box::pin(gate.enter());
    let mut cx = Context::from_waker(Waker::noop());
    assert!(first.as_mut().poll(&mut cx).is_pending());
    assert!(second.as_mut().poll(&mut cx).is_pending());
    drop(held);
    let held = first.await.unwrap();
    assert!(second.as_mut().poll(&mut cx).is_pending());
    drop(held);
    let held = second.await.unwrap();
    let mut waiting = Box::pin(gate.enter());
    assert!(waiting.as_mut().poll(&mut cx).is_pending());
    gate.graceful_shutdown(None);
    assert!(matches!(waiting.await, Err(continuo::gate::Error::ShuttingDown)));
    drop(held);
    assert_eq!(gate.wait_all_done().await, continuo::GateDrainOutcome::Drained);
}