use std::any::Any;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
use parking_lot::RwLock;
use crossbeam_channel::{Receiver, Sender};
use crate::engine::EngineInner;
use crate::message::Message;
use crate::metrics::ActorMetrics;
pub(crate) type StateStore = Arc<RwLock<HashMap<String, Box<dyn Any + Send + Sync>>>>;
#[derive(Clone)]
pub struct State<T: Clone + Send + Sync + 'static> {
key: String,
store: StateStore,
_phantom: PhantomData<T>,
}
impl<T: Clone + Send + Sync + 'static> State<T> {
pub fn get(&self) -> T {
let store = self.store.read();
store.get(&self.key)
.and_then(|v| v.downcast_ref::<T>())
.cloned()
.unwrap_or_else(|| panic!("state '{}' not found", self.key))
}
pub fn set(&self, value: T) {
let mut store = self.store.write();
store.insert(self.key.clone(), Box::new(value));
}
pub fn update<F>(&self, f: F)
where F: FnOnce(T) -> T,
{
let mut store = self.store.write();
let current = store.get(&self.key)
.and_then(|v| v.downcast_ref::<T>())
.cloned();
if let Some(c) = current {
store.insert(self.key.clone(), Box::new(f(c)));
}
}
}
pub struct Scratch<T> {
val: Option<T>,
}
impl<T> Scratch<T> {
pub fn new(val: T) -> Self { Self { val: Some(val) } }
pub fn get(&self) -> Option<&T> { self.val.as_ref() }
pub fn set(&mut self, val: T) { self.val = Some(val); }
pub fn take(&mut self) -> Option<T> { self.val.take() }
}
pub struct Context {
#[allow(dead_code)]
pub(crate) id: u64,
#[allow(dead_code)]
pub(crate) name: String,
pub(crate) state_store: StateStore,
pub(crate) rx: Receiver<Message>,
#[allow(dead_code)]
pub(crate) tx: Sender<Message>,
pub(crate) message_handler: Option<Arc<dyn Fn(Message) + Send + Sync>>,
pub(crate) panic_handler: Option<Arc<dyn Fn() + Send + Sync>>,
pub(crate) mount_handler: Option<Arc<dyn Fn() + Send + Sync>>,
pub(crate) unmount_handler: Option<Arc<dyn Fn() + Send + Sync>>,
pub(crate) engine: Arc<EngineInner>,
pub(crate) metrics: ActorMetrics,
pub(crate) is_first_mount: bool,
}
impl Context {
pub(crate) fn new(
id: u64, name: String, state_store: StateStore,
rx: Receiver<Message>, tx: Sender<Message>,
engine: Arc<EngineInner>,
) -> Self {
Self {
id, name, state_store, rx, tx,
message_handler: None,
panic_handler: None,
mount_handler: None,
unmount_handler: None,
engine,
metrics: ActorMetrics::new(),
is_first_mount: true,
}
}
pub fn use_state<T: Clone + Send + Sync + 'static>(&self, key: &str, initial: T) -> State<T> {
{
let mut store = self.state_store.write();
if !store.contains_key(key) {
store.insert(key.to_string(), Box::new(initial));
}
}
State { key: key.to_string(), store: self.state_store.clone(), _phantom: PhantomData }
}
pub fn use_scratch<T>(&self, initial: T) -> Scratch<T> {
Scratch::new(initial)
}
pub fn on_message<F>(&mut self, f: F)
where F: Fn(Message) + Send + Sync + 'static,
{
self.message_handler = Some(Arc::new(f));
}
pub fn on_panic<F>(&mut self, f: F)
where F: Fn() + Send + Sync + 'static,
{
self.panic_handler = Some(Arc::new(f));
}
pub fn on_mount<F>(&mut self, f: F)
where F: Fn() + Send + Sync + 'static,
{
self.mount_handler = Some(Arc::new(f));
}
pub fn on_unmount<F>(&mut self, f: F)
where F: Fn() + Send + Sync + 'static,
{
self.unmount_handler = Some(Arc::new(f));
}
pub fn spawn<F>(&self, name: &str, setup: F) -> crate::engine::Handle
where F: Fn(&mut Context) + Send + Sync + 'static,
{
self.engine.spawn_simple(name, setup)
}
pub fn send_to(&self, id: u64, msg: Message) {
self.engine.send_to(id, msg);
}
pub fn send_named(&self, name: &str, msg: Message) {
if let Some(id) = self.engine.registry.lookup(name) {
self.engine.send_to(id, msg);
}
}
pub fn request(&self, id: u64, msg: Message, timeout: Duration) -> Option<Message> {
self.engine.request(id, msg, timeout)
}
pub fn reply(&self, _msg: Message) {
}
pub fn metrics(&self) -> &ActorMetrics {
&self.metrics
}
pub fn sleep(&self, duration: Duration) {
let _ = self.rx.recv_timeout(duration);
}
pub fn poll(&self) -> Option<Message> {
self.rx.try_recv().ok()
}
pub fn id(&self) -> u64 { self.id }
pub fn name(&self) -> &str { &self.name }
}
#[cfg(test)]
mod tests {
use super::*;
use crossbeam_channel::unbounded;
fn dummy_ctx() -> Context {
let (tx, rx) = unbounded();
Context::new(1, "test".to_string(), Arc::new(RwLock::new(HashMap::new())), rx, tx, Arc::new(EngineInner::new()))
}
#[test]
fn use_state_get_set() {
let ctx = dummy_ctx();
let count = ctx.use_state("count", 42i64);
assert_eq!(count.get(), 42);
count.set(100);
assert_eq!(count.get(), 100);
}
#[test]
fn use_state_update() {
let ctx = dummy_ctx();
let count = ctx.use_state("count", 10i64);
count.update(|c| c * 2);
assert_eq!(count.get(), 20);
}
#[test]
fn state_survives_context_drop() {
let store = Arc::new(RwLock::new(HashMap::new()));
{
let (tx, rx) = unbounded();
let ctx = Context::new(1, "test".to_string(), store.clone(), rx, tx, Arc::new(EngineInner::new()));
let count = ctx.use_state("count", 42i64);
count.set(99);
}
{
let (tx, rx) = unbounded();
let ctx = Context::new(1, "test".to_string(), store.clone(), rx, tx, Arc::new(EngineInner::new()));
let count = ctx.use_state::<i64>("count", 0);
assert_eq!(count.get(), 99);
}
}
#[test]
fn scratch_does_not_persist() {
let ctx = dummy_ctx();
let mut s = ctx.use_scratch(42i64);
assert_eq!(s.get(), Some(&42));
s.set(100);
assert_eq!(s.get(), Some(&100));
}
}