tokactor 2.1.0

A actor model framework wrapped around tokio
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
143
144
145
146
147
148
149
150
151
152
153
154
155
156

use std::time::Duration;

use tokactor::{Actor, Ctx, DeadActorResult, Handler, IntoFutureError};

pub struct StringMessage(pub String);

/**
 * Test that IntoFuture integration works with just one actor. Actors will represent
 * a unit of work
 */
#[derive(Debug, Default)]
struct StateActor {
    state: Option<String>,
}

impl Actor for StateActor {}

impl Handler<StringMessage> for StateActor {
    fn handle(&mut self, message: StringMessage, _context: &mut Ctx<Self>) {
        self.state = Some(message.0)
    }
}

#[tokio::test]
async fn create_actor_process_message_and_complete() {
    let actor = StateActor { state: None }.start();
    actor.try_send(StringMessage("test".to_string()));
    let a = actor.await.unwrap();
    assert_eq!(a.state, Some("test".to_string()));
}

#[tokio::test]
async fn panic_when_awaiting_on_async_actor() {
    let a1 = StateActor { state: None }.start();
    let a2 = a1.clone();
    a1.try_send(StringMessage("test".to_string()));
    assert_eq!(a1.await.unwrap().state, Some("test".to_string()));
    assert_eq!(a2.await.unwrap_err(), IntoFutureError::MailboxClosed);
}

/**
 * Test that IntoFuture works when a parent actor spawns many child actors. All
 * Child actors should finish executing at some point at which point, the parent
 * actor completes
 */
struct Children {
    count: usize,
}

struct ParentActor {
    children: usize,
    total: usize,
    dead: usize,
}

impl Actor for ParentActor {}

impl Handler<Children> for ParentActor {
    fn handle(&mut self, message: Children, context: &mut Ctx<Self>) {
        self.children = message.count;
        for index in 1..=message.count {
            context.spawn(ChildActor { index });
        }
    }
}

impl Handler<DeadActorResult<ChildActor>> for ParentActor {
    fn handle(&mut self, message: DeadActorResult<ChildActor>, _context: &mut Ctx<Self>) {
        let dead = message.unwrap();
        self.total += dead.actor.index;
        self.dead += 1;
    }
}

struct ChildActor {
    index: usize,
}

impl Actor for ChildActor {}

#[tokio::test]
async fn supervisor_actor_fan_out_and_fan_in_complete() {
    let addr = ParentActor {
        children: 0,
        total: 0,
        dead: 0,
    }
    .start();
    addr.try_send(Children { count: 5 });
    let parent = addr.await.unwrap();
    assert_eq!(parent.children, 5);
    assert_eq!(parent.total, 15);
    assert_eq!(parent.dead, 5);
}

#[tokio::test]
async fn supervisor_actor_fan_out_and_fan_in_complete_2() {
    let addr = ParentActor {
        children: 0,
        total: 0,
        dead: 0,
    }
    .start();
    addr.try_send(Children { count: 40 });
    let parent = addr.await.unwrap();
    assert_eq!(parent.children, 40);
    // assert_eq!(parent.total, 15);
    assert_eq!(parent.dead, 40);
}

/**
 * Test that IntoFuture works when a parent actor spawns many async child actors. All
 * Child actors should finish executing at some point at which point, the parent
 * actor completes.
 */
struct AsyncChildren {
    count: usize,
}

struct AsyncChildrenResult {
    index: usize,
}

impl Handler<AsyncChildren> for ParentActor {
    fn handle(&mut self, message: AsyncChildren, context: &mut Ctx<Self>) {
        self.children = message.count;
        for index in 1..=message.count {
            context.anonymous(async move {
                tokio::time::sleep(Duration::from_secs(1)).await;
                AsyncChildrenResult { index }
            });
        }
    }
}

impl Handler<AsyncChildrenResult> for ParentActor {
    fn handle(&mut self, message: AsyncChildrenResult, _context: &mut Ctx<Self>) {
        self.total += message.index;
        self.dead += 1;
    }
}

#[tokio::test]
async fn supervisor_actor_async_fan_out_and_fan_in_complete() {
    let addr = ParentActor {
        children: 0,
        total: 0,
        dead: 0,
    }
    .start();
    addr.try_send(AsyncChildren { count: 5 });
    let parent = addr.await.unwrap();
    assert_eq!(parent.children, 5);
    assert_eq!(parent.total, 0);
    assert_eq!(parent.dead, 0);
}