use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex, Weak};
use std::fmt::{self, Debug};
use std::future::Future;
use std::borrow::Cow;
use std::time::SystemTime;
use async_lock::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use futures::future::{join, ready, BoxFuture};
use tracing::{debug, error, info};
use futures::stream::BoxStream;
use crate::dev::{All, AttrConfig, AttrInfo, AttrMemorized, AttrMode, AttrQuality, AttrRead, AttrValueEvent, CommandInfo, DevInfo, Named, NamedResult, ToNames, ToNamesNotAll,
DevSource, DevState, DisplayLevel};
use crate::value::{CommandDataKind, IntoValue, Scalar, Spectrum, Value};
use crate::error::{reason, Error, ErrorKind, Result};
use crate::database::{GetPropSpec, SetPropSpec};
use crate::local::{self, LocalDevHandle};
use crate::name::{Name, ArcName};
use crate::backend::BackendDev;
use crate::pool;
use super::def::{AttrDef, CommandDef, DeleteFn, InitFn, StateFn, StatusFn, WriteFn};
use super::{reserved_name, Model, Ref, DEFAULT_STATE, DEFAULT_STATUS, INIT_NAME,
MISSING_STATUS, RESERVED_ATTRS, RESERVED_COMMANDS, STATE_NAME, STATUS_NAME};
use super::class::{AnyClass, Class};
use super::attr::{Attr, AttrAlarm};
use super::command::Command;
use super::server::Server;
pub trait AnyDev: Debug + Send + Sync {
fn name(&self) -> &str;
fn class(&self) -> &dyn AnyClass;
fn init(&self) -> BoxFuture<'_, Result<()>>;
fn withdraw(&self) -> BoxFuture<'_, Result<()>>;
fn serve(&self) -> Result<Arc<str>>;
}
pub struct Dev<M: Model> {
this: Weak<Self>,
name: Arc<str>,
class: Arc<Class<M>>,
mutable: Mutex<DevMutable<M>>,
model: RwLock<Option<Result<M>>>,
local_handle: Mutex<Option<LocalDevHandle>>,
}
impl<M: Model> Dev<M> {
pub(super) fn new(class: Arc<Class<M>>, name: Arc<str>) -> Arc<Self> {
let state_attr = Arc::new(Attr::new(Arc::clone(&class.def().state_attr)));
let status_attr = Arc::new(Attr::new(Arc::clone(&class.def().status_attr)));
Arc::new_cyclic(|this| Self {
this: this.clone(),
name,
class,
mutable: Mutex::new(DevMutable {
black_box: VecDeque::new(),
commands: HashMap::new(),
attrs: HashMap::new(),
state_attr,
status_attr,
auto_alarm_states: Arc::from([DevState::On]),
}),
model: RwLock::new(None),
local_handle: Mutex::new(None),
})
}
#[inline]
pub fn name(&self) -> &str {
&self.name
}
#[inline]
pub fn class(&self) -> &Class<M> {
&self.class
}
#[inline]
pub fn server(&self) -> &Server {
self.class.server()
}
pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
self.server().database().list_dev_prop_async(&self.name, filter).await
}
pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
self.server().database().list_dev_prop(&self.name, filter)
}
pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
where
Spec: GetPropSpec<Names>,
{
self.server().database().get_dev_prop_async::<Spec, Names>(&self.name, names)
}
pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
where
Spec: GetPropSpec<Names>,
{
self.server().database().get_dev_prop::<Spec, Names>(&self.name, names)
}
pub fn set_prop_async<'a, Spec>(&'a self, spec: Spec) -> impl Future<Output = Result<()>> + Send
where
Spec: SetPropSpec<'a>,
{
self.server().database().set_dev_prop_async(&self.name, spec)
}
pub fn set_prop<'a, Spec>(&'a self, spec: Spec) -> Result<()>
where
Spec: SetPropSpec<'a>,
{
self.server().database().set_dev_prop(&self.name, spec)
}
pub fn del_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
self.server().database().del_dev_prop_async(&self.name, names)
}
pub fn del_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
self.server().database().del_dev_prop(&self.name, names)
}
#[inline]
fn arc(&self) -> Arc<Self> {
self.this.upgrade().unwrap()
}
pub fn add_command(&self, def: CommandDef<M>) -> Result<()> {
if let Some(reserved) = reserved_name(&def.name, RESERVED_COMMANDS) {
return Err(Error::new(ErrorKind::CommandNameInvalid)
.with_description(format!("Command {} cannot be defined in the device {}, \
{reserved} is answered by the server itself", def.name, self.name)));
}
let mut mutable = self.mutable.lock().unwrap();
if mutable.commands.contains_key(Name::new(&def.name)) {
return Err(Error::new(ErrorKind::CommandAlreadyDefined)
.with_append_description(format_args!("Command {:?}", def.name)));
}
let name = ArcName::from(Arc::<str>::from(&*def.name));
mutable.commands.insert(name, Arc::new(Command::new(Arc::new(def))));
Ok(())
}
pub fn remove_command(&self, name: &str) -> bool {
self.mutable.lock().unwrap().commands.remove(Name::new(name)).is_some()
}
pub fn add_attr(&self, def: AttrDef<M>) -> Result<()> {
if let Some(reserved) = reserved_name(&def.name, RESERVED_ATTRS) {
return Err(Error::new(ErrorKind::AttrNameInvalid)
.with_description(format!("Attribute {} cannot be defined in the device {}, \
{reserved} is answered by the server itself", def.name, self.name)));
}
if let Err(err) = def.check() {
return Err(Error::new(ErrorKind::AttrNameInvalid)
.with_description(format!("Attribute {} cannot be defined in the device {}, {err}",
def.name, self.name)));
}
let mut mutable = self.mutable.lock().unwrap();
if mutable.attrs.contains_key(Name::new(&def.name)) {
return Err(Error::new(ErrorKind::AttrAlreadyDefined)
.with_append_description(format_args!("Attribute {:?}", def.name)));
}
let name = ArcName::from(Arc::<str>::from(&*def.name));
mutable.attrs.insert(name, Arc::new(Attr::new(Arc::new(def))));
Ok(())
}
pub async fn set_attr_w_value_async(&self, name: &str, value: impl IntoValue) -> Result<()> {
let attr = self.find_attr(name)?;
attr.set_w_value(self, value.into_value()).await
}
pub fn set_attr_w_value(&self, name: &str, value: impl IntoValue) -> Result<()> {
self.server().database().block_on(self.set_attr_w_value_async(name, value))?
}
pub async fn set_attr_config_async(&self, name: &str, config: &AttrConfig) -> Result<()> {
self.change_attr_config(name, config).await
}
pub fn set_attr_config(&self, name: &str, config: &AttrConfig) -> Result<()> {
self.server().database().block_on(self.set_attr_config_async(name, config))?
}
pub fn set_auto_alarm_states(&self, states: &[DevState]) {
self.mutable.lock().unwrap().auto_alarm_states = Arc::from(states);
}
pub fn remove_attr(&self, name: &str) -> bool {
self.mutable.lock().unwrap().attrs.remove(Name::new(name)).is_some()
}
pub fn get_command(&self, name: &str) -> Option<Ref<'_, Command<M>>> {
self.find_command(name).ok().map(Ref::new)
}
pub fn get_attr(&self, name: &str) -> Option<Ref<'_, Attr<M>>> {
self.find_attr(name).ok().map(Ref::new)
}
pub fn commands(&self) -> Vec<Ref<'_, Command<M>>> {
self.all_commands().into_iter().map(Ref::new).collect()
}
pub fn attrs(&self) -> Vec<Ref<'_, Attr<M>>> {
self.all_attrs().into_iter().map(Ref::new).collect()
}
pub fn push_attr_event(&self, name: &str, event: AttrValueEvent, mut read: Result<AttrRead<Value>>) -> Result<()> {
let attr = self.resolve_attr(name)?.into_attr();
attr.check_push_event(self, event)?;
if let Ok(read) = &mut read {
attr.merge_w_value(read);
}
self.server().push_attr_value(&self.name, attr.name(), event, read);
Ok(())
}
pub fn push_change_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
self.push_attr_event(name, AttrValueEvent::Change, Ok(read_now(value)))
}
pub fn push_archive_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
self.push_attr_event(name, AttrValueEvent::Archive, Ok(read_now(value)))
}
pub fn push_user_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
self.push_attr_event(name, AttrValueEvent::User, Ok(read_now(value)))
}
pub fn push_alarm_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
self.push_attr_event(name, AttrValueEvent::Alarm, Ok(read_now(value)))
}
pub fn push_data_ready_event(&self, name: &str) -> Result<()> {
let target = self.resolve_attr(name)?;
self.server().push_attr_ready(&self.name, target.name(), Ok(()));
Ok(())
}
async fn init(&self) {
let def = self.class.def();
let mut model = self.model.write().await;
self.delete_model(&mut model).await;
self.reset_interface();
self.load_all_attr_config().await;
let mut res = match &def.init {
InitFn::Sync(init) => init(self),
InitFn::Async(init) => init(self).await,
};
if let Ok(new_model) = &mut res {
self.load_all_attr_config().await;
for attr in self.all_attrs() {
attr.restore(self, new_model).await;
}
}
*model = Some(match res {
Ok(new_model) => {
info!(dev = %&*self.name, class = %&*def.name, "Initialized the device");
Ok(new_model)
}
Err(err) => {
error!(dev = %&*self.name, class = %&*def.name, %err, "Failed to initialize the device");
Err(err)
}
});
}
async fn delete_model(&self, model: &mut Option<Result<M>>) {
let Some(Ok(prev)) = model.take() else {
return;
};
let def = self.class.def();
debug!(dev = %&*self.name, class = %&*def.name, "Deleting the model of the device");
match &def.delete {
Some(DeleteFn::Sync(delete)) => delete(prev, self),
Some(DeleteFn::Async(delete)) => delete(prev, self).await,
None => {}
}
}
fn reset_interface(&self) {
let def = self.class.def();
let mut mutable = self.mutable.lock().unwrap();
mutable.commands.clear();
mutable.attrs.clear();
mutable.state_attr = Arc::new(Attr::new(Arc::clone(&def.state_attr)));
mutable.status_attr = Arc::new(Attr::new(Arc::clone(&def.status_attr)));
for def in &def.commands {
let name = ArcName::from(Arc::<str>::from(&*def.name));
mutable.commands.insert(name, Arc::new(Command::new(Arc::clone(def))));
}
for def in &def.attrs {
let name = ArcName::from(Arc::<str>::from(&*def.name));
mutable.attrs.insert(name, Arc::new(Attr::new(Arc::clone(def))));
}
}
async fn load_all_attr_config(&self) {
let attrs = {
let mutable = self.mutable.lock().unwrap();
[&mutable.state_attr, &mutable.status_attr]
.into_iter()
.chain(mutable.attrs.values())
.filter(|attr| !attr.is_loaded())
.map(Arc::clone)
.collect::<Vec<_>>()
};
self.load_given_attr_config(&attrs).await;
}
async fn load_given_attr_config(&self, attrs: &[Arc<Attr<M>>]) {
if attrs.is_empty() {
return;
}
let names = attrs
.iter()
.map(|attr| attr.name())
.collect::<Vec<_>>();
let (
class_props,
dev_props,
) = join(
self.class.attr_props(&names),
self.server()
.database()
.get_dev_attr_prop_async::<Spectrum<str>, _, _>(&self.name, names.as_slice(), All),
).await;
let dev_props = dev_props.and_then(|props| props
.into_iter()
.collect::<Result<Vec<_>>>());
for (index, attr) in attrs.iter().enumerate() {
attr.load_config(self,
class_props.as_ref().map(|props| &*props[index]),
dev_props.as_ref().map(|props| &*props[index]));
}
}
async fn state(&self, ctx: &mut ReadCtx<M>) -> Result<DevState> {
let mut guard = self.lock(self.state_access(ctx)).await;
match guard.get() {
Some(Ok(_)) => self.model_state(&mut guard.model(), ctx).await,
Some(Err(_)) => Ok(DevState::Fault),
None => Ok(DevState::Unknown),
}
}
async fn status(&self, ctx: &mut ReadCtx<M>) -> Result<String> {
let mut guard = self.lock(self.status_access(ctx)).await;
match guard.get() {
Some(Ok(_)) => self.model_status(&mut guard.model(), ctx).await,
Some(Err(err)) => Ok(make_failed_status(err)),
None => Ok(MISSING_STATUS.to_string()),
}
}
async fn model_state(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<DevState> {
let state = self.model_state_inner(model, ctx).await?;
if self.attrs_in_alarm(ctx, state, model, false).await.is_empty() {
Ok(state)
} else {
Ok(DevState::Alarm)
}
}
async fn model_status(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<String> {
let mut status = match &self.class.def().status {
Some(StatusFn::SharedSync(func)) => func(model.shared(), self).map(Cow::into_owned),
Some(StatusFn::ExclusiveSync(func)) => func(model.exclusive()?, self).map(Cow::into_owned),
Some(StatusFn::SharedAsync(func)) => func(model.shared(), self).await.map(Cow::into_owned),
Some(StatusFn::ExclusiveAsync(func)) => func(model.exclusive()?, self).await.map(Cow::into_owned),
None => Ok(DEFAULT_STATUS.to_string()),
}?;
if ctx.has_alarm() {
let state = self.model_state_inner(model, ctx).await?;
for (attr, alarm) in self.attrs_in_alarm(ctx, state, model, true).await {
status.push('\n');
status.push_str(&alarm.describe(attr.name()));
}
}
Ok(status)
}
async fn model_state_inner(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<DevState> {
if let Some(state) = ctx.state {
return Ok(state);
}
let state = match &self.class.def().state {
Some(StateFn::SharedSync(func)) => func(model.shared(), self),
Some(StateFn::ExclusiveSync(func)) => func(model.exclusive()?, self),
Some(StateFn::SharedAsync(func)) => func(model.shared(), self).await,
Some(StateFn::ExclusiveAsync(func)) => func(model.exclusive()?, self).await,
None => Ok(DEFAULT_STATE),
}?;
ctx.state = Some(state);
Ok(state)
}
async fn attrs_in_alarm(&self, ctx: &mut ReadCtx<M>, state: DevState, model: &mut ModelRef<'_, M>, every: bool) -> Vec<(Arc<Attr<M>>, AttrAlarm)> {
let mut found = Vec::new();
if !ctx.may_alarm(state) {
return found;
}
for alarm_attr in ctx.alarm.iter_mut().flat_map(|alarm| &mut alarm.attrs) {
let alarm = match alarm_attr.read {
Some(alarm) => alarm,
None => {
let alarm = alarm_attr.attr.read_with_alarm(self, model).await
.ok()
.and_then(|(_, alarm)| alarm);
alarm_attr.read = Some(alarm);
alarm
}
};
if let Some(alarm) = alarm {
found.push((Arc::clone(&alarm_attr.attr), alarm));
if !every {
break;
}
}
}
found
}
fn state_access(&self, ctx: &ReadCtx<M>) -> Access {
let access = match &self.class.def().state {
Some(StateFn::ExclusiveSync(_) | StateFn::ExclusiveAsync(_)) => Access::Exclusive,
_ => Access::Shared,
};
access.merge(ctx.access())
}
fn status_access(&self, ctx: &ReadCtx<M>) -> Access {
let access = match &self.class.def().status {
Some(StatusFn::ExclusiveSync(_) | StatusFn::ExclusiveAsync(_)) => Access::Exclusive,
_ => Access::Shared,
};
if ctx.has_alarm() { access.merge(self.state_access(ctx)) } else { access }
}
fn init_may_block(&self) -> bool {
let def = self.class.def();
matches!(def.init, InitFn::Sync(_)) || self.delete_may_block() || def.attrs.iter().any(|attr| {
attr.memorized == Some(AttrMemorized::RestoreAndWrite)
&& matches!(attr.write, Some(WriteFn::SharedSync(_) | WriteFn::ExclusiveSync(_)))
})
}
fn delete_may_block(&self) -> bool {
matches!(self.class.def().delete, Some(DeleteFn::Sync(_)))
}
fn state_may_block(&self, ctx: &ReadCtx<M>) -> bool {
matches!(&self.class.def().state, Some(StateFn::SharedSync(_) | StateFn::ExclusiveSync(_)))
|| ctx.may_block()
}
fn status_may_block(&self, ctx: &ReadCtx<M>) -> bool {
matches!(&self.class.def().status, Some(StatusFn::SharedSync(_) | StatusFn::ExclusiveSync(_)))
|| (ctx.has_alarm() && self.state_may_block(ctx))
}
fn read_may_block(&self, target: &AttrTarget<M>, ctx: &ReadCtx<M>) -> bool {
match target {
AttrTarget::State(_) => self.state_may_block(ctx),
AttrTarget::Status(_) => self.status_may_block(ctx),
AttrTarget::Attr(attr) => attr.read_may_block(),
}
}
fn command_may_block(&self, target: &CommandTarget<M>, ctx: &ReadCtx<M>) -> bool {
match target {
CommandTarget::State(_) => self.state_may_block(ctx),
CommandTarget::Status(_) => self.status_may_block(ctx),
CommandTarget::Init => self.init_may_block(),
CommandTarget::Command(command) => command.may_block(),
}
}
fn read_access(&self, target: &AttrTarget<M>, ctx: &ReadCtx<M>) -> Access {
match target {
AttrTarget::State(_) => self.state_access(ctx),
AttrTarget::Status(_) => self.status_access(ctx),
AttrTarget::Attr(attr) => attr.read_access(),
}
}
async fn lock(&self, access: Access) -> ModelGuard<'_, M> {
match access {
Access::Shared => ModelGuard::Shared(self.model.read().await),
Access::Exclusive => ModelGuard::Exclusive(self.model.write().await),
}
}
async fn lock_model(&self, access: Access) -> Result<ModelGuard<'_, M>> {
let guard = self.lock(access).await;
match guard.get() {
Some(Ok(_)) => Ok(guard),
Some(Err(err)) => Err(init_err(Some(err), &self.name)),
None => Err(init_err(None, &self.name)),
}
}
pub(super) async fn check_model(&self) -> Result<()> {
match &*self.model.read().await {
Some(Ok(_)) => Ok(()),
Some(Err(err)) => Err(init_err(Some(err), &self.name)),
None => Err(init_err(None, &self.name)),
}
}
fn find_command(&self, name: &str) -> Result<Arc<Command<M>>> {
self.mutable.lock().unwrap().commands.get(Name::new(name))
.map(Arc::clone)
.ok_or_else(|| Error::new(ErrorKind::CommandNotFound)
.with_append_description(format_args!("Command {name:?} of the device {:?}", self.name)))
}
fn find_attr(&self, name: &str) -> Result<Arc<Attr<M>>> {
self.mutable.lock().unwrap().attrs.get(Name::new(name))
.map(Arc::clone)
.ok_or_else(|| Error::new(ErrorKind::AttrNotFound)
.with_append_description(format_args!("Attribute {name:?} of the device {:?}", self.name)))
}
fn all_commands(&self) -> Vec<Arc<Command<M>>> {
self.mutable.lock().unwrap().commands.values().map(Arc::clone).collect()
}
fn all_attrs(&self) -> Vec<Arc<Attr<M>>> {
self.mutable.lock().unwrap().attrs.values().map(Arc::clone).collect()
}
fn state_attr(&self) -> Arc<Attr<M>> {
Arc::clone(&self.mutable.lock().unwrap().state_attr)
}
fn status_attr(&self) -> Arc<Attr<M>> {
Arc::clone(&self.mutable.lock().unwrap().status_attr)
}
fn resolve_attr(&self, name: &str) -> Result<AttrTarget<M>> {
if name.eq_ignore_ascii_case(STATE_NAME) {
Ok(AttrTarget::State(self.state_attr()))
} else if name.eq_ignore_ascii_case(STATUS_NAME) {
Ok(AttrTarget::Status(self.status_attr()))
} else {
self.find_attr(name).map(AttrTarget::Attr)
}
}
fn resolve_writable_attr(&self, name: &str) -> Result<Arc<Attr<M>>> {
match self.resolve_attr(name)? {
AttrTarget::Attr(attr) => Ok(attr),
AttrTarget::State(_) | AttrTarget::Status(_) => Err(not_writable_err(self, name)),
}
}
fn resolve_command(&self, name: &str) -> Result<CommandTarget<M>> {
if name.eq_ignore_ascii_case(STATE_NAME) {
Ok(CommandTarget::State(self.state_attr()))
} else if name.eq_ignore_ascii_case(STATUS_NAME) {
Ok(CommandTarget::Status(self.status_attr()))
} else if name.eq_ignore_ascii_case(INIT_NAME) {
Ok(CommandTarget::Init)
} else {
self.find_command(name).map(CommandTarget::Command)
}
}
fn all_attr_targets(&self) -> Vec<Resolved<AttrTarget<M>>> {
[Ok(AttrTarget::State(self.state_attr())), Ok(AttrTarget::Status(self.status_attr()))]
.into_iter()
.chain(self.all_attrs().into_iter().map(|attr| Ok(AttrTarget::Attr(attr))))
.collect()
}
fn attr_targets(&self, names: Option<Spectrum<str>>) -> Vec<Resolved<AttrTarget<M>>> {
match names {
Some(names) => scalars(&names)
.map(|name| {
let target = self.resolve_attr(name.value());
resolved(name, target)
})
.collect(),
None => self.all_attr_targets(),
}
}
fn command_targets(&self, names: Option<Spectrum<str>>) -> Vec<Resolved<CommandTarget<M>>> {
match names {
Some(names) => scalars(&names)
.map(|name| {
let target = self.resolve_command(name.value());
resolved(name, target)
})
.collect(),
None => [Ok(CommandTarget::State(self.state_attr())), Ok(CommandTarget::Status(self.status_attr())), Ok(CommandTarget::Init)]
.into_iter()
.chain(self.all_commands().into_iter().map(|command| Ok(CommandTarget::Command(command))))
.collect(),
}
}
async fn exec_command(&self, target: Result<CommandTarget<M>>, input: Value, mut ctx: ReadCtx<M>) -> Result<Value> {
let target = match target {
Ok(target) => target,
Err(err) => return Err(self.check_model().await.err().unwrap_or(err)),
};
match target {
CommandTarget::State(_) => Ok(self.state(&mut ctx).await?.into()),
CommandTarget::Status(_) => Ok(self.status(&mut ctx).await?.into()),
CommandTarget::Init => {
self.init().await;
self.check_model().await?;
Ok(Value::Void)
}
CommandTarget::Command(command) => {
let mut guard = self.lock_model(command.access()).await?;
command.exec(self, &mut guard.model(), input).await
}
}
}
async fn exec_read_attr(&self, targets: Vec<Resolved<AttrTarget<M>>>, mut ctx: ReadCtx<M>) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
let access = targets.iter()
.filter_map(|target| target.as_ref().ok())
.fold(Access::Shared, |access, target| access.merge(self.read_access(target, &ctx)));
let mut guard = self.lock_model(access).await?;
let mut model = guard.model();
let mut reads = targets
.iter()
.map(|_| None)
.collect::<Vec<_>>();
for (target, slot) in targets.iter().zip(&mut reads) {
if let Ok(AttrTarget::Attr(attr)) = target {
let read = attr.read_with_alarm(self, &mut model).await;
ctx.record(attr, read.as_ref().ok().and_then(|(_, alarm)| *alarm));
*slot = Some(Named::new(Scalar::from(attr.name()), read.map(|(read, _)| read)));
}
}
for (target, slot) in targets.into_iter().zip(&mut reads) {
*slot = Some(match target {
Ok(AttrTarget::Attr(_)) => continue,
Ok(AttrTarget::State(_)) => Named::new(Scalar::from(STATE_NAME), self.model_state(&mut model, &mut ctx).await
.map(|state| AttrRead::new(state.into()))),
Ok(AttrTarget::Status(_)) => Named::new(Scalar::from(STATUS_NAME), self.model_status(&mut model, &mut ctx).await
.map(|status| AttrRead::new(status.into()))),
Err(unresolved) => unresolved.map(Err),
});
}
Ok(reads.into_iter().map(|read| read.expect("every target is read")).collect())
}
async fn exec_write_attr(&self, targets: Vec<(Result<Arc<Attr<M>>>, Value)>) -> Result<Vec<Result<()>>> {
let access = targets.iter()
.filter_map(|(target, _)| target.as_ref().ok())
.fold(Access::Shared, |access, attr| access.merge(attr.write_access()));
let mut guard = self.lock_model(access).await?;
let mut model = guard.model();
let mut writes = Vec::with_capacity(targets.len());
for (target, value) in targets {
writes.push(match target {
Ok(attr) => attr.write(self, &mut model, value).await,
Err(err) => Err(err),
});
}
Ok(writes)
}
async fn exec_set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> Vec<Result<()>> {
let mut results = Vec::with_capacity(configs.len());
for Named { name, value } in configs {
results.push(self.change_attr_config(name.value(), &value).await);
}
results
}
async fn change_attr_config(&self, name: &str, config: &AttrConfig) -> Result<()> {
let target = self.resolve_attr(name)?;
let attr = target.attr();
let class_props = self.class.attr_props(&[attr.name()]).await?;
let changes = attr.change_config(&class_props[0], config)
.map_err(|err| Error::new(ErrorKind::Reason(reason::API_IncoherentValues))
.with_description(format!("The configuration of the attribute {:?} of the device {:?} \
cannot be changed: {err}", attr.name(), self.name)))?;
if changes.is_empty() {
return Ok(());
}
self.server().push_attr_conf(&self.name, attr.name(), target.info());
let database = self.server().database();
if !changes.put.is_empty() {
let put = changes.put.into_iter()
.map(|(prop, values)| (prop.name(), values))
.collect::<Vec<_>>();
database.set_dev_attr_prop_async(&self.name, (attr.name(), put)).await?;
}
if !changes.del.is_empty() {
let del = changes.del.iter().map(|prop| prop.name()).collect::<Vec<_>>();
database.del_dev_attr_prop_async(&self.name, attr.name(), del.as_slice()).await?;
}
Ok(())
}
async fn exec_write_read_attr(&self, targets: Vec<(Resolved<Arc<Attr<M>>>, Value)>) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
let access = targets.iter()
.filter_map(|(target, _)| target.as_ref().ok())
.fold(Access::Shared, |access, attr| access.merge(attr.write_access()).merge(attr.read_access()));
let mut guard = self.lock_model(access).await?;
let mut model = guard.model();
let mut reads = Vec::with_capacity(targets.len());
for (target, value) in targets {
reads.push(match target {
Ok(attr) => {
let read = match attr.write(self, &mut model, value).await {
Ok(()) => attr.read(self, &mut model).await,
Err(err) => Err(err),
};
Named::new(Scalar::from(attr.name()), read)
}
Err(unresolved) => unresolved.map(Err),
});
}
Ok(reads)
}
}
impl<M: Model> AnyDev for Dev<M> {
#[inline]
fn name(&self) -> &str {
&self.name
}
#[inline]
fn class(&self) -> &dyn AnyClass {
&*self.class
}
fn init(&self) -> BoxFuture<'_, Result<()>> {
let dev = self.arc();
pool::run(self.init_may_block(), async move {
dev.init().await;
Ok(())
})
}
fn withdraw(&self) -> BoxFuture<'_, Result<()>> {
let dev = self.arc();
Box::pin(async move {
dev.local_handle.lock().unwrap().take();
if let Some(backend) = dev.server().backend() {
backend.remove(&dev.name);
}
pool::run(dev.delete_may_block(), async move {
dev.delete_model(&mut *dev.model.write().await).await;
Ok(())
}).await
})
}
fn serve(&self) -> Result<Arc<str>> {
let backend = self.server().served_backend()?;
let dev = DevImpl::from_arc(self.arc()) as Arc<dyn BackendDev>;
let itr = backend.add(&self.name, Arc::clone(&dev))?;
if self.server().local {
let local_handle = local::register(&dev, Arc::clone(&itr), backend.host(), backend.port(), Arc::clone(&self.name));
*self.local_handle.lock().unwrap() = Some(local_handle);
}
Ok(itr)
}
}
impl<M: Model> Debug for Dev<M> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Dev").field("name", &self.name).finish_non_exhaustive()
}
}
struct DevMutable<M: Model> {
black_box: VecDeque<String>,
commands: HashMap<ArcName, Arc<Command<M>>>,
attrs: HashMap<ArcName, Arc<Attr<M>>>,
state_attr: Arc<Attr<M>>,
status_attr: Arc<Attr<M>>,
auto_alarm_states: Arc<[DevState]>,
}
struct ReadCtx<M: Model> {
alarm: Option<AlarmCtx<M>>,
state: Option<DevState>,
}
struct AlarmCtx<M: Model> {
states: Arc<[DevState]>,
attrs: Vec<AlarmAttr<M>>,
}
struct AlarmAttr<M: Model> {
attr: Arc<Attr<M>>,
read: Option<Option<AttrAlarm>>,
}
impl<M: Model> ReadCtx<M> {
fn new(dev: &Dev<M>, involves_state: bool) -> Self {
let alarm = involves_state.then(|| {
let mutable = dev.mutable.lock().unwrap();
if mutable.auto_alarm_states.is_empty() {
return None;
}
let mut attrs = mutable.attrs.values()
.filter(|attr| attr.has_alarms())
.map(|attr| AlarmAttr {
attr: Arc::clone(attr),
read: None,
})
.collect::<Vec<_>>();
if attrs.is_empty() {
return None;
}
attrs.sort_by(|a, b| a.attr.name().cmp(b.attr.name()));
Some(AlarmCtx {
states: Arc::clone(&mutable.auto_alarm_states),
attrs,
})
}).flatten();
Self {
alarm,
state: None,
}
}
fn has_alarm(&self) -> bool {
self.alarm.is_some()
}
fn may_alarm(&self, state: DevState) -> bool {
self.alarm.as_ref().is_some_and(|alarm| alarm.states.contains(&state))
}
fn record(&mut self, attr: &Arc<Attr<M>>, alarm: Option<AttrAlarm>) {
let alarm_attr = self.alarm.iter_mut()
.flat_map(|alarm| &mut alarm.attrs)
.find(|alarm_attr| Arc::ptr_eq(&alarm_attr.attr, attr));
if let Some(alarm_attr) = alarm_attr {
alarm_attr.read = Some(alarm);
}
}
fn access(&self) -> Access {
self.alarm.iter()
.flat_map(|alarm| &alarm.attrs)
.fold(Access::Shared, |access, alarm_attr| access.merge(alarm_attr.attr.read_access()))
}
fn may_block(&self) -> bool {
self.alarm.iter()
.flat_map(|alarm| &alarm.attrs)
.any(|alarm_attr| alarm_attr.attr.read_may_block())
}
}
enum AttrTarget<M: Model> {
State(Arc<Attr<M>>),
Status(Arc<Attr<M>>),
Attr(Arc<Attr<M>>),
}
impl<M: Model> AttrTarget<M> {
fn involves_state(&self) -> bool {
matches!(self, Self::State(_) | Self::Status(_))
}
fn attr(&self) -> &Arc<Attr<M>> {
match self {
Self::State(attr) |
Self::Status(attr) |
Self::Attr(attr) => attr,
}
}
fn into_attr(self) -> Arc<Attr<M>> {
match self {
Self::State(attr) |
Self::Status(attr) |
Self::Attr(attr) => attr,
}
}
fn name(&self) -> &str {
match self {
Self::State(attr) |
Self::Status(attr) |
Self::Attr(attr) => attr.name(),
}
}
fn info(&self) -> Result<AttrInfo> {
match self {
Self::State(attr) |
Self::Status(attr) => attr.info_as(AttrMode::Read),
Self::Attr(attr) => attr.info(),
}
}
}
enum CommandTarget<M: Model> {
State(Arc<Attr<M>>),
Status(Arc<Attr<M>>),
Init,
Command(Arc<Command<M>>),
}
impl<M: Model> CommandTarget<M> {
fn involves_state(&self) -> bool {
matches!(self, Self::State(_) | Self::Status(_))
}
fn name(&self) -> &str {
match self {
Self::State(attr) |
Self::Status(attr) => attr.name(),
Self::Init => INIT_NAME,
Self::Command(command) => command.name(),
}
}
fn info(&self) -> CommandInfo {
match self {
Self::State(_) => CommandInfo {
display_level: DisplayLevel::Operator,
in_kind: CommandDataKind::Void,
out_kind: CommandDataKind::State,
in_doc: None,
out_doc: Some(Cow::Borrowed("The state of the device")),
},
Self::Status(_) => CommandInfo {
display_level: DisplayLevel::Operator,
in_kind: CommandDataKind::Void,
out_kind: CommandDataKind::String,
in_doc: None,
out_doc: Some(Cow::Borrowed("The status of the device")),
},
Self::Init => CommandInfo {
display_level: DisplayLevel::Operator,
in_kind: CommandDataKind::Void,
out_kind: CommandDataKind::Void,
in_doc: None,
out_doc: None,
},
Self::Command(command) => command.info(),
}
}
}
fn read_now(value: impl IntoValue) -> AttrRead<Value> {
AttrRead {
value: Some(value.into_value()),
w_value: None,
quality: AttrQuality::Valid,
time: SystemTime::now(),
}
}
fn not_writable_err<M: Model>(dev: &Dev<M>, name: &str) -> Error {
Error::new(ErrorKind::AttrNotWritable)
.with_append_description(format_args!("Attribute {name:?} of the device {:?}",
dev.name()))
}
fn check_polled(source: DevSource, what: fmt::Arguments<'_>) -> Result<()> {
match source {
DevSource::Dev | DevSource::CacheDev => Ok(()),
DevSource::Cache => Err(Error::new(ErrorKind::Reason(reason::API_DeviceNotPolled))
.with_description(format!("Cannot answer from the polling cache, this server \
polls nothing yet ({what})"))),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Access {
Shared,
Exclusive,
}
impl Access {
pub fn merge(self, other: Self) -> Self {
match (self, other) {
(Self::Shared, Self::Shared) => Self::Shared,
_ => Self::Exclusive,
}
}
}
pub enum ModelGuard<'a, M> {
Shared(RwLockReadGuard<'a, Option<Result<M>>>),
Exclusive(RwLockWriteGuard<'a, Option<Result<M>>>),
}
impl<M> ModelGuard<'_, M> {
#[inline]
fn get(&self) -> &Option<Result<M>> {
match self {
Self::Shared(guard) => guard,
Self::Exclusive(guard) => guard,
}
}
fn model(&mut self) -> ModelRef<'_, M> {
match self {
Self::Shared(guard) => match &**guard {
Some(Ok(model)) => ModelRef::Shared(model),
_ => panic!("the device has no model"),
}
Self::Exclusive(guard) => match &mut **guard {
Some(Ok(model)) => ModelRef::Exclusive(model),
_ => panic!("the device has no model"),
}
}
}
}
pub enum ModelRef<'m, M> {
Shared(&'m M),
Exclusive(&'m mut M),
}
impl<M> ModelRef<'_, M> {
#[inline]
pub fn shared(&self) -> &M {
match self {
Self::Shared(model) => model,
Self::Exclusive(model) => model,
}
}
#[inline]
pub fn exclusive(&mut self) -> Result<&mut M> {
match self {
Self::Exclusive(model) => Ok(model),
Self::Shared(_) => Err(Error::new(ErrorKind::Internal)
.with_description("The model was borrowed shared while a handler of that \
request needs it exclusively")),
}
}
}
fn make_failed_status(err: &Error) -> String {
format!("The initialization of the device failed:\n{err}")
}
fn init_err(failure: Option<&Error>, name: &str) -> Error {
match failure {
Some(err) => Error::new(ErrorKind::DevInitFailed)
.with_append_description(format_args!("Device {name:?}"))
.with_append_description(err),
None => Error::new(ErrorKind::DevNotConnected)
.with_description(format!("The device {name} is not initialized")),
}
}
pub fn some_doc(doc: &Cow<'static, str>) -> Option<Cow<'static, str>> {
(!doc.is_empty()).then(|| doc.clone())
}
#[repr(transparent)]
struct DevImpl<M: Model>(Dev<M>);
impl<M: Model> DevImpl<M> {
#[inline]
fn from_arc(dev: Arc<Dev<M>>) -> Arc<Self> {
unsafe { Arc::from_raw(Arc::into_raw(dev) as *const Self) }
}
}
impl<M: Model> Debug for DevImpl<M> {
#[inline]
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
Debug::fmt(&self.0, f)
}
}
impl<M: Model> BackendDev for DevImpl<M> {
fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
Box::pin(ready(Ok(Arc::clone(&self.0.name))))
}
fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
Box::pin(ready(Ok(Arc::clone(&self.0.server().admin_dev_name))))
}
fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
Box::pin(ready(Ok(Arc::from(&*self.0.class.def().doc))))
}
fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {
let def = self.0.class.def();
let server = self.0.server();
let backend = match server.served_backend() {
Ok(backend) => backend,
Err(err) => return Box::pin(ready(Err(err))),
};
Box::pin(ready(Ok(DevInfo {
class: def.name.to_string(),
server_id: server.name().to_string(),
server_host: backend.host().to_string(),
server_version: backend.version(),
doc_url: def.doc_url.clone(),
dev_type: Some(def.doc.clone()),
version_info: vec![
(Cow::Borrowed("rstango"), Cow::Borrowed(env!("CARGO_PKG_VERSION"))),
],
})))
}
fn ping(&self) -> BoxFuture<'_, Result<()>> {
Box::pin(ready(Ok(())))
}
fn state(&self) -> BoxFuture<'_, Result<DevState>> {
let dev = self.0.arc();
let mut ctx = ReadCtx::new(&dev, true);
pool::run(dev.state_may_block(&ctx), async move { dev.state(&mut ctx).await })
}
fn status(&self) -> BoxFuture<'_, Result<String>> {
let dev = self.0.arc();
let mut ctx = ReadCtx::new(&dev, true);
pool::run(dev.status_may_block(&ctx), async move { dev.status(&mut ctx).await })
}
fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Vec<String>>> {
let mutable = self.0.mutable.lock().unwrap();
let black_box = mutable.black_box.iter()
.take(max.unwrap_or(usize::MAX))
.cloned()
.collect();
Box::pin(ready(Ok(black_box)))
}
fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {
if let Err(err) = check_polled(source, format_args!("the command {} of the device {}", name.value(), self.0.name)) {
return Box::pin(ready(Err(err)));
}
let dev = self.0.arc();
let target = dev.resolve_command(name.value());
let ctx = ReadCtx::new(&dev, target.as_ref().is_ok_and(CommandTarget::involves_state));
let blocking = target.as_ref().is_ok_and(|target| dev.command_may_block(target, &ctx));
pool::run(blocking, async move {
let res = dev.exec_command(target, input, ctx).await;
if let Err(err) = &res {
debug!(dev = %&*dev.name, command = %name.value(), %err, "Failed to execute the command");
}
res
})
}
fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
if let Err(err) = check_polled(source, format_args!("the attributes of the device {}", self.0.name)) {
return Box::pin(ready(Err(err)));
}
let dev = self.0.arc();
let targets = dev.attr_targets(names);
let ctx = ReadCtx::new(&dev, targets.iter().flatten().any(AttrTarget::involves_state));
let blocking = targets.iter().flatten().any(|target| dev.read_may_block(target, &ctx));
pool::run(blocking, async move { dev.exec_read_attr(targets, ctx).await })
}
fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
let dev = self.0.arc();
let targets = values.into_iter()
.map(|write| (dev.resolve_writable_attr(write.name.value()), write.value))
.collect::<Vec<_>>();
let blocking = targets.iter()
.filter_map(|(target, _)| target.as_ref().ok())
.any(|attr| attr.write_may_block());
pool::run(blocking, async move { dev.exec_write_attr(targets).await })
}
fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
let dev = self.0.arc();
let targets = values.into_iter()
.map(|write| {
let target = dev.resolve_writable_attr(write.name.value());
(resolved(write.name, target), write.value)
})
.collect::<Vec<_>>();
let blocking = targets.iter()
.filter_map(|(target, _)| target.as_ref().ok())
.any(|attr| attr.write_may_block() || attr.read_may_block());
pool::run(blocking, async move { dev.exec_write_read_attr(targets).await })
}
fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {
let infos = self.0.command_targets(names).into_iter()
.map(|target| match target {
Ok(target) => Named::new(Scalar::from(target.name()), Ok(target.info())),
Err(unresolved) => unresolved.map(Err),
})
.collect();
Box::pin(ready(Ok(infos)))
}
fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {
let infos = self.0.attr_targets(names).into_iter()
.map(|target| match target {
Ok(target) => Named::new(Scalar::from(target.name()), target.info()),
Err(unresolved) => unresolved.map(Err),
})
.collect();
Box::pin(ready(Ok(infos)))
}
fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
Box::pin(async move { Ok(self.0.exec_set_attr_config(configs).await) })
}
fn command_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<(SystemTime, Result<Value>)>>> {
Box::pin(ready(Err(Error::not_implemented()
.with_append_description("The recorded history of a command"))))
}
fn attr_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<Result<AttrRead<Value>>>>> {
Box::pin(ready(Err(Error::not_implemented()
.with_append_description("The recorded history of an attribute"))))
}
fn subscribe_attr_value<'a>(&'a self, _admin: Arc<dyn BackendDev>, name: &'a str, event: AttrValueEvent) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
Box::pin(ready(self.0.resolve_attr(name).and_then(|target| {
target.attr().check_push_event(&self.0, event)?;
Ok(self.0.server().subscribe_attr_value(&self.0.name, target.name(), event))
})))
}
fn subscribe_attr_ready<'a>(&'a self, _admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<()>>>> {
Box::pin(ready(self.0.resolve_attr(name).map(|target| {
self.0.server().subscribe_attr_ready(&self.0.name, target.name())
})))
}
fn subscribe_attr_conf<'a>(&'a self, _admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> {
Box::pin(ready(self.0.resolve_attr(name).map(|target| {
self.0.server().subscribe_attr_conf(&self.0.name, target.name())
})))
}
}
type Resolved<T> = std::result::Result<T, Named<Error>>;
fn resolved<T>(name: Scalar<str>, res: Result<T>) -> Resolved<T> {
res.map_err(|err| Named::new(name, err))
}
fn scalars(names: &Spectrum<str>) -> impl Iterator<Item = Scalar<str>> + '_ {
(0..names.len()).map(|index| names.get_scalar([index]).expect("a name at every index"))
}
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use std::thread;
use futures::channel::oneshot;
use futures::executor::block_on;
use futures::{FutureExt, StreamExt};
use crate::dev::{ToNames, WriteAttrSpec};
use crate::value::FromValue;
use super::*;
use super::super::def::{AttrDef, ClassDef, CommandDef};
use super::super::server::ServerBuilder;
use crate::testing::TestContext;
fn serve<M: Model>(def: ClassDef<M>) -> (TestContext, Arc<DevImpl<M>>) {
let ctx = TestContext::new().unwrap();
ctx.database().add_server("Test/test", &[]).unwrap();
ctx.database().add_dev("test/dev/1", &def.name, "Test/test").unwrap();
let server = ctx.serve(ServerBuilder::new("Test", "test")).unwrap();
let class = server.add_class(def).unwrap();
let dev = class.add_dev("test/dev/1").unwrap();
let dev = DevImpl::from_arc(Arc::clone(&dev.inner));
drop(class);
(ctx, dev)
}
#[test]
fn push_event() {
let (ctx, dev) = serve(model_def());
let server = ctx.server("Test/test").unwrap();
server.sync_blocking().unwrap();
let mut state = server.subscribe_attr_value("test/dev/1", "State", AttrValueEvent::Change);
let mut count = server.subscribe_attr_value("test/dev/1", "Count", AttrValueEvent::Change);
thread::scope(|scope| {
scope.spawn(|| dev.0.push_change_event("state", DevState::Moving).unwrap());
});
let read = block_on(state.next()).unwrap().unwrap();
assert_eq!(read.value, Some(Value::from(DevState::Moving)));
assert_eq!(read.w_value, None);
assert_eq!(read.quality, AttrQuality::Valid);
dev.0.set_attr_w_value("Count", 5).unwrap();
dev.0.push_change_event("Count", 3).unwrap();
let read = block_on(count.next()).unwrap().unwrap();
assert_eq!(read.value, Some(Value::from(3)));
assert_eq!(read.w_value, Some(Value::from(5)));
assert!(dev.0.push_change_event("Missing", 1.0).is_err());
let err = dev.0.push_archive_event("Count", 3).unwrap_err();
assert_eq!(err.kind(), ErrorKind::Reason(reason::API_AttributePollingNotStarted));
dev.0.push_user_event("Count", 3).unwrap();
}
#[test]
fn set_attr_config() {
let (ctx, dev) = serve(model_def());
let server = ctx.server("Test/test").unwrap();
server.sync_blocking().unwrap();
let mut conf = server.subscribe_attr_conf("test/dev/1", "Count");
let mut config = dev.0.get_attr("Count").unwrap().config();
config.label = Some("Counter".into());
dev.0.set_attr_config("count", &config).unwrap();
assert_eq!(dev.0.get_attr("Count").unwrap().config().label.as_deref(), Some("Counter"));
let info = block_on(conf.next()).unwrap().unwrap();
assert_eq!(info.config.label.as_deref(), Some("Counter"));
dev.0.set_attr_config("Count", &config).unwrap();
assert!(conf.next().now_or_never().is_none());
}
fn thread_name() -> String {
thread::current().name().unwrap_or_default().to_string()
}
struct Motor {
count: i32,
}
impl Motor {
async fn init(_dev: &Dev<Self>) -> Result<Self> {
Ok(Self { count: 0 })
}
async fn incr(&mut self, _dev: &Dev<Self>, arg: i32) -> Result<i32> {
self.count += arg;
Ok(self.count)
}
async fn count(&self, _dev: &Dev<Self>) -> Result<i32> {
Ok(self.count)
}
async fn set_count(&mut self, _dev: &Dev<Self>, arg: i32) -> Result<()> {
self.count = arg;
Ok(())
}
async fn state(&self, _dev: &Dev<Self>) -> Result<DevState> {
Ok(if self.count == 0 { DevState::On } else { DevState::Moving })
}
fn thread(&self, _dev: &Dev<Self>) -> Result<String> {
Ok(thread_name())
}
async fn thread_async(&self, _dev: &Dev<Self>) -> Result<String> {
Ok(thread_name())
}
}
fn model_def() -> ClassDef<Motor> {
ClassDef::new_async("Motor", Motor::init)
.state_async(Motor::state)
.push_state_events(Some(false), None)
.command(CommandDef::new("Incr").exec_exclusive_async(Motor::incr))
.attr(AttrDef::new("Count").read_async(Motor::count).write_exclusive_async(Motor::set_count)
.push_change_event(false))
.attr(AttrDef::new("Thread").read(Motor::thread))
.attr(AttrDef::new("ThreadAsync").read_async(Motor::thread_async))
}
fn read_one<T: FromValue>(dev: &DevImpl<Motor>, name: &str) -> T {
let mut reads = block_on(dev.read_attr(name.to_names(), DevSource::Dev)).unwrap();
let read = reads.pop().unwrap().value;
T::from_value(read.unwrap().value.unwrap()).unwrap()
}
#[test]
fn async_handlers() {
let (_ctx, dev) = serve(model_def());
dev.0.server().sync_blocking().unwrap();
assert_eq!(block_on(dev.state()).unwrap(), DevState::On);
let count = block_on(dev.command(Scalar::from("Incr"), Value::from(3), DevSource::Dev)).unwrap();
assert_eq!(i32::from_value(count).unwrap(), 3);
assert_eq!(block_on(dev.state()).unwrap(), DevState::Moving);
let writes = block_on(dev.write_attr(("Count", 7).into_values())).unwrap();
assert!(writes.into_iter().all(|write| write.is_ok()));
assert_eq!(read_one::<i32>(&dev, "Count"), 7);
block_on(dev.command(Scalar::from("Init"), Value::Void, DevSource::Dev)).unwrap();
assert_eq!(read_one::<i32>(&dev, "Count"), 0);
}
#[test]
fn unknown_attr_name() {
let (_ctx, dev) = serve(model_def());
dev.0.server().sync_blocking().unwrap();
let names = Spectrum::from(&["count", "Nothing"][..]);
let reads = block_on(dev.read_attr(Some(names.clone()), DevSource::Dev)).unwrap();
assert_eq!(reads[0].name.value(), "Count");
assert!(reads[0].value.is_ok());
assert_eq!(reads[1].name.value().as_ptr(), names.get([1]).unwrap().as_ptr());
assert_eq!(reads[1].value.as_ref().unwrap_err().kind(), ErrorKind::AttrNotFound);
}
#[test]
fn request_threads() {
let (_ctx, dev) = serve(model_def());
dev.0.server().sync_blocking().unwrap();
let on_pool = |name: &str| name.starts_with("pool-");
assert!(on_pool(&read_one::<String>(&dev, "Thread")));
assert!(!on_pool(&read_one::<String>(&dev, "ThreadAsync")));
let reads = block_on(dev.read_attr(["ThreadAsync", "Thread"].to_names(), DevSource::Dev)).unwrap();
for read in reads {
assert!(on_pool(&String::from_value(read.value.unwrap().value.unwrap()).unwrap()));
}
}
#[test]
fn lock_across_await() {
let (tx, rx) = oneshot::channel::<()>();
let gate = Mutex::new(Some(rx));
let def = model_def()
.command(CommandDef::new("Wait").exec_exclusive_async_with(move |_model: &mut Motor, _dev, _input| {
let rx = gate.lock().unwrap().take();
Box::pin(async move {
if let Some(rx) = rx {
let _ = rx.await;
}
Ok(Value::Void)
})
}));
let (_ctx, dev) = serve(def);
dev.0.server().sync_blocking().unwrap();
let mut wait = dev.command(Scalar::from("Wait"), Value::Void, DevSource::Dev);
assert!((&mut wait).now_or_never().is_none());
let mut read = dev.read_attr("Count".to_names(), DevSource::Dev);
assert!((&mut read).now_or_never().is_none());
tx.send(()).unwrap();
block_on(wait).unwrap();
assert!(block_on(read).unwrap().pop().unwrap().value.is_ok());
}
#[test]
fn async_init_failed() {
let def = ClassDef::<Motor>::new_async_with("Motor", |_dev| Box::pin(async {
Err(Error::new(ErrorKind::Internal).with_description("No hardware"))
}));
let (_ctx, dev) = serve(def);
assert_eq!(block_on(dev.state()).unwrap(), DevState::Unknown);
dev.0.server().sync_blocking().unwrap();
assert_eq!(block_on(dev.state()).unwrap(), DevState::Fault);
assert!(block_on(dev.status()).unwrap().contains("No hardware"));
assert!(block_on(dev.command(Scalar::from("Init"), Value::Void, DevSource::Dev)).is_err());
}
}