coerce 0.8.6

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

use crate::actor::context::ActorContext;
use crate::actor::message::Handler;
use crate::actor::scheduler::ActorType;
use crate::actor::system::ActorSystem;
use crate::actor::{Actor, LocalActorRef};
use crate::remote::actor::message::{ClientConnected, ClientWrite, NewClient};
use crate::remote::net::client::send::Write;
use crate::remote::net::client::RemoteClient;
use crate::remote::system::NodeId;
use std::collections::hash_map::Entry;
use std::collections::HashMap;

pub struct RemoteClientRegistry {
    node_addr_registry: HashMap<String, LocalActorRef<RemoteClient>>,
    node_id_registry: HashMap<NodeId, LocalActorRef<RemoteClient>>,
}

#[async_trait]
impl Actor for RemoteClientRegistry {
    async fn stopped(&mut self, _ctx: &mut ActorContext) {
        for client in &self.node_id_registry {
            let _ = client.1.stop().await;
        }
    }
}

impl RemoteClientRegistry {
    pub async fn new(ctx: &ActorSystem) -> LocalActorRef<RemoteClientRegistry> {
        ctx.new_actor(
            "remote-client-registry",
            RemoteClientRegistry {
                node_addr_registry: HashMap::new(),
                node_id_registry: HashMap::new(),
            },
            ActorType::Tracked,
        )
        .await
        .expect("RemoteClientRegistry")
    }
}

#[async_trait]
impl Handler<NewClient> for RemoteClientRegistry {
    async fn handle(
        &mut self,
        message: NewClient,
        _ctx: &mut ActorContext,
    ) -> Option<LocalActorRef<RemoteClient>> {
        let node_addr_entry = self.node_addr_registry.entry(message.addr.clone());
        let entry = match node_addr_entry {
            Entry::Vacant(vacant_entry) => vacant_entry,
            Entry::Occupied(entry) => {
                return {
                    debug!("RemoteClient already exists for addr={}", &message.addr);
                    let client = entry.get();
                    Some(client.clone())
                }
            }
        };

        debug!("creating RemoteClient, addr={}", &message.addr);

        let client_actor =
            RemoteClient::new(message.addr.clone(), message.system, message.client_type).await;

        debug!("created RemoteClient, addr={}", &message.addr);
        entry.insert(client_actor.clone());

        Some(client_actor)
    }
}

#[async_trait]
impl Handler<ClientConnected> for RemoteClientRegistry {
    async fn handle(&mut self, message: ClientConnected, _ctx: &mut ActorContext) {
        self.node_id_registry
            .insert(message.remote_node_id, message.client_actor_ref);
    }
}

#[async_trait]
impl Handler<ClientWrite> for RemoteClientRegistry {
    async fn handle(&mut self, message: ClientWrite, _ctx: &mut ActorContext) {
        let node_id = message.0;
        let message = message.1;

        // TODO: we could open multiple clients per node and use some routing mechanism
        //       to potentially improve throughput, whilst still maintainingnStore
        //       message ordering

        if let Some(client) = self.node_id_registry.get(&node_id) {
            trace!("emitting message ({:?}) to node_id={}", &message, &node_id);
            client.notify(Write(message)).expect("send client msg");
            trace!("written data to client");
        } else {
            // TODO: should we buffer the message incase the client will eventually exist
            warn!("attempted to write message to node_id={} but no client was registered (message={:?})", &node_id, &message);
        }
    }
}