use std::any::Any;
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::mem;
use std::net::TcpStream;
use std::sync::{Arc, Mutex};
use base::cross_process_instant::CrossProcessInstant;
use log::{debug, warn};
use serde_json::{Map, Value};
use crate::StreamId;
#[derive(PartialEq)]
pub enum ActorMessageStatus {
Processed,
Ignored,
}
pub(crate) trait Actor: Any + ActorAsAny {
fn handle_message(
&self,
registry: &ActorRegistry,
msg_type: &str,
msg: &Map<String, Value>,
stream: &mut TcpStream,
id: StreamId,
) -> Result<ActorMessageStatus, ()>;
fn name(&self) -> String;
fn cleanup(&self, _id: StreamId) {}
}
pub(crate) trait ActorAsAny {
fn actor_as_any(&self) -> &dyn Any;
fn actor_as_any_mut(&mut self) -> &mut dyn Any;
}
impl<T: Actor> ActorAsAny for T {
fn actor_as_any(&self) -> &dyn Any {
self
}
fn actor_as_any_mut(&mut self) -> &mut dyn Any {
self
}
}
pub struct ActorRegistry {
actors: HashMap<String, Box<dyn Actor + Send>>,
new_actors: RefCell<Vec<Box<dyn Actor + Send>>>,
old_actors: RefCell<Vec<String>>,
script_actors: RefCell<HashMap<String, String>>,
shareable: Option<Arc<Mutex<ActorRegistry>>>,
next: Cell<u32>,
start_stamp: CrossProcessInstant,
}
impl ActorRegistry {
pub fn new() -> ActorRegistry {
ActorRegistry {
actors: HashMap::new(),
new_actors: RefCell::new(vec![]),
old_actors: RefCell::new(vec![]),
script_actors: RefCell::new(HashMap::new()),
shareable: None,
next: Cell::new(0),
start_stamp: CrossProcessInstant::now(),
}
}
pub(crate) fn cleanup(&self, id: StreamId) {
for actor in self.actors.values() {
actor.cleanup(id);
}
}
pub fn create_shareable(self) -> Arc<Mutex<ActorRegistry>> {
if let Some(shareable) = self.shareable {
return shareable;
}
let shareable = Arc::new(Mutex::new(self));
{
let mut lock = shareable.lock();
let registry = lock.as_mut().unwrap();
registry.shareable = Some(shareable.clone());
}
shareable
}
pub fn shareable(&self) -> Arc<Mutex<ActorRegistry>> {
self.shareable.as_ref().unwrap().clone()
}
pub fn start_stamp(&self) -> CrossProcessInstant {
self.start_stamp
}
pub fn register_script_actor(&self, script_id: String, actor: String) {
debug!("registering {} ({})", actor, script_id);
let mut script_actors = self.script_actors.borrow_mut();
script_actors.insert(script_id, actor);
}
pub fn script_to_actor(&self, script_id: String) -> String {
if script_id.is_empty() {
return "".to_owned();
}
self.script_actors.borrow().get(&script_id).unwrap().clone()
}
pub fn script_actor_registered(&self, script_id: String) -> bool {
self.script_actors.borrow().contains_key(&script_id)
}
pub fn actor_to_script(&self, actor: String) -> String {
for (key, value) in &*self.script_actors.borrow() {
debug!("checking {}", value);
if *value == actor {
return key.to_owned();
}
}
panic!("couldn't find actor named {}", actor)
}
pub fn new_name(&self, prefix: &str) -> String {
let suffix = self.next.get();
self.next.set(suffix + 1);
format!("{}{}", prefix, suffix)
}
pub(crate) fn register(&mut self, actor: Box<dyn Actor + Send>) {
self.actors.insert(actor.name(), actor);
}
pub(crate) fn register_later(&self, actor: Box<dyn Actor + Send>) {
let mut actors = self.new_actors.borrow_mut();
actors.push(actor);
}
pub fn find<'a, T: Any>(&'a self, name: &str) -> &'a T {
let actor = self.actors.get(name).unwrap();
actor.actor_as_any().downcast_ref::<T>().unwrap()
}
pub fn find_mut<'a, T: Any>(&'a mut self, name: &str) -> &'a mut T {
let actor = self.actors.get_mut(name).unwrap();
actor.actor_as_any_mut().downcast_mut::<T>().unwrap()
}
pub(crate) fn handle_message(
&mut self,
msg: &Map<String, Value>,
stream: &mut TcpStream,
id: StreamId,
) -> Result<(), ()> {
let to = match msg.get("to") {
Some(to) => to.as_str().unwrap(),
None => {
warn!("Received unexpected message: {:?}", msg);
return Err(());
},
};
match self.actors.get(to) {
None => debug!("message received for unknown actor \"{}\"", to),
Some(actor) => {
let msg_type = msg.get("type").unwrap().as_str().unwrap();
if actor.handle_message(self, msg_type, msg, stream, id)? !=
ActorMessageStatus::Processed
{
debug!(
"unexpected message type \"{}\" found for actor \"{}\"",
msg_type, to
);
}
},
}
let new_actors = mem::take(&mut *self.new_actors.borrow_mut());
for actor in new_actors.into_iter() {
self.actors.insert(actor.name().to_owned(), actor);
}
let old_actors = mem::take(&mut *self.old_actors.borrow_mut());
for name in old_actors {
self.drop_actor(name);
}
Ok(())
}
pub fn drop_actor(&mut self, name: String) {
self.actors.remove(&name);
}
pub fn drop_actor_later(&self, name: String) {
let mut actors = self.old_actors.borrow_mut();
actors.push(name);
}
}