rutis 0.2.2

Cordis core paradigms in idiomatic Rust: plugin assembly, fiber lifecycle, typed service registry, dependency-driven reload
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

//! 终态释放(0.2.1)单元验证:keyed 通道摘除监听后删除空条目,
//! 派发尾链任务完成后自摘——随实例 churn 的通道与尾链不残留。

use super::*;
use crate::ctx::Ctx;
use crate::error::CordisError;
use crate::event::{Event, Listener, Next};
use crate::BoxFuture;
use std::time::Duration;

struct Ping;

impl Event for Ping {
    const NAME: &'static str = "test::TransientPing";
    type Value = ();
}

struct Nop;

impl Listener<Ping> for Nop {
    fn call<'a>(
        &'a self,
        _ctx: &'a Ctx,
        _e: &'a Ping,
    ) -> BoxFuture<'a, Result<Option<()>, CordisError>> {
        Box::pin(async { Ok(None) })
    }
}

struct NopWaterfall;

impl crate::event::WaterfallListener<Ping> for NopWaterfall {
    fn call<'a>(
        &'a self,
        _ctx: &'a Ctx,
        _e: &'a Ping,
        next: Next<'a, Ping>,
    ) -> BoxFuture<'a, Result<(), CordisError>> {
        Box::pin(async move { next.call().await })
    }
}

#[tokio::test]
async fn keyed_channels_and_dispatch_tails_prune() {
    let ctx = Ctx::root().expect("runtime in scope");
    let bus = ctx.events().clone();
    for i in 0..25 {
        let name = format!("ch/{i}");
        let listener = bus.on_keyed::<Ping>(&ctx, name.clone(), Nop).unwrap();
        let wf = bus
            .on_waterfall_keyed::<Ping>(&ctx, name.clone(), NopWaterfall)
            .unwrap();
        bus.emit_keyed::<Ping>(&ctx, name.clone(), Arc::new(Ping));
        listener.dispose().await.unwrap();
        wf.dispose().await.unwrap();
    }
    // 最后一次派发任务完成后自摘尾链;监听条目已在 dispose 内删除
    for _ in 0..500 {
        let empty = {
            let inner = bus.inner.lock().unwrap();
            inner.hooks.is_empty() && inner.wf_hooks.is_empty() && inner.dispatch_tail.is_empty()
        };
        if empty {
            break;
        }
        tokio::time::sleep(Duration::from_millis(2)).await;
    }
    {
        let inner = bus.inner.lock().unwrap();
        assert!(
            inner.hooks.is_empty() && inner.wf_hooks.is_empty(),
            "keyed channels must drop empty listener lists"
        );
        assert!(
            inner.dispatch_tail.is_empty(),
            "dispatch tails must self-remove on completion"
        );
    }
    // 摘除后再次注册/派发同键通道仍工作(条目按需重建)
    let listener = bus.on_keyed::<Ping>(&ctx, "ch/0", Nop).unwrap();
    bus.emit_keyed::<Ping>(&ctx, "ch/0", Arc::new(Ping));
    listener.dispose().await.unwrap();
}