atomr_core/actor/
actor_system.rs

1//! `ActorSystem` — root of the actor hierarchy. akka.net: `Actor/ActorSystem.cs`.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5
6use atomr_config::Config;
7use parking_lot::Mutex;
8use thiserror::Error;
9use tokio::sync::{mpsc, Notify};
10
11use super::actor_cell::{spawn_cell, ChildEntry, SystemMsg};
12use super::actor_ref::{ActorRef, UntypedActorRef};
13use super::address::Address;
14use super::extensions::Extensions;
15use super::observer::{DeadLetterObserver, SpawnObserver};
16use super::path::ActorPath;
17use super::props::Props;
18use super::remote::RemoteProvider;
19use super::scheduler::{Scheduler, TokioScheduler};
20use super::traits::Actor;
21
22pub(crate) struct ActorSystemInner {
23    pub name: String,
24    pub config: Config,
25    pub address: Address,
26    pub scheduler: Arc<dyn Scheduler>,
27    pub extensions: Extensions,
28    pub user_guardian: Mutex<HashMap<String, ChildEntry>>,
29    pub(crate) spawn_observer: parking_lot::RwLock<Option<Arc<dyn SpawnObserver>>>,
30    pub(crate) dead_letter_observer: parking_lot::RwLock<Option<Arc<dyn DeadLetterObserver>>>,
31    pub(crate) remote_provider: parking_lot::RwLock<Option<Arc<dyn RemoteProvider>>>,
32    terminated: Notify,
33    terminated_flag: std::sync::atomic::AtomicBool,
34}
35
36/// Public handle to the actor system.
37#[derive(Clone)]
38pub struct ActorSystem {
39    pub(crate) inner: Arc<ActorSystemInner>,
40}
41
42impl ActorSystem {
43    /// Create an actor system with the given name and configuration.
44    pub async fn create(name: impl Into<String>, config: Config) -> Result<Self, ActorSystemError> {
45        let name = name.into();
46        let address = Address::local(&name);
47        let inner = Arc::new(ActorSystemInner {
48            name,
49            config,
50            address,
51            scheduler: Arc::new(TokioScheduler::new()),
52            extensions: Extensions::default(),
53            user_guardian: Mutex::new(HashMap::new()),
54            spawn_observer: parking_lot::RwLock::new(None),
55            dead_letter_observer: parking_lot::RwLock::new(None),
56            remote_provider: parking_lot::RwLock::new(None),
57            terminated: Notify::new(),
58            terminated_flag: std::sync::atomic::AtomicBool::new(false),
59        });
60        Ok(Self { inner })
61    }
62
63    pub fn name(&self) -> &str {
64        &self.inner.name
65    }
66
67    pub fn address(&self) -> &Address {
68        &self.inner.address
69    }
70
71    pub fn config(&self) -> &Config {
72        &self.inner.config
73    }
74
75    pub fn scheduler(&self) -> Arc<dyn Scheduler> {
76        self.inner.scheduler.clone()
77    }
78
79    pub fn extensions(&self) -> &Extensions {
80        &self.inner.extensions
81    }
82
83    /// Install a [`SpawnObserver`]. Only one observer may be installed;
84    /// subsequent calls replace the previous one. This is the hook used by
85    /// `atomr-telemetry` to populate its actor registry.
86    pub fn set_spawn_observer(&self, obs: Arc<dyn SpawnObserver>) {
87        *self.inner.spawn_observer.write() = Some(obs);
88    }
89
90    /// Install a [`DeadLetterObserver`] that is notified when a `tell`
91    /// fails because the target has stopped.
92    pub fn set_dead_letter_observer(&self, obs: Arc<dyn DeadLetterObserver>) {
93        *self.inner.dead_letter_observer.write() = Some(obs);
94    }
95
96    /// Install the remote provider. Done by `atomr-remote::RemoteSystemExt::enable_remote`.
97    /// Replaces any previous provider.
98    pub fn set_remote_provider(&self, provider: Arc<dyn RemoteProvider>) {
99        *self.inner.remote_provider.write() = Some(provider);
100    }
101
102    /// True if a remote provider is installed and the address has global scope.
103    pub fn is_remote_path(&self, path: &ActorPath) -> bool {
104        path.address.has_global_scope() && self.inner.remote_provider.read().is_some()
105    }
106
107    /// Look up an actor by full path string. Local paths return the local
108    /// child if it exists; remote paths consult the installed remote provider.
109    /// akka.net: `ActorSystem.ActorSelection`.
110    pub fn actor_selection(&self, path_str: &str) -> Option<UntypedActorRef> {
111        let path = parse_actor_path(path_str)?;
112        if path.address.has_local_scope() || path.address == self.inner.address {
113            // Local: best-effort look-up among top-level user actors.
114            if path.elements.len() >= 2 && path.elements[0].as_str() == "user" {
115                let name = path.elements[1].as_str();
116                let g = self.inner.user_guardian.lock();
117                return g.get(name).map(|c| c.untyped.clone());
118            }
119            return None;
120        }
121        let provider = self.inner.remote_provider.read().clone()?;
122        let handle = provider.resolve(&path)?;
123        Some(UntypedActorRef::from_remote(handle))
124    }
125
126    /// Resolve a remote path and produce a *typed* `ActorRef<M>`. The caller
127    /// supplies a serializer closure for `M`; `atomr-remote::RemoteSystem`
128    /// provides a default that uses bincode + `type_name::<M>()`.
129    pub fn actor_selection_with<M>(
130        &self,
131        path_str: &str,
132        serialize: Arc<dyn Fn(M, Option<ActorPath>) -> super::remote::SerializedMessage + Send + Sync>,
133    ) -> Option<ActorRef<M>>
134    where
135        M: Send + 'static,
136    {
137        let path = parse_actor_path(path_str)?;
138        if path.address.has_local_scope() || path.address == self.inner.address {
139            return None;
140        }
141        let provider = self.inner.remote_provider.read().clone()?;
142        let handle = provider.resolve(&path)?;
143        Some(ActorRef::from_remote(handle, serialize))
144    }
145
146    /// Spawn a top-level actor under `/user`. akka.net: `ActorOf`.
147    pub fn actor_of<A: Actor>(
148        &self,
149        props: Props<A>,
150        name: &str,
151    ) -> Result<ActorRef<A::Msg>, ActorSystemError> {
152        let root = ActorPath::root(self.inner.address.clone());
153        let parent = root.child("user");
154        let path = parent.child(name);
155        let mut guardian = self.inner.user_guardian.lock();
156        if guardian.contains_key(name) {
157            return Err(ActorSystemError::NameTaken(name.into()));
158        }
159        let r = spawn_cell::<A>(self.inner.clone(), props, path.clone())
160            .map_err(|e| ActorSystemError::Spawn(e.to_string()))?;
161        if let Some(obs) = self.inner.spawn_observer.read().as_ref() {
162            obs.on_spawn(&path, Some(&parent), std::any::type_name::<A>());
163        }
164        guardian.insert(
165            name.to_string(),
166            ChildEntry { path, untyped: r.as_untyped(), system_tx: r.system_sender() },
167        );
168        Ok(r)
169    }
170
171    /// Stop a top-level actor by name.
172    pub fn stop(&self, name: &str) {
173        if let Some(c) = self.inner.user_guardian.lock().get(name) {
174            let _ = c.system_tx.send(SystemMsg::Stop);
175        }
176    }
177
178    /// Initiate orderly shutdown. Awaits actor termination best-effort.
179    pub async fn terminate(&self) {
180        {
181            let guardian = self.inner.user_guardian.lock();
182            for (_, c) in guardian.iter() {
183                let _ = c.system_tx.send(SystemMsg::Stop);
184            }
185        }
186        self.inner.terminated_flag.store(true, std::sync::atomic::Ordering::Release);
187        self.inner.terminated.notify_waiters();
188        // Give in-flight tasks a moment to unwind.
189        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
190    }
191
192    pub async fn when_terminated(&self) {
193        if self.inner.terminated_flag.load(std::sync::atomic::Ordering::Acquire) {
194            return;
195        }
196        self.inner.terminated.notified().await;
197    }
198}
199
200#[derive(Debug, Error)]
201pub enum ActorSystemError {
202    #[error("top-level actor name `{0}` already taken")]
203    NameTaken(String),
204    #[error("failed to spawn actor: {0}")]
205    Spawn(String),
206    #[error("system already terminated")]
207    Terminated,
208}
209
210// Keep channel import referenced to avoid unused imports in stub paths.
211#[allow(dead_code)]
212type _SysChan = mpsc::UnboundedSender<SystemMsg>;
213
214/// Parse a string like `akka.tcp://Sys@host:port/user/foo/bar` into an
215/// `ActorPath`. Returns `None` on malformed input.
216fn parse_actor_path(s: &str) -> Option<ActorPath> {
217    let (addr_part, path_part) = split_addr_path(s)?;
218    let address = Address::parse(addr_part)?;
219    let mut path = ActorPath::root(address);
220    for seg in path_part.split('/').filter(|s| !s.is_empty()) {
221        // Strip optional `#uid` suffix on the leaf segment.
222        if let Some((name, uid)) = seg.split_once('#') {
223            let uid_n: u64 = uid.parse().ok()?;
224            path = path.child(name).with_uid(uid_n);
225        } else {
226            path = path.child(seg);
227        }
228    }
229    Some(path)
230}
231
232fn split_addr_path(s: &str) -> Option<(&str, &str)> {
233    // Address always contains `://`. The path starts at the next `/` *after*
234    // the host:port section.
235    let scheme_end = s.find("://")?;
236    let after_scheme = &s[scheme_end + 3..];
237    // The address ends at the first `/` in the after-scheme section.
238    if let Some(slash) = after_scheme.find('/') {
239        let split = scheme_end + 3 + slash;
240        Some((&s[..split], &s[split..]))
241    } else {
242        Some((s, ""))
243    }
244}
atomr_core/actor/actor_system.rs

atomr_core/actor/
actor_system.rs
