use std::borrow::Cow;
use std::sync::{Arc, OnceLock};
use ek_core::context::PluginContext;
use ek_core::envelope::{Envelope, ReplyErr, e_req_type};
use ek_core::main_engine::MainEngine;
use ek_core::plugin::Plugin;
use ek_event::Event;
use crate::factory::PlugKind;
use crate::wire::{self, ListReq, ListRes, PlugRow, RemoveReq, RemoveRes, StartReq, StartRes};
pub const NAME: &str = "svctl";
pub struct Supervisor {
engine: Arc<MainEngine>,
kinds: Vec<PlugKind>,
ctx: OnceLock<PluginContext>,
}
impl Supervisor {
#[must_use]
pub fn new(engine: Arc<MainEngine>, kinds: Vec<PlugKind>) -> Self {
Self {
engine,
kinds,
ctx: OnceLock::new(),
}
}
fn on_request(&self, event: &Event) {
let Some(env) = event.data.downcast_ref::<Envelope>() else {
return; };
macro_rules! arm {
($ty:ty, $res:ty, $body:expr) => {
if let Some(req) = env.payload.downcast_ref::<$ty>() {
let res: $res = $body(req);
let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
return;
}
};
}
arm!(ListReq, ListRes, |_: &ListReq| self.rows());
arm!(StartReq, StartRes, |r: &StartReq| self.start_by(r));
arm!(RemoveReq, RemoveRes, |r: &RemoveReq| self.remove_by(r));
let _ = env
.reply
.send(Box::new(Err::<(), ReplyErr>(ReplyErr::InvalidPayload)));
}
fn rows(&self) -> ListRes {
self.engine
.plugin_names()
.iter()
.map(|name| {
let info = self.engine.get_plugin(name);
let stats = self.engine.mail_stats(name);
PlugRow {
name: name.clone(),
version: info
.as_ref()
.map_or_else(String::new, |i| i.version.clone()),
state: info
.as_ref()
.map_or_else(String::new, |i| format!("{:?}", i.state)),
deps: info.map_or_else(Vec::new, |i| i.deps),
spilled: stats.map_or(0, |s| s.spilled),
panics: stats.map_or(0, |s| s.panics),
}
})
.collect()
}
fn start_by(&self, req: &StartReq) -> StartRes {
let Some(kind) = self.kinds.iter().find(|k| k.name == req.factory) else {
return StartRes::Rejected {
reason: format!("unknown factory: {}", req.factory),
};
};
let want = req.name.trim();
let built = if want.is_empty() {
(kind.build)()
} else {
match &kind.build_named {
Some(f) => f(want),
None => {
return StartRes::Rejected {
reason: format!("factory {} does not support named instances", req.factory),
};
}
}
};
let name = built.name().to_string();
match self
.engine
.add_plugin(crate::factory::DynPlugin::wrap(built))
{
Ok(()) => {} Err(e) => {
return StartRes::Rejected {
reason: format!("{e:?}"),
};
}
}
StartRes::Started { name }
}
fn remove_by(&self, req: &RemoveReq) -> RemoveRes {
match self.engine.remove_plugin(req.name.as_str()) {
Ok(()) => RemoveRes::Removed {
name: req.name.clone(),
},
Err(ek_core::error::PluginError::UnknownName(_)) => RemoveRes::NoSuchName {
name: req.name.clone(),
},
Err(e) => RemoveRes::NoSuchName {
name: format!("{e:?}"),
},
}
}
}
impl Plugin for Supervisor {
fn name(&self) -> &str {
NAME
}
fn routes(&self) -> Vec<Cow<'static, str>> {
vec![Cow::Owned(e_req_type(NAME))]
}
fn version(&self) -> Cow<'static, str> {
Cow::Borrowed(env!("CARGO_PKG_VERSION"))
}
fn services(&self) -> Vec<ek_core::service::ServiceDesc> {
wire::svctl_services()
}
fn on_init(&self, ctx: &PluginContext) {
let _ = self.ctx.set(ctx.clone());
}
fn on_mail(&self, event: &Event) {
if event.event_type == e_req_type(NAME) {
self.on_request(event);
}
}
}