linkd 0.1.3

A simple actor model framework that takes inspiration from OTP in the erlang ecosystem.
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
170

pub mod registry;

use std::{any::TypeId, marker::PhantomData, sync::Arc};

use async_trait::async_trait;
use registry::Registry;
use tokio::sync::{
    mpsc::{self, Receiver, Sender},
    oneshot,
};

// #[async_trait]
// trait ActorBehavior<State, Messages, Responses> {
//     fn new() -> Self;
//     async fn startup(&mut self, state: State);
//     async fn call(&mut self, message: Messages) -> Result<Responses, oneshot::error::RecvError>;
//     async fn cast(&mut self, message: Messages);
//     async fn fetch_from_registry(registry: Registry) -> &Self;
// }

pub struct Actor<State, Messages, Responses> {
    incoming_tx: Option<Sender<(Messages, Option<oneshot::Sender<Responses>>)>>,
    incoming_rx: Option<Receiver<(Messages, Option<oneshot::Sender<Responses>>)>>,
    context: PhantomData<State>,
}

impl<
        State: Send + 'static,
        Messages: Send + 'static + Handler<State, Responses>,
        Responses: Send + 'static,
    > Actor<State, Messages, Responses>
{
    pub fn new() -> Self {
        let (tx, rx) = mpsc::channel::<(Messages, Option<oneshot::Sender<Responses>>)>(100);
        Self {
            incoming_tx: Some(tx),
            incoming_rx: Some(rx),
            context: PhantomData,
        }
    }

    pub async fn startup(&mut self, state: State) {
        let mut rx = self
            .incoming_rx
            .take()
            .expect("You cannot startup an actor twice.");
        tokio::spawn(async move {
            let mut local_state = state;

            while let Some((m, maybe_reply_handle)) = rx.recv().await {
                let resp = m.handle(&mut local_state).await;
                if let Some(reply_handle) = maybe_reply_handle {
                    let _ = reply_handle.send(resp); // Maybe handle this result if we fail
                                                     // to reply better?
                }
            }
        });
    }

    pub async fn call(&self, message: Messages) -> Result<Responses, oneshot::error::RecvError> {
        let (tx, rx) = oneshot::channel::<Responses>();

        let _ = self
            .incoming_tx
            .as_ref()
            .expect("Actor must be started before attempting to send messages")
            .send((message, Some(tx)))
            .await;

        rx.await
    }

    pub async fn cast(&self, message: Messages) {
        let _ = self
            .incoming_tx
            .as_ref()
            .expect("Actor must be started before attempting to send messages")
            .send((message, None))
            .await;
    }
}

trait Registerable {
    fn register(self, registry: &mut Registry);
    fn fetch_from_registry(registry: &Registry) -> &Self;
}

impl<
        State: Send + Sync + 'static,
        Messages: Send + Sync + 'static + Handler<State, Responses>,
        Responses: Send + Sync + 'static,
    > Registerable for Actor<State, Messages, Responses>
{
    fn register(self, registry: &mut Registry) {
        let type_ = TypeId::of::<Actor<State, Messages, Responses>>();

        registry.add_child(type_, Arc::new(self));
    }

    fn fetch_from_registry(registry: &Registry) -> &Self {
        let type_ = TypeId::of::<Actor<State, Messages, Responses>>();
        println!("{}", std::any::type_name::<Actor<State, Messages, Responses>>());

        registry
            .get_child(type_)
            .unwrap()
            .downcast_ref::<Actor<State, Messages, Responses>>()
            .expect("Guaranteed by HashMap structure")
    }
}

#[async_trait]
pub trait Handler<State, Responses> {
    async fn handle(&self, state: &mut State) -> Responses;
}

#[cfg(test)]
mod tests {
    use crate::{Actor, Handler, registry::Registry, Registerable};
    use async_trait::async_trait;


    type TestActor = Actor::<MyState, Messages, Responses>;

    struct MyState;

    enum Messages {
        V4,
        V6,
    }

    #[derive(PartialEq)]
    enum Responses {
        V4,
        V6,
    }

    #[async_trait]
    impl Handler<MyState, Responses> for Messages {
        async fn handle(&self, _state: &mut MyState) -> Responses {
            match self {
                Messages::V4 => {
                    println!("Hello darkness my old friend");

                    Responses::V4
                }
                Messages::V6 => Responses::V6,
            }
        }
    }

    #[tokio::test]
    async fn create_actor_call_and_cast() {
        let mut registry = Registry::new();

        let mut background_worker = TestActor::new();
        background_worker.startup(MyState {}).await;
        background_worker.register(&mut registry);
        
        let local_ref = TestActor::fetch_from_registry(&registry);
        let res = local_ref.call(Messages::V4).await;
        let res_next = local_ref.call(Messages::V6).await;

        assert!(res.is_ok());
        assert!(res.unwrap() == Responses::V4);

        assert!(res_next.is_ok());
        assert!(res_next.unwrap() == Responses::V6);
    }
}