use crate::cluster::ClusterSystem;
use crate::cluster::config::ClusterConfig;
use crate::cluster::error::ClusterError;
use crate::cluster::sync::{SpawnRegistry, TypeRegistry};
use crate::lifecycle::{ActorFactory, RestartConfig, RestartPolicy};
use crate::listing::{Listing, ReceptionKey};
use crate::ready::ReadyHandle;
use crate::receptionist::{ActorEvent, Receptionist};
use crate::{Actor, Endpoint, RemoteDispatch};
pub struct System {
inner: SystemInner,
}
enum SystemInner {
Local { receptionist: Receptionist },
Clustered { cluster: Box<ClusterSystem> },
}
impl System {
pub fn local() -> Self {
Self {
inner: SystemInner::Local {
receptionist: Receptionist::new(),
},
}
}
pub async fn clustered(
config: ClusterConfig,
type_registry: TypeRegistry,
spawn_registry: SpawnRegistry,
) -> Result<Self, ClusterError> {
let cluster = ClusterSystem::start(config, type_registry, spawn_registry).await?;
Ok(Self {
inner: SystemInner::Clustered {
cluster: Box::new(cluster),
},
})
}
pub async fn clustered_auto(config: ClusterConfig) -> Result<Self, ClusterError> {
Self::clustered(config, TypeRegistry::from_auto(), SpawnRegistry::new()).await
}
pub fn start<A>(&self, label: &str, actor: A, state: A::State) -> Endpoint<A>
where
A: Actor + RemoteDispatch + 'static,
{
self.receptionist().start(label, actor, state)
}
pub fn start_public<A>(&self, label: &str, actor: A, state: A::State) -> Endpoint<A>
where
A: Actor + RemoteDispatch + 'static,
{
self.receptionist().start_public(label, actor, state)
}
pub fn start_private<A>(&self, label: &str, actor: A, state: A::State) -> Endpoint<A>
where
A: Actor + RemoteDispatch + 'static,
{
self.receptionist().start_private(label, actor, state)
}
pub fn prepare<A>(
&self,
label: &str,
actor: A,
state: A::State,
) -> (Endpoint<A>, ReadyHandle<A>)
where
A: Actor + RemoteDispatch + 'static,
{
self.receptionist().prepare(label, actor, state)
}
pub fn start_with_policy<F: ActorFactory>(
&self,
label: &str,
factory: F,
policy: RestartPolicy,
) -> Endpoint<F::Actor>
where
F::Actor: RemoteDispatch,
{
self.receptionist()
.start_with_policy(label, factory, policy)
}
pub fn start_with_config<F: ActorFactory>(
&self,
label: &str,
factory: F,
config: RestartConfig,
) -> Endpoint<F::Actor>
where
F::Actor: RemoteDispatch,
{
self.receptionist()
.start_with_config(label, factory, config)
}
pub fn lookup<A: Actor + 'static>(&self, label: &str) -> Option<Endpoint<A>> {
self.receptionist().lookup(label)
}
pub fn check_in<A: Actor + 'static>(&self, label: &str, key: ReceptionKey<A>) {
self.receptionist().check_in(label, key);
}
pub fn listing<A: Actor + 'static>(&self, key: ReceptionKey<A>) -> Listing<A> {
self.receptionist().listing(key)
}
pub fn subscribe_events(&self) -> tokio::sync::mpsc::UnboundedReceiver<ActorEvent> {
self.receptionist().subscribe_events()
}
pub fn stop(&self, label: &str) {
self.receptionist().stop(label);
}
pub async fn shutdown(&self) {
match &self.inner {
SystemInner::Local { .. } => {}
SystemInner::Clustered { cluster } => {
cluster.shutdown().await;
}
}
}
pub fn receptionist(&self) -> &Receptionist {
match &self.inner {
SystemInner::Local { receptionist } => receptionist,
SystemInner::Clustered { cluster } => cluster.receptionist(),
}
}
pub fn is_clustered(&self) -> bool {
matches!(self.inner, SystemInner::Clustered { .. })
}
pub fn cluster_system(&self) -> Option<&ClusterSystem> {
match &self.inner {
SystemInner::Local { .. } => None,
SystemInner::Clustered { cluster } => Some(cluster),
}
}
pub fn local_addr(&self) -> Option<std::net::SocketAddr> {
match &self.inner {
SystemInner::Local { .. } => None,
SystemInner::Clustered { cluster } => Some(cluster.local_addr()),
}
}
}