use std::collections::HashMap; use std::sync::{Arc, Mutex}; use futures::future::BoxFuture; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use uuid::Uuid; #[derive(Debug, Clone)] pub struct Mailbox { tx: mpsc::UnboundedSender, } impl Mailbox { pub fn send(&self, msg: M) -> bool { self.tx.send(msg).is_ok() } pub fn is_closed(&self) -> bool { self.tx.is_closed() } } type SpawnFn = dyn Fn(Arc>, Uuid, mpsc::UnboundedReceiver) -> BoxFuture<'static, ()> + Send + Sync; pub struct ActorRegistry { name: &'static str, actors: Mutex>>, spawn: Arc>, shutdown: CancellationToken, } impl std::fmt::Debug for ActorRegistry { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let live = self.actors.lock().unwrap().len(); f.debug_struct("ActorRegistry") .field("name", &self.name) .field("live_actors", &live) .finish() } } impl ActorRegistry where M: Send + 'static, { pub fn new(name: &'static str, spawn: F, shutdown: CancellationToken) -> Arc where F: Fn(Arc, Uuid, mpsc::UnboundedReceiver) -> Fut + Send + Sync + 'static, Fut: Future + Send + 'static, { let spawn: Arc> = Arc::new(move |reg, key, rx| Box::pin(spawn(reg, key, rx))); Arc::new(Self { name, actors: Mutex::new(HashMap::new()), spawn, shutdown, }) } pub fn shutdown_token(&self) -> CancellationToken { self.shutdown.clone() } pub fn live_count(&self) -> usize { self.actors.lock().unwrap().len() } pub fn dispatch(self: &Arc, key: Uuid, mut msg: M) { { let guard = self.actors.lock().unwrap(); if let Some(tx) = guard.get(&key) { match tx.send(msg) { Ok(()) => return, Err(e) => msg = e.0, } } } let (tx, rx) = mpsc::unbounded_channel(); let mailbox = Mailbox { tx: tx.clone() }; { let mut guard = self.actors.lock().unwrap(); if let Some(existing) = guard.get(&key) { match existing.send(msg) { Ok(()) => return, Err(e) => msg = e.0, } } guard.insert(key, tx); } let registry = self.clone(); let spawn = self.spawn.clone(); tokio::spawn(async move { let fut = (spawn)(registry.clone(), key, rx); fut.await; registry.actors.lock().unwrap().remove(&key); }); let _ = mailbox.send(msg); crate::metrics::actor_spawned(self.name); } } pub async fn run_actor( mut rx: mpsc::UnboundedReceiver, shutdown: CancellationToken, mut apply: impl FnMut(M) -> BoxFuture<'static, ()>, ) { loop { tokio::select! { _ = shutdown.cancelled() => break, maybe = rx.recv() => { match maybe { Some(msg) => apply(msg).await, None => break, } } } } }