yaaf 0.2.0

Yet Another Actor 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

use crate::{
    actor::{Actor, ActorAddress, ActorId},
    context::Context,
    error::SystemError,
    message::detail::MessageList,
    router::{ConcreteRouter, Router, SysRouter, SystemMessage},
    source::{Source, SourceMeta},
};
use ::std::{any::TypeId, collections::HashMap, sync::Arc};
use ::tokio::{spawn, sync::mpsc::Receiver};

pub struct System {
    next_id: u32,
    routers: HashMap<TypeId, Arc<dyn Router>>,
    sys_router: SysRouter,
    done: Vec<Receiver<()>>,
}

impl System {
    pub async fn new() -> Result<Self, SystemError> {
        Ok(System {
            next_id: 0,
            routers: HashMap::new(),
            sys_router: ConcreteRouter::<SystemMessage>::new_system_router()
                .await
                .map_err(|source| SystemError::CreateError { source })?,
            done: Vec::new(),
        })
    }

    fn get_id(&mut self) -> ActorId {
        let result = ActorId(self.next_id);
        self.next_id += 1;
        result
    }

    pub async fn add_actor<A: Actor>(&mut self, actor: A) -> Result<ActorAddress<A>, SystemError> {
        let publish_routers = A::Publishes::setup_routers(&self.sys_router, &mut self.routers)
            .await
            .map_err(|source| SystemError::AddActorFailure { source })?;
        let handle_routers = A::Handles::setup_routers(&self.sys_router, &mut self.routers)
            .await
            .map_err(|source| SystemError::AddActorFailure { source })?;

        let actor_id = self.get_id();
        self.done.extend(
            actor
                .setup_mailboxes(
                    actor_id,
                    &handle_routers,
                    &publish_routers,
                    &self.sys_router,
                )
                .await
                .map_err(|source| SystemError::AddActorFailure { source })?,
        );

        Ok(ActorAddress::new(actor_id, handle_routers))
    }

    pub async fn add_source<S: 'static + Source + SourceMeta>(
        &mut self,
        source: S,
    ) -> Result<(), SystemError> {
        let publish_routers = S::Publishes::setup_routers(&self.sys_router, &mut self.routers)
            .await
            .map_err(|source| SystemError::AddSourceFailure { source })?;
        let ctx = Context::new(publish_routers);
        spawn(source.run(ctx));
        Ok(())
    }

    pub async fn shutdown(&mut self) -> Result<(), SystemError> {
        self.sys_router
            .broadcast(SystemMessage::Shutdown)
            .map_err(|source| SystemError::ShutdownError { source })?;
        for r in &mut self.done {
            // TODO: log this error
            let _ = r.recv().await;
        }
        Ok(())
    }
}