speare 0.4.2

actor-like thin abstraction over tokio::task and flume channels
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use speare::{req_res, Actor, Ctx, Handle, Node, Request, Supervision};
use tokio::task;

struct Foo;

impl Actor for Foo {
    type Props = ();
    type Msg = ();
    type Err = ();

    async fn init(_: &mut Ctx<Self>) -> Result<Self, Self::Err> {
        Ok(Foo)
    }
}

struct Bar;

impl Actor for Bar {
    type Props = ();
    type Msg = ();
    type Err = ();

    async fn init(_: &mut Ctx<Self>) -> Result<Self, Self::Err> {
        Ok(Bar)
    }
}

#[allow(clippy::disallowed_names)]
#[tokio::test]
async fn node_stops_all_actors_when_dropped() {
    // Arrange
    let mut node = Node::default();
    let foo = node.actor::<Foo>(()).spawn();
    let bar = node.actor::<Bar>(()).spawn();

    // Act
    drop(node);
    task::yield_now().await;

    // Assert
    assert!(!foo.is_alive());
    assert!(!bar.is_alive());
}

struct Quitter;

impl Actor for Quitter {
    type Props = bool;
    type Msg = ();
    type Err = ();

    async fn init(ctx: &mut Ctx<Self>) -> Result<Self, Self::Err> {
        if *ctx.props() {
            Err(())
        } else {
            Ok(Quitter)
        }
    }

    async fn handle(&mut self, _: Self::Msg, _: &mut Ctx<Self>) -> Result<(), Self::Err> {
        Err(())
    }
}

#[tokio::test]
async fn root_supervision_works() {
    // Arrange
    let mut node = Node::default();
    let quit_on_start = false;
    let quitter = node
        .actor::<Quitter>(quit_on_start)
        .supervision(Supervision::Stop)
        .spawn();
    assert!(quitter.is_alive());

    // Error on handle
    // Act
    quitter.send(());
    task::yield_now().await;

    // Assert
    assert!(!quitter.is_alive());

    // Error on init
    // Act
    let quit_on_start = true;
    let quitter2 = node
        .actor::<Quitter>(quit_on_start)
        .supervision(Supervision::Stop)
        .spawn();
    task::yield_now().await;

    // Assert
    assert!(!quitter2.is_alive());
}

struct Parent {
    foo: Handle<()>,
    bar: Handle<()>,
}

impl Actor for Parent {
    type Props = ();
    type Msg = Request<(), (Handle<()>, Handle<()>)>;
    type Err = ();

    async fn init(ctx: &mut Ctx<Self>) -> Result<Self, Self::Err> {
        Ok(Parent {
            foo: ctx.actor::<Foo>(()).spawn(),
            bar: ctx.actor::<Bar>(()).spawn(),
        })
    }

    async fn handle(&mut self, msg: Self::Msg, _: &mut Ctx<Self>) -> Result<(), Self::Err> {
        msg.reply((self.foo.clone(), self.bar.clone()));

        Ok(())
    }
}

#[allow(clippy::disallowed_names)]
#[tokio::test]
async fn stopping_a_root_actor_stops_all_its_children() {
    // Arrange
    let mut node = Node::default();
    let parent = node.actor::<Parent>(()).spawn();

    let (req, res) = req_res(());
    parent.send(req);
    let (foo, bar) = res.recv().await.unwrap();

    assert!(foo.is_alive());
    assert!(bar.is_alive());

    // Act
    parent.stop();
    task::yield_now().await;

    // Assert
    assert!(!foo.is_alive());
    assert!(!bar.is_alive());
}