elfo 0.1.40

Your next actor system
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

#![cfg(feature = "test-util")]

use std::convert::TryFrom;

use elfo::{config::AnyConfig, prelude::*, scope, stream, trace_id::TraceId};

#[message]
#[derive(PartialEq)]
struct SomeMessage(u32);

#[message]
struct Set(Vec<u32>);

#[message]
struct Replace(Vec<u32>);

#[tokio::test]
async fn it_handles_basic_operations() {
    let group = ActorGroup::new().exec(|ctx| async move {
        let stream = stream::Stream::new(futures::stream::iter(vec![SomeMessage(0)]));

        let mut ctx = ctx.with(&stream);
        let mut prev_trace_id = scope::trace_id();

        while let Some(envelope) = ctx.recv().await {
            assert_ne!(scope::trace_id(), prev_trace_id);
            prev_trace_id = scope::trace_id();

            msg!(match envelope {
                m @ SomeMessage(_) => ctx.send(m).await.unwrap(),
                Set(d) => stream.set(futures::stream::iter(
                    d.into_iter().map(SomeMessage).collect::<Vec<_>>()
                )),
                Replace(d) => {
                    let _ = stream.replace(futures::stream::iter(
                        d.into_iter().map(SomeMessage).collect::<Vec<_>>(),
                    ));
                }
            })
        }
    });

    let mut proxy = elfo::test::proxy(group, AnyConfig::default()).await;
    assert_msg_eq!(proxy.recv().await, SomeMessage(0));
    assert!(proxy.try_recv().is_none());

    proxy.send(Set((1..5).collect())).await;
    for i in 1..5 {
        assert_msg_eq!(proxy.recv().await, SomeMessage(i));
    }
    assert!(proxy.try_recv().is_none());

    proxy.send(Replace((6..8).collect())).await;
    for i in 6..8 {
        assert_msg_eq!(proxy.recv().await, SomeMessage(i));
    }
}

#[tokio::test]
async fn it_restores_trace_id() {
    let group = ActorGroup::new().exec(|ctx| async move {
        let stream = stream::Stream::new(futures::stream::iter(vec![
            (TraceId::try_from(5).unwrap(), SomeMessage(5)),
            (TraceId::try_from(6).unwrap(), SomeMessage(6)),
        ]));

        let mut ctx = ctx.with(&stream);

        while let Some(envelope) = ctx.recv().await {
            msg!(match envelope {
                SomeMessage(x) => {
                    assert_eq!(u64::from(scope::trace_id()), u64::from(x));
                    ctx.send(SomeMessage(x)).await.unwrap()
                }
            })
        }
    });

    let mut proxy = elfo::test::proxy(group, AnyConfig::default()).await;
    assert_msg_eq!(proxy.recv().await, SomeMessage(5));
    assert_msg_eq!(proxy.recv().await, SomeMessage(6));
}

#[message]
#[derive(PartialEq)]
struct SomeMessage2(u32);

#[tokio::test]
async fn it_generates_stream() {
    let group = ActorGroup::new().exec(|ctx| async move {
        let stream = stream::Stream::generate(|mut y| async move {
            y.emit(SomeMessage(1)).await;
            y.emit(SomeMessage2(2)).await;
        });

        let mut ctx = ctx.with(&stream);

        while let Some(envelope) = ctx.recv().await {
            msg!(match envelope {
                SomeMessage(x) => {
                    ctx.send(SomeMessage(x)).await.unwrap()
                }
                SomeMessage2(x) => {
                    ctx.send(SomeMessage2(x)).await.unwrap()
                }
            })
        }
    });

    let mut proxy = elfo::test::proxy(group, AnyConfig::default()).await;
    assert_msg_eq!(proxy.recv().await, SomeMessage(1));
    assert_msg_eq!(proxy.recv().await, SomeMessage2(2));
}