xtor 0.9.10

Async Actor framework for Rust which is blazing fast and rock solid.
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

use std::sync::atomic::{AtomicBool, AtomicUsize};

use super::*;
use crate::{utils::default_supervisor::DefaultSupervisor, ActorRestart, Supervise, WeakAddr};

struct Dummy(AtomicBool, AtomicUsize);
impl Default for Dummy {
    fn default() -> Self {
        Dummy(AtomicBool::new(false), AtomicUsize::new(0))
    }
}
impl Actor for Dummy {}

#[crate::message(result = "()")]
struct Die;

#[async_trait::async_trait]
impl Handler<Die> for Dummy {
    async fn handle(&self, _ctx: &Context, _msg: Die) -> anyhow::Result<()> {
        println!("Dummy actor is died");
        self.0.store(true, std::sync::atomic::Ordering::SeqCst);
        Err(anyhow::anyhow!("dummy died"))
    }
}

#[crate::message(result = "(bool, usize)")]
struct IsAlive;

#[async_trait::async_trait]
impl Handler<IsAlive> for Dummy {
    async fn handle(&self, _ctx: &Context, _msg: IsAlive) -> anyhow::Result<(bool, usize)> {
        Ok((
            self.0.load(std::sync::atomic::Ordering::SeqCst),
            self.1.load(std::sync::atomic::Ordering::SeqCst),
        ))
    }
}

#[async_trait::async_trait]
impl ActorRestart for Dummy {
    async fn on_restart(&self, _addr: &WeakAddr) {
        println!("restarting Dummy");
        self.0.store(false, std::sync::atomic::Ordering::SeqCst);
        self.1.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    }
}

#[crate::test]
async fn test_supervisor_save_one() {
    let supervisor = DefaultSupervisor::new(
        xtor::utils::default_supervisor::DefaultSupervisorRestartStrategy::OneForOne,
    )
    .spawn()
    .await
    .unwrap();
    let supervise_proxy = supervisor.proxy::<DefaultSupervisor, Supervise>().await;
    let actor = Dummy::default()
        .spawn_supervisable()
        .await
        .unwrap()
        .chain_link_to_supervisor(&supervise_proxy)
        .await
        .unwrap();
    let _ = actor.call::<Dummy, Die>(Die).await;
    assert!(!actor.is_stopped().await);
    assert!(!actor.call::<Dummy, IsAlive>(IsAlive).await.unwrap().0);
}

#[crate::test]
async fn test_supervisor_save_all() {
    let supervisor = DefaultSupervisor::new(
        xtor::utils::default_supervisor::DefaultSupervisorRestartStrategy::OneForAll,
    )
    .spawn()
    .await
    .unwrap();
    let supervise_proxy = supervisor.proxy::<DefaultSupervisor, Supervise>().await;
    let actors = futures::future::join_all((0..10).map(|_| async {
        Dummy::default()
            .spawn_supervisable()
            .await
            .unwrap()
            .chain_link_to_supervisor(&supervise_proxy)
            .await
            .unwrap()
    }))
    .await;
    let _ = actors.get(0).unwrap().call::<Dummy, Die>(Die).await;
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    for a in actors {
        assert_eq!(a.call::<Dummy, IsAlive>(IsAlive).await.unwrap().1, 1);
    }
}