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
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

use chrono::{DateTime, Utc};
use std::sync::atomic::AtomicI64;
use std::sync::Arc;

use crate::actor::system::ActorSystem;
use crate::actor::LocalActorRef;
use crate::remote::actor::{
    clients::RemoteClientRegistry, registry::RemoteRegistry, RemoteHandler,
};
use crate::remote::cluster::builder::client::ClusterClientBuilder;
use crate::remote::cluster::builder::worker::ClusterWorkerBuilder;
use crate::remote::cluster::discovery::NodeDiscovery;
use crate::remote::heartbeat::Heartbeat;
use crate::remote::stream::mediator::StreamMediator;
use crate::remote::system::builder::RemoteActorSystemBuilder;

pub mod actor;
pub mod builder;
pub mod cluster;
pub mod raft;
pub mod rpc;

use crate::remote::config::RemoteSystemConfig;
pub use actor::*;
pub use cluster::*;
pub use rpc::*;

#[derive(Clone)]
pub struct RemoteActorSystem {
    inner: Arc<RemoteSystemCore>,
}

pub type NodeId = u64;
pub type AtomicNodeId = AtomicI64;

#[derive(Clone)]
pub struct RemoteSystemCore {
    node_id: NodeId,
    inner: ActorSystem,
    started_at: DateTime<Utc>,
    handler_ref: Arc<parking_lot::Mutex<RemoteHandler>>,
    registry_ref: LocalActorRef<RemoteRegistry>,
    clients_ref: LocalActorRef<RemoteClientRegistry>,
    discovery_ref: LocalActorRef<NodeDiscovery>,
    heartbeat_ref: LocalActorRef<Heartbeat>,
    mediator_ref: Option<LocalActorRef<StreamMediator>>,
    config: Arc<RemoteSystemConfig>,
    current_leader: Arc<AtomicNodeId>,
}

impl RemoteActorSystem {
    pub async fn shutdown(&self) {
        self.inner.shutdown().await;
    }
}

impl RemoteSystemCore {
    pub async fn shutdown(&self) {
        let _ = self.heartbeat_ref.stop().await;
        let _ = self.clients_ref.stop().await;

        if let Some(mediator_ref) = self.mediator_ref.as_ref() {
            let _ = mediator_ref.stop().await;
        }

        let _ = self.discovery_ref.stop().await;
        let _ = self.registry_ref.stop().await;

        info!("shutdown complete");
    }
}

impl RemoteActorSystem {
    pub fn builder() -> RemoteActorSystemBuilder {
        RemoteActorSystemBuilder::new()
    }

    pub fn cluster_worker(self) -> ClusterWorkerBuilder {
        ClusterWorkerBuilder::new(self)
    }

    pub fn cluster_client(self) -> ClusterClientBuilder {
        ClusterClientBuilder::new(self)
    }

    pub fn config(&self) -> &RemoteSystemConfig {
        &self.inner.config
    }

    pub fn node_tag(&self) -> &str {
        self.inner.config.node_tag()
    }

    pub fn node_version(&self) -> &str {
        self.inner.config.node_version()
    }

    pub fn node_id(&self) -> NodeId {
        self.inner.node_id
    }

    pub fn started_at(&self) -> &DateTime<Utc> {
        &self.inner.started_at
    }

    pub fn heartbeat(&self) -> &LocalActorRef<Heartbeat> {
        &self.inner.heartbeat_ref
    }

    pub fn registry(&self) -> &LocalActorRef<RemoteRegistry> {
        &self.inner.registry_ref
    }

    pub fn client_registry(&self) -> &LocalActorRef<RemoteClientRegistry> {
        &self.inner.clients_ref
    }

    pub fn node_discovery(&self) -> &LocalActorRef<NodeDiscovery> {
        &self.inner.discovery_ref
    }

    pub fn stream_mediator(&self) -> Option<&LocalActorRef<StreamMediator>> {
        self.inner.mediator_ref.as_ref()
    }

    pub fn actor_system(&self) -> &ActorSystem {
        &self.inner.actor_system()
    }
}

impl RemoteSystemCore {
    pub fn actor_system(&self) -> &ActorSystem {
        &self.inner
    }
}