yaaf 0.1.0

Yet Another Actor Framework
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

use ::async_trait::async_trait;
use ::std::sync::Arc;
use ::tokio::sync::{
    oneshot::{channel, Sender},
    Mutex,
};
use ::yaaf::prelude::*;

#[derive(Clone)]
struct Communication(String);

#[derive(Source)]
#[publish(Communication)]
struct Alice;

#[async_trait]
impl Source for Alice {
    async fn run(mut self, mut ctx: SourceContext) {
        ctx.publish::<Alice, Communication>(&self, Communication("hello".into()));
    }
}

#[derive(Actor)]
#[handle(Communication)]
struct Bob {
    done: Option<Sender<()>>,
    visited: Arc<Mutex<bool>>,
}

impl Bob {
    fn complete(&mut self) {
        let done = self.done.take();
        if let Some(done) = done {
            done.send(()).unwrap();
        }
    }
}

#[async_trait]
impl Handler<Communication> for Bob {
    async fn handle(&mut self, _ctx: &mut HandleContext, message: Communication) {
        println!("Received Communication(\"{}\")", message.0);
        {
            let mut visited = self.visited.lock().await;
            *visited = true;
        }
        self.complete();
    }
}

#[tokio::test]
async fn simple_pubsub() -> Result<(), Box<dyn ::std::error::Error>> {
    let mut system = System::new().await;

    let alice = Alice;

    let (send, recv) = channel();
    let visited = Arc::new(Mutex::new(false));
    let bob = Bob {
        done: Some(send),
        visited: visited.clone(),
    };

    system.add_actor(bob).await;
    system.add_source(alice).await;

    recv.await.unwrap();
    system.shutdown().await;

    assert_eq!(true, *visited.lock().await);
    Ok(())
}