ractor_actors 0.5.0

Helpful actors built with Ractor
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

// Copyright (c) Sean Lawlor
//
// This source code is licensed under both the MIT license found in the
// LICENSE-MIT file in the root directory of this source tree.

use ractor::call;
use ractor::concurrency::Duration;
use ractor::Actor;
use ractor::ActorProcessingErr;
use ractor::ActorRef;
use ractor::RpcReplyPort;
use tokio_stream as stream;

use crate::common_test::periodic_async_check;

use super::spawn_stream_pump;

struct StreamActor;

enum StreamActorMessage {
    GetCount(RpcReplyPort<u64>),
    Add(u64),
}

#[cfg_attr(feature = "async-trait", async_trait::async_trait)]
impl Actor for StreamActor {
    type Msg = StreamActorMessage;
    type State = u64;
    type Arguments = ();

    async fn pre_start(
        &self,
        myself: ActorRef<Self::Msg>,
        _: Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        // just for the test, drop the cell
        let _ = spawn_stream_pump(
            stream::iter(1u64..=500u64),
            myself.clone(),
            |a| {
                if a.is_some() {
                    StreamActorMessage::Add(1)
                } else {
                    StreamActorMessage::Add(0)
                }
            },
            None,
        )
        .await?;

        Ok(0)
    }

    async fn handle(
        &self,
        _: ActorRef<Self::Msg>,
        message: Self::Msg,
        state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        match message {
            StreamActorMessage::GetCount(reply) => {
                println!("Received count request");
                let _ = reply.send(*state);
            }
            StreamActorMessage::Add(i) => {
                *state += i;
            }
        }
        Ok(())
    }

    async fn handle_supervisor_evt(
        &self,
        _myself: ActorRef<Self::Msg>,
        _message: ractor::SupervisionEvent,
        _state: &mut Self::State,
    ) -> Result<(), ActorProcessingErr> {
        // ignore child fails (stream exits)
        Ok(())
    }
}

#[ractor::concurrency::test]
#[tracing_test::traced_test]
async fn test_streaming_operation() {
    // Setup
    // Create the actor
    let (actor, handle) = Actor::spawn(None, StreamActor, ())
        .await
        .expect("Failed to spawn non-blocking actor tree");

    periodic_async_check(
        || async {
            // get the count
            let reply = call!(actor, StreamActorMessage::GetCount).expect("Failed to get count");
            reply >= 1
        },
        Duration::from_secs(3),
    )
    .await;

    // Cleanup
    actor.stop(None);
    handle.await.unwrap();
}