troupe 0.1.0

Library for modelling Rust applications with actors
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
157
158
159
160
161
162
163
164
165
166
167
168
169

use futures::StreamExt;
use instant::{Duration, Instant};
use tokio::sync::oneshot::error::TryRecvError;
use troupe::{compat::sleep_for, prelude::*};

#[derive(Debug, PartialEq, Eq)]
struct Started;

#[derive(Debug, Clone, PartialEq, Eq)]
struct Processed;

#[derive(Debug, PartialEq, Eq)]
struct Completed;

struct DummyStream {
    started: Option<OneshotSender<Started>>,
    completed: OneshotSender<Completed>,
}

#[async_trait]
impl ActorState for DummyStream {
    type ActorType = StreamActor;
    type Permanence = Permanent;
    type Message = Processed;
    type Output = Processed;

    async fn start_up(&mut self, _: &mut Scheduler<Self>) {
        self.started.take().unwrap().send(Started).unwrap();
    }

    async fn process(&mut self, scheduler: &mut Scheduler<Self>, msg: Self::Message) {
        scheduler.broadcast(msg);
    }

    async fn finalize(self, _: &mut Scheduler<Self>) {
        self.completed.send(Completed).unwrap();
    }
}

async fn sleep() {
    sleep_for(Duration::from_millis(10)).await
}

#[tokio::test]
async fn startup_and_teardown() {
    let (started, mut started_recv) = oneshot_channel();
    let (completed, mut comped_recv) = oneshot_channel();
    let state = DummyStream {
        started: Some(started),
        completed,
    };
    assert_eq!(Err(TryRecvError::Empty), started_recv.try_recv());
    assert_eq!(Err(TryRecvError::Empty), comped_recv.try_recv());
    let stream = futures::stream::iter(std::iter::once(Processed)).fuse();
    let mut client = ActorBuilder::new(state).launch(stream);
    /* ----- Successful startup test ----- */
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to start!!");
        }
        _ = started_recv => { }
    }
    /* ----- Processes stream test ----- */
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to process message!!");
        }
        msg = client.next() => {
            assert_eq!(Some(Ok(Processed)), msg)
        }
    }
    /* ----- Successful shutdown test ----- */
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to shutdown!!");
        }
        _ = comped_recv => { }
    }
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to process message!!");
        }
        msg = client.next() => {
            assert_eq!(None, msg)
        }
    }
}

struct Counter {
    started: Option<OneshotSender<Started>>,
    completed: OneshotSender<Completed>,
    count: usize,
}

#[async_trait]
impl ActorState for Counter {
    type ActorType = StreamActor;
    type Permanence = Transient;
    type Message = ();
    type Output = usize;

    async fn start_up(&mut self, scheduler: &mut Scheduler<Self>) {
        self.started.take().unwrap().send(Started).unwrap();
        scheduler.schedule(Instant::now() + Duration::from_millis(5), ());
    }

    async fn process(&mut self, scheduler: &mut Scheduler<Self>, (): Self::Message) {
        println!("Processing message!!");
        let next = self.count + 1;
        scheduler.broadcast(std::mem::replace(&mut self.count, next));
        scheduler.schedule(Instant::now() + Duration::from_millis(5), ());
        if self.count > 10 {
            scheduler.shutdown();
        }
    }

    async fn finalize(self, _: &mut Scheduler<Self>) {
        self.completed.send(Completed).unwrap();
    }
}

#[tokio::test]
async fn stop_after_ten_messages() {
    let (started, mut started_recv) = oneshot_channel();
    let (completed, mut comped_recv) = oneshot_channel();
    let state = Counter {
        started: Some(started),
        completed,
        count: 0,
    };
    assert_eq!(Err(TryRecvError::Empty), started_recv.try_recv());
    assert_eq!(Err(TryRecvError::Empty), comped_recv.try_recv());
    let stream = futures::stream::iter(std::iter::empty()).fuse();
    let mut builder = ActorBuilder::new(state);
    let _client = builder.client();
    let mut client = builder.launch(stream);
    /* ----- Successful startup test ----- */
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to start!!");
        }
        _ = started_recv => { }
    }
    /* ----- Processes stream test ----- */
    let mut counter = 0;
    while counter < 11 {
        let Some(Ok(num)) = client.next().await else {
            panic!("Actor closed prematurely!!")
        };
        println!("Broadcast received: {num}");
        assert_eq!(num, counter);
        counter += 1;
    }
    /* ----- Successful shutdown test ----- */
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to shutdown!!");
        }
        _ = comped_recv => { }
    }
    tokio::select! {
        _ = sleep() => {
            panic!("Actor failed to close stream!!");
        }
        msg = client.next() => {
            assert_eq!(None, msg)
        }
    }
}