speare 0.4.3

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
143
144
145
146
147
148
149
150
151
152
153

mod sync_vec;

use speare::{Actor, Ctx, Node, SourceSet, Sources};
use std::time::Duration;
use sync_vec::SyncVec;
use tokio::time;

#[derive(Debug, PartialEq, Clone)]
enum Msg {
    Tick,
    Manual(u32),
}

struct Ticker;

impl Actor for Ticker {
    type Props = SyncVec<Msg>;
    type Msg = Msg;
    type Err = ();

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

    async fn sources(&self, _ctx: &Ctx<Self>) -> Result<impl Sources<Self>, Self::Err> {
        Ok(SourceSet::new().interval(time::interval(Duration::from_millis(10)), || Msg::Tick))
    }

    async fn handle(&mut self, msg: Self::Msg, ctx: &mut Ctx<Self>) -> Result<(), Self::Err> {
        ctx.props().push(msg).await;
        Ok(())
    }
}

#[tokio::test]
async fn interval_sends_ticks_to_actor() {
    // Arrange
    let mut node = Node::default();
    let recvd: SyncVec<Msg> = Default::default();
    let handle = node.actor::<Ticker>(recvd.clone()).spawn();

    // Act
    time::sleep(Duration::from_millis(35)).await;
    handle.stop();
    time::sleep(Duration::from_millis(1)).await;

    // Assert
    let msgs = recvd.clone_vec().await;
    assert!(
        msgs.len() >= 3,
        "expected at least 3 ticks, got {}",
        msgs.len()
    );
    assert!(msgs.iter().all(|m| *m == Msg::Tick));
}

#[tokio::test]
async fn interval_and_handle_messages_both_received() {
    // Arrange
    let mut node = Node::default();
    let recvd: SyncVec<Msg> = Default::default();
    let handle = node.actor::<Ticker>(recvd.clone()).spawn();

    // Act
    time::sleep(Duration::from_millis(25)).await;
    handle.send(Msg::Manual(1));
    time::sleep(Duration::from_millis(1)).await;
    handle.stop();
    time::sleep(Duration::from_millis(1)).await;

    // Assert
    let msgs = recvd.clone_vec().await;
    assert!(msgs.contains(&Msg::Manual(1)));
    assert!(msgs.contains(&Msg::Tick));
}

#[tokio::test]
async fn multiple_intervals() {
    // Arrange
    #[derive(Debug, PartialEq, Clone)]
    enum MultiMsg {
        Fast,
        Slow,
    }

    struct Multi;

    impl Actor for Multi {
        type Props = SyncVec<MultiMsg>;
        type Msg = MultiMsg;
        type Err = ();

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

        async fn sources(&self, _ctx: &Ctx<Self>) -> Result<impl Sources<Self>, Self::Err> {
            Ok(SourceSet::new()
                .interval(time::interval(Duration::from_millis(10)), || MultiMsg::Fast)
                .interval(time::interval(Duration::from_millis(30)), || MultiMsg::Slow))
        }

        async fn handle(&mut self, msg: Self::Msg, ctx: &mut Ctx<Self>) -> Result<(), Self::Err> {
            ctx.props().push(msg).await;
            Ok(())
        }
    }

    let mut node = Node::default();
    let recvd: SyncVec<MultiMsg> = Default::default();
    let handle = node.actor::<Multi>(recvd.clone()).spawn();

    // Act
    time::sleep(Duration::from_millis(65)).await;
    handle.stop();
    time::sleep(Duration::from_millis(1)).await;

    // Assert
    let msgs = recvd.clone_vec().await;
    let fast_count = msgs.iter().filter(|m| **m == MultiMsg::Fast).count();
    let slow_count = msgs.iter().filter(|m| **m == MultiMsg::Slow).count();
    assert!(
        fast_count >= 5,
        "expected at least 5 fast ticks, got {}",
        fast_count
    );
    assert!(
        slow_count >= 2,
        "expected at least 2 slow ticks, got {}",
        slow_count
    );
}

#[tokio::test]
async fn interval_stops_when_actor_stops() {
    // Arrange
    let mut node = Node::default();
    let recvd: SyncVec<Msg> = Default::default();
    let handle = node.actor::<Ticker>(recvd.clone()).spawn();

    // Act
    time::sleep(Duration::from_millis(25)).await;
    handle.stop();
    time::sleep(Duration::from_millis(1)).await;
    let count_at_stop = recvd.clone_vec().await.len();

    time::sleep(Duration::from_millis(50)).await;
    let count_after_wait = recvd.clone_vec().await.len();

    // Assert
    assert!(!handle.is_alive());
    assert_eq!(count_at_stop, count_after_wait);
}