compio-actor 0.1.0

Actors for Compio
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

use std::{convert::Infallible, io};

use compio_actor::{Actor, ActorExit, Broker, Call, Cluster, Handler, Mailbox};

#[derive(Debug)]
struct Add(usize);

#[derive(Debug)]
struct Read;

struct Counter;
struct Gauge;

impl Actor for Counter {
    type Arguments = ();
    type Error = Infallible;
    type State = usize;

    async fn pre_start(
        &self,
        _myself: &Mailbox<Self>,
        (): Self::Arguments,
    ) -> Result<Self::State, Self::Error> {
        Ok(0)
    }
}

impl Handler<Add> for Counter {
    async fn handle(
        &self,
        _myself: &Mailbox<Self>,
        Add(value): Add,
        state: &mut Self::State,
    ) -> Result<(), Self::Error> {
        *state += value;
        Ok(())
    }
}

impl Handler<Call<Read, usize>> for Counter {
    async fn handle(
        &self,
        _myself: &Mailbox<Self>,
        call: Call<Read, usize>,
        state: &mut Self::State,
    ) -> Result<(), Self::Error> {
        call.reply(*state).ok();
        Ok(())
    }
}

impl Actor for Gauge {
    type Arguments = ();
    type Error = Infallible;
    type State = usize;

    async fn pre_start(
        &self,
        _myself: &Mailbox<Self>,
        (): Self::Arguments,
    ) -> Result<Self::State, Self::Error> {
        Ok(0)
    }
}

impl Handler<Add> for Gauge {
    async fn handle(
        &self,
        _myself: &Mailbox<Self>,
        Add(value): Add,
        state: &mut Self::State,
    ) -> Result<(), Self::Error> {
        *state += value;
        Ok(())
    }
}

impl Handler<Call<Read, usize>> for Gauge {
    async fn handle(
        &self,
        _myself: &Mailbox<Self>,
        call: Call<Read, usize>,
        state: &mut Self::State,
    ) -> Result<(), Self::Error> {
        call.reply(*state).ok();
        Ok(())
    }
}

#[compio_macros::main]
async fn main() -> io::Result<()> {
    let cluster = Cluster::new()?;
    let (counter, counter_handle) = cluster.spawn(|| Counter, ()).await.unwrap();
    let (gauge, gauge_handle) = cluster.spawn(|| Gauge, ()).await.unwrap();

    let actors: Vec<Broker<Add>> = vec![counter.broker(), gauge.broker()];
    for actor in &actors {
        actor.send(Add(2)).unwrap();
    }
    assert_eq!(counter.call(Read).await.unwrap(), 2);
    assert_eq!(gauge.call(Read).await.unwrap(), 2);

    counter.stop();
    gauge.stop();
    assert_eq!(counter_handle.await.unwrap(), ActorExit::Stopped);
    assert_eq!(gauge_handle.await.unwrap(), ActorExit::Stopped);
    cluster.join().await
}