coerce 0.8.11

Async actor runtime and distributed systems framework
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

use coerce::actor::context::ActorContext;

use coerce::actor::{Actor, ActorCreationErr, ActorFactory, ActorRecipe};

use std::collections::HashMap;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering::Relaxed;
use std::sync::Arc;

#[derive(Clone)]
pub struct ShardedActor {
    id: String,
    actor_started: Arc<AtomicBool>,
    actor_stopped: Arc<AtomicBool>,
    actor_dropped: Arc<AtomicBool>,
}

#[async_trait]
impl Actor for ShardedActor {
    async fn stopped(&mut self, _ctx: &mut ActorContext) {
        self.actor_stopped.store(true, Relaxed);
    }
}

pub struct ShardedActorRecipe {
    id: String,
}

impl ActorRecipe for ShardedActorRecipe {
    fn read_from_bytes(bytes: &Vec<u8>) -> Option<Self> {
        Some(Self {
            id: String::from_utf8(bytes.clone()).unwrap(),
        })
    }

    fn write_to_bytes(&self) -> Option<Vec<u8>> {
        Some(self.id.clone().into_bytes())
    }
}

pub struct ShardedActorFactory {
    state_store: Arc<HashMap<String, ShardedActor>>,
}

impl Clone for ShardedActorFactory {
    fn clone(&self) -> Self {
        Self {
            state_store: self.state_store.clone(),
        }
    }
}

impl ShardedActorFactory {
    pub fn new(actors: Vec<ShardedActor>) -> Self {
        Self {
            state_store: Arc::new(actors.into_iter().map(|a| (a.id.clone(), a)).collect()),
        }
    }

    pub fn assert_actor_started(&self, id: &str) {
        let actor_started = self
            .state_store
            .get(id)
            .unwrap()
            .actor_started
            .load(Relaxed);
        assert!(actor_started)
    }
    pub fn assert_actor_stopped(&self, id: &str) {
        let actor_stopped = self
            .state_store
            .get(id)
            .unwrap()
            .actor_stopped
            .load(Relaxed);
        assert!(actor_stopped)
    }

    pub fn assert_actor_dropped(&self, id: &str) {
        let actor_dropped = self
            .state_store
            .get(id)
            .unwrap()
            .actor_dropped
            .load(Relaxed);
        assert!(actor_dropped)
    }
}

#[async_trait]
impl ActorFactory for ShardedActorFactory {
    type Actor = ShardedActor;
    type Recipe = ShardedActorRecipe;

    async fn create(&self, recipe: ShardedActorRecipe) -> Result<Self::Actor, ActorCreationErr> {
        self.state_store.get(&recipe.id).cloned().ok_or_else(|| {
            ActorCreationErr::InvalidRecipe(format!(
                "no actor with id `{}` in state_store",
                &recipe.id
            ))
        })
    }
}

impl Drop for ShardedActor {
    fn drop(&mut self) {
        self.actor_dropped.store(true, Relaxed);
    }
}