elfo 0.2.0-alpha.21

An asynchronous distributed actor framework with robust observability
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

#![allow(missing_docs)]
#![cfg(feature = "test-util")]
#![allow(clippy::never_loop)]

use elfo::{
    messages::{Terminate, TerminateReason},
    prelude::*,
    TerminationPolicy,
};

#[message]
#[derive(PartialEq)]
struct BeforeExit;

#[tokio::test]
async fn it_terminates_closing_policy() {
    let blueprint = ActorGroup::new().exec(move |mut ctx| async move {
        while let Some(envelope) = ctx.recv().await {
            msg!(match envelope {
                _ => unreachable!(),
            });
        }

        ctx.send(BeforeExit).await.unwrap();
    });

    let mut proxy = elfo::test::proxy(blueprint, elfo::config::AnyConfig::default()).await;

    proxy.send(Terminate::default()).await;
    proxy.finished().await;
    assert_msg_eq!(proxy.recv().await, BeforeExit);
    proxy.sync().await;
}

#[tokio::test]
async fn it_terminates_manually_policy() {
    let blueprint = ActorGroup::new()
        .termination_policy(TerminationPolicy::manually())
        .exec(move |mut ctx| async move {
            while let Some(envelope) = ctx.recv().await {
                msg!(match envelope {
                    Terminate => {
                        ctx.send(BeforeExit).await.unwrap();
                        return;
                    }
                    _ => unreachable!(),
                });
            }
        });

    let mut proxy = elfo::test::proxy(blueprint, elfo::config::AnyConfig::default()).await;

    proxy.send(Terminate::default()).await;
    proxy.finished().await;
    assert_msg_eq!(proxy.recv().await, BeforeExit);
    proxy.sync().await;
}

#[tokio::test]
async fn it_terminates_manually_policy_via_closing_terminate() {
    let blueprint = ActorGroup::new()
        .termination_policy(TerminationPolicy::manually())
        .exec(move |mut ctx| async move {
            while let Some(envelope) = ctx.recv().await {
                msg!(match envelope {
                    Terminate => {
                        ctx.send(BeforeExit).await.unwrap();
                    }
                    _ => unreachable!(),
                });
            }

            ctx.send(BeforeExit).await.unwrap();
        });

    let mut proxy = elfo::test::proxy(blueprint, elfo::config::AnyConfig::default()).await;

    proxy.send(Terminate::default()).await;
    assert_msg_eq!(proxy.recv().await, BeforeExit);
    assert!(proxy.try_recv().await.is_none());

    proxy.send(Terminate::closing()).await;
    assert_msg_eq!(proxy.recv().await, BeforeExit);
    proxy.finished().await;
    proxy.sync().await;
}

#[tokio::test]
async fn terminate_with_reason() {
    let blueprint = ActorGroup::new()
        .termination_policy(TerminationPolicy::manually())
        .exec(move |mut ctx| async move {
            while let Some(envelope) = ctx.recv().await {
                msg!(match envelope {
                    msg @ Terminate => {
                        ctx.send(TerminateResponse(msg.reason)).await.unwrap();
                        break;
                    }
                });
            }
        });

    let mut proxy = elfo::test::proxy(blueprint, elfo::config::AnyConfig::default()).await;

    let reason = TerminateReason::custom("custom reason");
    proxy
        .send(Terminate::default().with_reason(reason.clone()))
        .await;
    assert_msg_eq!(proxy.recv().await, TerminateResponse(reason));
    assert!(proxy.try_recv().await.is_none());
}

#[message]
#[derive(PartialEq)]
struct TerminateResponse(TerminateReason);