1use std::collections::{HashMap, VecDeque};
4use std::sync::{Arc, Mutex, Weak};
5use std::fmt::{self, Debug};
6use std::future::Future;
7use std::borrow::Cow;
8use std::time::SystemTime;
9
10use async_lock::{RwLock, RwLockReadGuard, RwLockWriteGuard};
11use futures::future::{join, ready, BoxFuture};
12use tracing::{debug, error, info};
13use futures::stream::BoxStream;
14
15use crate::dev::{All, AttrConfig, AttrInfo, AttrMemorized, AttrMode, AttrQuality, AttrRead, AttrValueEvent, CommandInfo, DevInfo, Named, NamedResult, ToNames, ToNamesNotAll,
16 DevSource, DevState, DisplayLevel};
17use crate::value::{CommandDataKind, IntoValue, Scalar, Spectrum, Value};
18use crate::error::{reason, Error, ErrorKind, Result};
19use crate::database::{GetPropSpec, SetPropSpec};
20use crate::local::{self, LocalDevHandle};
21use crate::name::{Name, ArcName};
22use crate::backend::BackendDev;
23use crate::pool;
24
25use super::def::{AttrDef, CommandDef, DeleteFn, InitFn, StateFn, StatusFn, WriteFn};
26use super::{reserved_name, Model, Ref, DEFAULT_STATE, DEFAULT_STATUS, INIT_NAME,
27 MISSING_STATUS, RESERVED_ATTRS, RESERVED_COMMANDS, STATE_NAME, STATUS_NAME};
28use super::class::{AnyClass, Class};
29use super::attr::{Attr, AttrAlarm};
30use super::command::Command;
31use super::server::Server;
32
33
34pub trait AnyDev: Debug + Send + Sync {
37
38 fn name(&self) -> &str;
40
41 fn class(&self) -> &dyn AnyClass;
43
44 fn init(&self) -> BoxFuture<'_, Result<()>>;
49
50 fn withdraw(&self) -> BoxFuture<'_, Result<()>>;
61
62 fn serve(&self) -> Result<Arc<str>>;
66
67}
68
69pub struct Dev<M: Model> {
81 this: Weak<Self>,
83 name: Arc<str>,
85 class: Arc<Class<M>>,
87 mutable: Mutex<DevMutable<M>>,
90 model: RwLock<Option<Result<M>>>,
94 local_handle: Mutex<Option<LocalDevHandle>>,
97}
98
99impl<M: Model> Dev<M> {
100
101 pub(super) fn new(class: Arc<Class<M>>, name: Arc<str>) -> Arc<Self> {
102 let state_attr = Arc::new(Attr::new(Arc::clone(&class.def().state_attr)));
103 let status_attr = Arc::new(Attr::new(Arc::clone(&class.def().status_attr)));
104 Arc::new_cyclic(|this| Self {
105 this: this.clone(),
106 name,
107 class,
108 mutable: Mutex::new(DevMutable {
109 black_box: VecDeque::new(),
110 commands: HashMap::new(),
111 attrs: HashMap::new(),
112 state_attr,
113 status_attr,
114 auto_alarm_states: Arc::from([DevState::On]),
115 }),
116 model: RwLock::new(None),
117 local_handle: Mutex::new(None),
118 })
119 }
120
121 #[inline]
122 pub fn name(&self) -> &str {
123 &self.name
124 }
125
126 #[inline]
127 pub fn class(&self) -> &Class<M> {
128 &self.class
129 }
130
131 #[inline]
132 pub fn server(&self) -> &Server {
133 self.class.server()
134 }
135
136 pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
143 self.server().database().list_dev_prop_async(&self.name, filter).await
144 }
145
146 pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
148 self.server().database().list_dev_prop(&self.name, filter)
149 }
150
151 pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
154 where
155 Spec: GetPropSpec<Names>,
156 {
157 self.server().database().get_dev_prop_async::<Spec, Names>(&self.name, names)
158 }
159
160 pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
162 where
163 Spec: GetPropSpec<Names>,
164 {
165 self.server().database().get_dev_prop::<Spec, Names>(&self.name, names)
166 }
167
168 pub fn set_prop_async<'a, Spec>(&'a self, spec: Spec) -> impl Future<Output = Result<()>> + Send
170 where
171 Spec: SetPropSpec<'a>,
172 {
173 self.server().database().set_dev_prop_async(&self.name, spec)
174 }
175
176 pub fn set_prop<'a, Spec>(&'a self, spec: Spec) -> Result<()>
178 where
179 Spec: SetPropSpec<'a>,
180 {
181 self.server().database().set_dev_prop(&self.name, spec)
182 }
183
184 pub fn del_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
186 self.server().database().del_dev_prop_async(&self.name, names)
187 }
188
189 pub fn del_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
191 self.server().database().del_dev_prop(&self.name, names)
192 }
193
194 #[inline]
196 fn arc(&self) -> Arc<Self> {
197 self.this.upgrade().unwrap()
199 }
200
201 pub fn add_command(&self, def: CommandDef<M>) -> Result<()> {
210
211 if let Some(reserved) = reserved_name(&def.name, RESERVED_COMMANDS) {
216 return Err(Error::new(ErrorKind::CommandNameInvalid)
217 .with_description(format!("Command {} cannot be defined in the device {}, \
218 {reserved} is answered by the server itself", def.name, self.name)));
219 }
220
221 let mut mutable = self.mutable.lock().unwrap();
222 if mutable.commands.contains_key(Name::new(&def.name)) {
223 return Err(Error::new(ErrorKind::CommandAlreadyDefined)
224 .with_append_description(format_args!("Command {:?}", def.name)));
225 }
226
227 let name = ArcName::from(Arc::<str>::from(&*def.name));
228 mutable.commands.insert(name, Arc::new(Command::new(Arc::new(def))));
229 Ok(())
230
231 }
232
233 pub fn remove_command(&self, name: &str) -> bool {
235 self.mutable.lock().unwrap().commands.remove(Name::new(name)).is_some()
236 }
237
238 pub fn add_attr(&self, def: AttrDef<M>) -> Result<()> {
247
248 if let Some(reserved) = reserved_name(&def.name, RESERVED_ATTRS) {
249 return Err(Error::new(ErrorKind::AttrNameInvalid)
250 .with_description(format!("Attribute {} cannot be defined in the device {}, \
251 {reserved} is answered by the server itself", def.name, self.name)));
252 }
253
254 if let Err(err) = def.check() {
255 return Err(Error::new(ErrorKind::AttrNameInvalid)
256 .with_description(format!("Attribute {} cannot be defined in the device {}, {err}",
257 def.name, self.name)));
258 }
259
260 let mut mutable = self.mutable.lock().unwrap();
261 if mutable.attrs.contains_key(Name::new(&def.name)) {
262 return Err(Error::new(ErrorKind::AttrAlreadyDefined)
263 .with_append_description(format_args!("Attribute {:?}", def.name)));
264 }
265
266 let name = ArcName::from(Arc::<str>::from(&*def.name));
267 mutable.attrs.insert(name, Arc::new(Attr::new(Arc::new(def))));
268 Ok(())
269
270 }
271
272 pub async fn set_attr_w_value_async(&self, name: &str, value: impl IntoValue) -> Result<()> {
280 let attr = self.find_attr(name)?;
281 attr.set_w_value(self, value.into_value()).await
282 }
283
284 pub fn set_attr_w_value(&self, name: &str, value: impl IntoValue) -> Result<()> {
288 self.server().database().block_on(self.set_attr_w_value_async(name, value))?
289 }
290
291 pub async fn set_attr_config_async(&self, name: &str, config: &AttrConfig) -> Result<()> {
296 self.change_attr_config(name, config).await
297 }
298
299 pub fn set_attr_config(&self, name: &str, config: &AttrConfig) -> Result<()> {
304 self.server().database().block_on(self.set_attr_config_async(name, config))?
305 }
306
307 pub fn set_auto_alarm_states(&self, states: &[DevState]) {
317 self.mutable.lock().unwrap().auto_alarm_states = Arc::from(states);
318 }
319
320 pub fn remove_attr(&self, name: &str) -> bool {
322 self.mutable.lock().unwrap().attrs.remove(Name::new(name)).is_some()
323 }
324
325 pub fn get_command(&self, name: &str) -> Option<Ref<'_, Command<M>>> {
332 self.find_command(name).ok().map(Ref::new)
333 }
334
335 pub fn get_attr(&self, name: &str) -> Option<Ref<'_, Attr<M>>> {
338 self.find_attr(name).ok().map(Ref::new)
339 }
340
341 pub fn commands(&self) -> Vec<Ref<'_, Command<M>>> {
343 self.all_commands().into_iter().map(Ref::new).collect()
344 }
345
346 pub fn attrs(&self) -> Vec<Ref<'_, Attr<M>>> {
351 self.all_attrs().into_iter().map(Ref::new).collect()
352 }
353
354 pub fn push_attr_event(&self, name: &str, event: AttrValueEvent, mut read: Result<AttrRead<Value>>) -> Result<()> {
368 let attr = self.resolve_attr(name)?.into_attr();
369 attr.check_push_event(self, event)?;
370 if let Ok(read) = &mut read {
371 attr.merge_w_value(read);
372 }
373 self.server().push_attr_value(&self.name, attr.name(), event, read);
374 Ok(())
375 }
376
377 pub fn push_change_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
380 self.push_attr_event(name, AttrValueEvent::Change, Ok(read_now(value)))
381 }
382
383 pub fn push_archive_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
385 self.push_attr_event(name, AttrValueEvent::Archive, Ok(read_now(value)))
386 }
387
388 pub fn push_user_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
390 self.push_attr_event(name, AttrValueEvent::User, Ok(read_now(value)))
391 }
392
393 pub fn push_alarm_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
395 self.push_attr_event(name, AttrValueEvent::Alarm, Ok(read_now(value)))
396 }
397
398 pub fn push_data_ready_event(&self, name: &str) -> Result<()> {
401 let target = self.resolve_attr(name)?;
402 self.server().push_attr_ready(&self.name, target.name(), Ok(()));
403 Ok(())
404 }
405
406 async fn init(&self) {
424
425 let def = self.class.def();
426 let mut model = self.model.write().await;
427
428 self.delete_model(&mut model).await;
432
433 self.reset_interface();
441 self.load_all_attr_config().await;
442
443 let mut res = match &def.init {
444 InitFn::Sync(init) => init(self),
445 InitFn::Async(init) => init(self).await,
446 };
447
448 if let Ok(new_model) = &mut res {
452 self.load_all_attr_config().await;
453 for attr in self.all_attrs() {
454 attr.restore(self, new_model).await;
455 }
456 }
457
458 *model = Some(match res {
460 Ok(new_model) => {
461 info!(dev = %&*self.name, class = %&*def.name, "Initialized the device");
462 Ok(new_model)
463 }
464 Err(err) => {
465 error!(dev = %&*self.name, class = %&*def.name, %err, "Failed to initialize the device");
466 Err(err)
467 }
468 });
469
470 }
471
472 async fn delete_model(&self, model: &mut Option<Result<M>>) {
481
482 let Some(Ok(prev)) = model.take() else {
485 return;
486 };
487
488 let def = self.class.def();
489 debug!(dev = %&*self.name, class = %&*def.name, "Deleting the model of the device");
490
491 match &def.delete {
492 Some(DeleteFn::Sync(delete)) => delete(prev, self),
493 Some(DeleteFn::Async(delete)) => delete(prev, self).await,
494 None => {}
495 }
496
497 }
498
499 fn reset_interface(&self) {
503
504 let def = self.class.def();
505 let mut mutable = self.mutable.lock().unwrap();
506
507 mutable.commands.clear();
508 mutable.attrs.clear();
509 mutable.state_attr = Arc::new(Attr::new(Arc::clone(&def.state_attr)));
510 mutable.status_attr = Arc::new(Attr::new(Arc::clone(&def.status_attr)));
511
512 for def in &def.commands {
513 let name = ArcName::from(Arc::<str>::from(&*def.name));
514 mutable.commands.insert(name, Arc::new(Command::new(Arc::clone(def))));
515 }
516
517 for def in &def.attrs {
518 let name = ArcName::from(Arc::<str>::from(&*def.name));
519 mutable.attrs.insert(name, Arc::new(Attr::new(Arc::clone(def))));
520 }
521
522 }
523
524 async fn load_all_attr_config(&self) {
530
531 let attrs = {
534 let mutable = self.mutable.lock().unwrap();
535 [&mutable.state_attr, &mutable.status_attr]
536 .into_iter()
537 .chain(mutable.attrs.values())
538 .filter(|attr| !attr.is_loaded())
539 .map(Arc::clone)
540 .collect::<Vec<_>>()
541 };
542
543 self.load_given_attr_config(&attrs).await;
544
545 }
546
547 async fn load_given_attr_config(&self, attrs: &[Arc<Attr<M>>]) {
550
551 if attrs.is_empty() {
552 return;
553 }
554
555 let names = attrs
557 .iter()
558 .map(|attr| attr.name())
559 .collect::<Vec<_>>();
560
561 let (
563 class_props,
564 dev_props,
565 ) = join(
566 self.class.attr_props(&names),
567 self.server()
568 .database()
569 .get_dev_attr_prop_async::<Spectrum<str>, _, _>(&self.name, names.as_slice(), All),
570 ).await;
571
572 let dev_props = dev_props.and_then(|props| props
573 .into_iter()
574 .collect::<Result<Vec<_>>>());
575
576 for (index, attr) in attrs.iter().enumerate() {
577 attr.load_config(self,
578 class_props.as_ref().map(|props| &*props[index]),
579 dev_props.as_ref().map(|props| &*props[index]));
580 }
581
582 }
583
584 async fn state(&self, ctx: &mut ReadCtx<M>) -> Result<DevState> {
592 let mut guard = self.lock(self.state_access(ctx)).await;
593 match guard.get() {
594 Some(Ok(_)) => self.model_state(&mut guard.model(), ctx).await,
595 Some(Err(_)) => Ok(DevState::Fault),
596 None => Ok(DevState::Unknown),
597 }
598 }
599
600 async fn status(&self, ctx: &mut ReadCtx<M>) -> Result<String> {
604 let mut guard = self.lock(self.status_access(ctx)).await;
605 match guard.get() {
606 Some(Ok(_)) => self.model_status(&mut guard.model(), ctx).await,
607 Some(Err(err)) => Ok(make_failed_status(err)),
608 None => Ok(MISSING_STATUS.to_string()),
609 }
610 }
611
612 async fn model_state(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<DevState> {
620 let state = self.model_state_inner(model, ctx).await?;
621 if self.attrs_in_alarm(ctx, state, model, false).await.is_empty() {
622 Ok(state)
623 } else {
624 Ok(DevState::Alarm)
625 }
626 }
627
628 async fn model_status(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<String> {
631
632 let mut status = match &self.class.def().status {
633 Some(StatusFn::SharedSync(func)) => func(model.shared(), self).map(Cow::into_owned),
634 Some(StatusFn::ExclusiveSync(func)) => func(model.exclusive()?, self).map(Cow::into_owned),
635 Some(StatusFn::SharedAsync(func)) => func(model.shared(), self).await.map(Cow::into_owned),
636 Some(StatusFn::ExclusiveAsync(func)) => func(model.exclusive()?, self).await.map(Cow::into_owned),
637 None => Ok(DEFAULT_STATUS.to_string()),
638 }?;
639
640 if ctx.has_alarm() {
641 let state = self.model_state_inner(model, ctx).await?;
642 for (attr, alarm) in self.attrs_in_alarm(ctx, state, model, true).await {
643 status.push('\n');
644 status.push_str(&alarm.describe(attr.name()));
645 }
646 }
647
648 Ok(status)
649
650 }
651
652 async fn model_state_inner(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<DevState> {
655
656 if let Some(state) = ctx.state {
657 return Ok(state);
658 }
659
660 let state = match &self.class.def().state {
661 Some(StateFn::SharedSync(func)) => func(model.shared(), self),
662 Some(StateFn::ExclusiveSync(func)) => func(model.exclusive()?, self),
663 Some(StateFn::SharedAsync(func)) => func(model.shared(), self).await,
664 Some(StateFn::ExclusiveAsync(func)) => func(model.exclusive()?, self).await,
665 None => Ok(DEFAULT_STATE),
666 }?;
667
668 ctx.state = Some(state);
669 Ok(state)
670
671 }
672
673 async fn attrs_in_alarm(&self, ctx: &mut ReadCtx<M>, state: DevState, model: &mut ModelRef<'_, M>, every: bool) -> Vec<(Arc<Attr<M>>, AttrAlarm)> {
679
680 let mut found = Vec::new();
681 if !ctx.may_alarm(state) {
682 return found;
683 }
684
685 for alarm_attr in ctx.alarm.iter_mut().flat_map(|alarm| &mut alarm.attrs) {
686
687 let alarm = match alarm_attr.read {
688 Some(alarm) => alarm,
689 None => {
690
691 let alarm = alarm_attr.attr.read_with_alarm(self, model).await
694 .ok()
695 .and_then(|(_, alarm)| alarm);
696
697 alarm_attr.read = Some(alarm);
698 alarm
699
700 }
701 };
702
703 if let Some(alarm) = alarm {
704 found.push((Arc::clone(&alarm_attr.attr), alarm));
705 if !every {
706 break;
707 }
708 }
709
710 }
711
712 found
713
714 }
715
716 fn state_access(&self, ctx: &ReadCtx<M>) -> Access {
719 let access = match &self.class.def().state {
720 Some(StateFn::ExclusiveSync(_) | StateFn::ExclusiveAsync(_)) => Access::Exclusive,
721 _ => Access::Shared,
722 };
723 access.merge(ctx.access())
724 }
725
726 fn status_access(&self, ctx: &ReadCtx<M>) -> Access {
729 let access = match &self.class.def().status {
730 Some(StatusFn::ExclusiveSync(_) | StatusFn::ExclusiveAsync(_)) => Access::Exclusive,
731 _ => Access::Shared,
732 };
733 if ctx.has_alarm() { access.merge(self.state_access(ctx)) } else { access }
734 }
735
736 fn init_may_block(&self) -> bool {
742 let def = self.class.def();
743 matches!(def.init, InitFn::Sync(_)) || self.delete_may_block() || def.attrs.iter().any(|attr| {
744 attr.memorized == Some(AttrMemorized::RestoreAndWrite)
745 && matches!(attr.write, Some(WriteFn::SharedSync(_) | WriteFn::ExclusiveSync(_)))
746 })
747 }
748
749 fn delete_may_block(&self) -> bool {
752 matches!(self.class.def().delete, Some(DeleteFn::Sync(_)))
753 }
754
755 fn state_may_block(&self, ctx: &ReadCtx<M>) -> bool {
758 matches!(&self.class.def().state, Some(StateFn::SharedSync(_) | StateFn::ExclusiveSync(_)))
759 || ctx.may_block()
760 }
761
762 fn status_may_block(&self, ctx: &ReadCtx<M>) -> bool {
765 matches!(&self.class.def().status, Some(StatusFn::SharedSync(_) | StatusFn::ExclusiveSync(_)))
766 || (ctx.has_alarm() && self.state_may_block(ctx))
767 }
768
769 fn read_may_block(&self, target: &AttrTarget<M>, ctx: &ReadCtx<M>) -> bool {
771 match target {
772 AttrTarget::State(_) => self.state_may_block(ctx),
773 AttrTarget::Status(_) => self.status_may_block(ctx),
774 AttrTarget::Attr(attr) => attr.read_may_block(),
775 }
776 }
777
778 fn command_may_block(&self, target: &CommandTarget<M>, ctx: &ReadCtx<M>) -> bool {
780 match target {
781 CommandTarget::State(_) => self.state_may_block(ctx),
782 CommandTarget::Status(_) => self.status_may_block(ctx),
783 CommandTarget::Init => self.init_may_block(),
784 CommandTarget::Command(command) => command.may_block(),
785 }
786 }
787
788 fn read_access(&self, target: &AttrTarget<M>, ctx: &ReadCtx<M>) -> Access {
790 match target {
791 AttrTarget::State(_) => self.state_access(ctx),
792 AttrTarget::Status(_) => self.status_access(ctx),
793 AttrTarget::Attr(attr) => attr.read_access(),
794 }
795 }
796
797 async fn lock(&self, access: Access) -> ModelGuard<'_, M> {
799 match access {
800 Access::Shared => ModelGuard::Shared(self.model.read().await),
801 Access::Exclusive => ModelGuard::Exclusive(self.model.write().await),
802 }
803 }
804
805 async fn lock_model(&self, access: Access) -> Result<ModelGuard<'_, M>> {
812 let guard = self.lock(access).await;
813 match guard.get() {
814 Some(Ok(_)) => Ok(guard),
815 Some(Err(err)) => Err(init_err(Some(err), &self.name)),
816 None => Err(init_err(None, &self.name)),
817 }
818 }
819
820 pub(super) async fn check_model(&self) -> Result<()> {
823 match &*self.model.read().await {
824 Some(Ok(_)) => Ok(()),
825 Some(Err(err)) => Err(init_err(Some(err), &self.name)),
826 None => Err(init_err(None, &self.name)),
827 }
828 }
829
830 fn find_command(&self, name: &str) -> Result<Arc<Command<M>>> {
833 self.mutable.lock().unwrap().commands.get(Name::new(name))
834 .map(Arc::clone)
835 .ok_or_else(|| Error::new(ErrorKind::CommandNotFound)
836 .with_append_description(format_args!("Command {name:?} of the device {:?}", self.name)))
837 }
838
839 fn find_attr(&self, name: &str) -> Result<Arc<Attr<M>>> {
841 self.mutable.lock().unwrap().attrs.get(Name::new(name))
842 .map(Arc::clone)
843 .ok_or_else(|| Error::new(ErrorKind::AttrNotFound)
844 .with_append_description(format_args!("Attribute {name:?} of the device {:?}", self.name)))
845 }
846
847 fn all_commands(&self) -> Vec<Arc<Command<M>>> {
849 self.mutable.lock().unwrap().commands.values().map(Arc::clone).collect()
850 }
851
852 fn all_attrs(&self) -> Vec<Arc<Attr<M>>> {
854 self.mutable.lock().unwrap().attrs.values().map(Arc::clone).collect()
855 }
856
857 fn state_attr(&self) -> Arc<Attr<M>> {
860 Arc::clone(&self.mutable.lock().unwrap().state_attr)
861 }
862
863 fn status_attr(&self) -> Arc<Attr<M>> {
865 Arc::clone(&self.mutable.lock().unwrap().status_attr)
866 }
867
868 fn resolve_attr(&self, name: &str) -> Result<AttrTarget<M>> {
872 if name.eq_ignore_ascii_case(STATE_NAME) {
873 Ok(AttrTarget::State(self.state_attr()))
874 } else if name.eq_ignore_ascii_case(STATUS_NAME) {
875 Ok(AttrTarget::Status(self.status_attr()))
876 } else {
877 self.find_attr(name).map(AttrTarget::Attr)
878 }
879 }
880
881 fn resolve_writable_attr(&self, name: &str) -> Result<Arc<Attr<M>>> {
884 match self.resolve_attr(name)? {
885 AttrTarget::Attr(attr) => Ok(attr),
886 AttrTarget::State(_) | AttrTarget::Status(_) => Err(not_writable_err(self, name)),
887 }
888 }
889
890 fn resolve_command(&self, name: &str) -> Result<CommandTarget<M>> {
894 if name.eq_ignore_ascii_case(STATE_NAME) {
895 Ok(CommandTarget::State(self.state_attr()))
896 } else if name.eq_ignore_ascii_case(STATUS_NAME) {
897 Ok(CommandTarget::Status(self.status_attr()))
898 } else if name.eq_ignore_ascii_case(INIT_NAME) {
899 Ok(CommandTarget::Init)
900 } else {
901 self.find_command(name).map(CommandTarget::Command)
902 }
903 }
904
905 fn all_attr_targets(&self) -> Vec<Resolved<AttrTarget<M>>> {
908 [Ok(AttrTarget::State(self.state_attr())), Ok(AttrTarget::Status(self.status_attr()))]
909 .into_iter()
910 .chain(self.all_attrs().into_iter().map(|attr| Ok(AttrTarget::Attr(attr))))
911 .collect()
912 }
913
914 fn attr_targets(&self, names: Option<Spectrum<str>>) -> Vec<Resolved<AttrTarget<M>>> {
918 match names {
919 Some(names) => scalars(&names)
920 .map(|name| {
921 let target = self.resolve_attr(name.value());
922 resolved(name, target)
923 })
924 .collect(),
925 None => self.all_attr_targets(),
926 }
927 }
928
929 fn command_targets(&self, names: Option<Spectrum<str>>) -> Vec<Resolved<CommandTarget<M>>> {
932 match names {
933 Some(names) => scalars(&names)
934 .map(|name| {
935 let target = self.resolve_command(name.value());
936 resolved(name, target)
937 })
938 .collect(),
939 None => [Ok(CommandTarget::State(self.state_attr())), Ok(CommandTarget::Status(self.status_attr())), Ok(CommandTarget::Init)]
940 .into_iter()
941 .chain(self.all_commands().into_iter().map(|command| Ok(CommandTarget::Command(command))))
942 .collect(),
943 }
944 }
945
946 async fn exec_command(&self, target: Result<CommandTarget<M>>, input: Value, mut ctx: ReadCtx<M>) -> Result<Value> {
948
949 let target = match target {
953 Ok(target) => target,
954 Err(err) => return Err(self.check_model().await.err().unwrap_or(err)),
955 };
956
957 match target {
958 CommandTarget::State(_) => Ok(self.state(&mut ctx).await?.into()),
962 CommandTarget::Status(_) => Ok(self.status(&mut ctx).await?.into()),
963 CommandTarget::Init => {
964 self.init().await;
965 self.check_model().await?;
969 Ok(Value::Void)
970 }
971 CommandTarget::Command(command) => {
972 let mut guard = self.lock_model(command.access()).await?;
973 command.exec(self, &mut guard.model(), input).await
974 }
975 }
976
977 }
978
979 async fn exec_read_attr(&self, targets: Vec<Resolved<AttrTarget<M>>>, mut ctx: ReadCtx<M>) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
981
982 let access = targets.iter()
986 .filter_map(|target| target.as_ref().ok())
987 .fold(Access::Shared, |access, target| access.merge(self.read_access(target, &ctx)));
988
989 let mut guard = self.lock_model(access).await?;
990 let mut model = guard.model();
991
992 let mut reads = targets
997 .iter()
998 .map(|_| None)
999 .collect::<Vec<_>>();
1000
1001 for (target, slot) in targets.iter().zip(&mut reads) {
1002 if let Ok(AttrTarget::Attr(attr)) = target {
1003 let read = attr.read_with_alarm(self, &mut model).await;
1004 ctx.record(attr, read.as_ref().ok().and_then(|(_, alarm)| *alarm));
1005 *slot = Some(Named::new(Scalar::from(attr.name()), read.map(|(read, _)| read)));
1006 }
1007 }
1008
1009 for (target, slot) in targets.into_iter().zip(&mut reads) {
1010 *slot = Some(match target {
1011 Ok(AttrTarget::Attr(_)) => continue,
1012 Ok(AttrTarget::State(_)) => Named::new(Scalar::from(STATE_NAME), self.model_state(&mut model, &mut ctx).await
1013 .map(|state| AttrRead::new(state.into()))),
1014 Ok(AttrTarget::Status(_)) => Named::new(Scalar::from(STATUS_NAME), self.model_status(&mut model, &mut ctx).await
1015 .map(|status| AttrRead::new(status.into()))),
1016 Err(unresolved) => unresolved.map(Err),
1017 });
1018 }
1019
1020 Ok(reads.into_iter().map(|read| read.expect("every target is read")).collect())
1021
1022 }
1023
1024 async fn exec_write_attr(&self, targets: Vec<(Result<Arc<Attr<M>>>, Value)>) -> Result<Vec<Result<()>>> {
1026
1027 let access = targets.iter()
1030 .filter_map(|(target, _)| target.as_ref().ok())
1031 .fold(Access::Shared, |access, attr| access.merge(attr.write_access()));
1032
1033 let mut guard = self.lock_model(access).await?;
1034 let mut model = guard.model();
1035
1036 let mut writes = Vec::with_capacity(targets.len());
1037 for (target, value) in targets {
1038 writes.push(match target {
1039 Ok(attr) => attr.write(self, &mut model, value).await,
1040 Err(err) => Err(err),
1041 });
1042 }
1043
1044 Ok(writes)
1045
1046 }
1047
1048 async fn exec_set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> Vec<Result<()>> {
1054 let mut results = Vec::with_capacity(configs.len());
1055 for Named { name, value } in configs {
1056 results.push(self.change_attr_config(name.value(), &value).await);
1057 }
1058 results
1059 }
1060
1061 async fn change_attr_config(&self, name: &str, config: &AttrConfig) -> Result<()> {
1066
1067 let target = self.resolve_attr(name)?;
1068 let attr = target.attr();
1069
1070 let class_props = self.class.attr_props(&[attr.name()]).await?;
1071 let changes = attr.change_config(&class_props[0], config)
1072 .map_err(|err| Error::new(ErrorKind::Reason(reason::API_IncoherentValues))
1073 .with_description(format!("The configuration of the attribute {:?} of the device {:?} \
1074 cannot be changed: {err}", attr.name(), self.name)))?;
1075
1076 if changes.is_empty() {
1077 return Ok(());
1078 }
1079
1080 self.server().push_attr_conf(&self.name, attr.name(), target.info());
1081
1082 let database = self.server().database();
1083
1084 if !changes.put.is_empty() {
1085 let put = changes.put.into_iter()
1086 .map(|(prop, values)| (prop.name(), values))
1087 .collect::<Vec<_>>();
1088 database.set_dev_attr_prop_async(&self.name, (attr.name(), put)).await?;
1089 }
1090
1091 if !changes.del.is_empty() {
1092 let del = changes.del.iter().map(|prop| prop.name()).collect::<Vec<_>>();
1093 database.del_dev_attr_prop_async(&self.name, attr.name(), del.as_slice()).await?;
1094 }
1095
1096 Ok(())
1097
1098 }
1099
1100 async fn exec_write_read_attr(&self, targets: Vec<(Resolved<Arc<Attr<M>>>, Value)>) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
1103
1104 let access = targets.iter()
1108 .filter_map(|(target, _)| target.as_ref().ok())
1109 .fold(Access::Shared, |access, attr| access.merge(attr.write_access()).merge(attr.read_access()));
1110
1111 let mut guard = self.lock_model(access).await?;
1112 let mut model = guard.model();
1113
1114 let mut reads = Vec::with_capacity(targets.len());
1115 for (target, value) in targets {
1116 reads.push(match target {
1117 Ok(attr) => {
1120 let read = match attr.write(self, &mut model, value).await {
1121 Ok(()) => attr.read(self, &mut model).await,
1122 Err(err) => Err(err),
1123 };
1124 Named::new(Scalar::from(attr.name()), read)
1125 }
1126 Err(unresolved) => unresolved.map(Err),
1127 });
1128 }
1129
1130 Ok(reads)
1131
1132 }
1133}
1134
1135impl<M: Model> AnyDev for Dev<M> {
1136
1137 #[inline]
1138 fn name(&self) -> &str {
1139 &self.name
1140 }
1141
1142 #[inline]
1143 fn class(&self) -> &dyn AnyClass {
1144 &*self.class
1145 }
1146
1147 fn init(&self) -> BoxFuture<'_, Result<()>> {
1148 let dev = self.arc();
1149 pool::run(self.init_may_block(), async move {
1150 dev.init().await;
1151 Ok(())
1152 })
1153 }
1154
1155 fn withdraw(&self) -> BoxFuture<'_, Result<()>> {
1156 let dev = self.arc();
1157 Box::pin(async move {
1158
1159 dev.local_handle.lock().unwrap().take();
1161 if let Some(backend) = dev.server().backend() {
1162 backend.remove(&dev.name);
1163 }
1164
1165 pool::run(dev.delete_may_block(), async move {
1166 dev.delete_model(&mut *dev.model.write().await).await;
1167 Ok(())
1168 }).await
1169
1170 })
1171 }
1172
1173 fn serve(&self) -> Result<Arc<str>> {
1174 let backend = self.server().served_backend()?;
1175 let dev = DevImpl::from_arc(self.arc()) as Arc<dyn BackendDev>;
1176 let itr = backend.add(&self.name, Arc::clone(&dev))?;
1177 if self.server().local {
1178 let local_handle = local::register(&dev, Arc::clone(&itr), backend.host(), backend.port(), Arc::clone(&self.name));
1179 *self.local_handle.lock().unwrap() = Some(local_handle);
1180 }
1181 Ok(itr)
1182 }
1183
1184}
1185
1186impl<M: Model> Debug for Dev<M> {
1187 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1188 f.debug_struct("Dev").field("name", &self.name).finish_non_exhaustive()
1189 }
1190}
1191
1192struct DevMutable<M: Model> {
1195 black_box: VecDeque<String>,
1196 commands: HashMap<ArcName, Arc<Command<M>>>,
1197 attrs: HashMap<ArcName, Arc<Attr<M>>>,
1198 state_attr: Arc<Attr<M>>,
1201 status_attr: Arc<Attr<M>>,
1202 auto_alarm_states: Arc<[DevState]>,
1205}
1206
1207struct ReadCtx<M: Model> {
1211 alarm: Option<AlarmCtx<M>>,
1214 state: Option<DevState>,
1217}
1218
1219struct AlarmCtx<M: Model> {
1224 states: Arc<[DevState]>,
1225 attrs: Vec<AlarmAttr<M>>,
1226}
1227
1228struct AlarmAttr<M: Model> {
1230 attr: Arc<Attr<M>>,
1231 read: Option<Option<AttrAlarm>>,
1236}
1237
1238impl<M: Model> ReadCtx<M> {
1239
1240 fn new(dev: &Dev<M>, involves_state: bool) -> Self {
1245
1246 let alarm = involves_state.then(|| {
1247
1248 let mutable = dev.mutable.lock().unwrap();
1249 if mutable.auto_alarm_states.is_empty() {
1250 return None;
1251 }
1252
1253 let mut attrs = mutable.attrs.values()
1254 .filter(|attr| attr.has_alarms())
1255 .map(|attr| AlarmAttr {
1256 attr: Arc::clone(attr),
1257 read: None,
1258 })
1259 .collect::<Vec<_>>();
1260
1261 if attrs.is_empty() {
1262 return None;
1263 }
1264
1265 attrs.sort_by(|a, b| a.attr.name().cmp(b.attr.name()));
1266 Some(AlarmCtx {
1267 states: Arc::clone(&mutable.auto_alarm_states),
1268 attrs,
1269 })
1270
1271 }).flatten();
1272
1273 Self {
1274 alarm,
1275 state: None,
1276 }
1277
1278 }
1279
1280 fn has_alarm(&self) -> bool {
1282 self.alarm.is_some()
1283 }
1284
1285 fn may_alarm(&self, state: DevState) -> bool {
1288 self.alarm.as_ref().is_some_and(|alarm| alarm.states.contains(&state))
1289 }
1290
1291 fn record(&mut self, attr: &Arc<Attr<M>>, alarm: Option<AttrAlarm>) {
1294
1295 let alarm_attr = self.alarm.iter_mut()
1296 .flat_map(|alarm| &mut alarm.attrs)
1297 .find(|alarm_attr| Arc::ptr_eq(&alarm_attr.attr, attr));
1298
1299 if let Some(alarm_attr) = alarm_attr {
1300 alarm_attr.read = Some(alarm);
1301 }
1302
1303 }
1304
1305 fn access(&self) -> Access {
1308 self.alarm.iter()
1309 .flat_map(|alarm| &alarm.attrs)
1310 .fold(Access::Shared, |access, alarm_attr| access.merge(alarm_attr.attr.read_access()))
1311 }
1312
1313 fn may_block(&self) -> bool {
1316 self.alarm.iter()
1317 .flat_map(|alarm| &alarm.attrs)
1318 .any(|alarm_attr| alarm_attr.attr.read_may_block())
1319 }
1320
1321}
1322
1323enum AttrTarget<M: Model> {
1327 State(Arc<Attr<M>>),
1328 Status(Arc<Attr<M>>),
1329 Attr(Arc<Attr<M>>),
1330}
1331
1332impl<M: Model> AttrTarget<M> {
1333
1334 fn involves_state(&self) -> bool {
1337 matches!(self, Self::State(_) | Self::Status(_))
1338 }
1339
1340 fn attr(&self) -> &Arc<Attr<M>> {
1342 match self {
1343 Self::State(attr) |
1344 Self::Status(attr) |
1345 Self::Attr(attr) => attr,
1346 }
1347 }
1348
1349 fn into_attr(self) -> Arc<Attr<M>> {
1351 match self {
1352 Self::State(attr) |
1353 Self::Status(attr) |
1354 Self::Attr(attr) => attr,
1355 }
1356 }
1357
1358 fn name(&self) -> &str {
1360 match self {
1361 Self::State(attr) |
1362 Self::Status(attr) |
1363 Self::Attr(attr) => attr.name(),
1364 }
1365 }
1366
1367 fn info(&self) -> Result<AttrInfo> {
1370 match self {
1371 Self::State(attr) |
1372 Self::Status(attr) => attr.info_as(AttrMode::Read),
1373 Self::Attr(attr) => attr.info(),
1374 }
1375 }
1376
1377}
1378
1379enum CommandTarget<M: Model> {
1383 State(Arc<Attr<M>>),
1384 Status(Arc<Attr<M>>),
1385 Init,
1386 Command(Arc<Command<M>>),
1387}
1388
1389impl<M: Model> CommandTarget<M> {
1390
1391 fn involves_state(&self) -> bool {
1394 matches!(self, Self::State(_) | Self::Status(_))
1395 }
1396
1397 fn name(&self) -> &str {
1399 match self {
1400 Self::State(attr) |
1401 Self::Status(attr) => attr.name(),
1402 Self::Init => INIT_NAME,
1403 Self::Command(command) => command.name(),
1404 }
1405 }
1406
1407 fn info(&self) -> CommandInfo {
1410 match self {
1411 Self::State(_) => CommandInfo {
1412 display_level: DisplayLevel::Operator,
1413 in_kind: CommandDataKind::Void,
1414 out_kind: CommandDataKind::State,
1415 in_doc: None,
1416 out_doc: Some(Cow::Borrowed("The state of the device")),
1417 },
1418 Self::Status(_) => CommandInfo {
1419 display_level: DisplayLevel::Operator,
1420 in_kind: CommandDataKind::Void,
1421 out_kind: CommandDataKind::String,
1422 in_doc: None,
1423 out_doc: Some(Cow::Borrowed("The status of the device")),
1424 },
1425 Self::Init => CommandInfo {
1426 display_level: DisplayLevel::Operator,
1427 in_kind: CommandDataKind::Void,
1428 out_kind: CommandDataKind::Void,
1429 in_doc: None,
1430 out_doc: None,
1431 },
1432 Self::Command(command) => command.info(),
1433 }
1434 }
1435
1436}
1437
1438fn read_now(value: impl IntoValue) -> AttrRead<Value> {
1442 AttrRead {
1443 value: Some(value.into_value()),
1444 w_value: None,
1445 quality: AttrQuality::Valid,
1446 time: SystemTime::now(),
1447 }
1448}
1449
1450fn not_writable_err<M: Model>(dev: &Dev<M>, name: &str) -> Error {
1451 Error::new(ErrorKind::AttrNotWritable)
1452 .with_append_description(format_args!("Attribute {name:?} of the device {:?}",
1453 dev.name()))
1454}
1455
1456fn check_polled(source: DevSource, what: fmt::Arguments<'_>) -> Result<()> {
1462 match source {
1463 DevSource::Dev | DevSource::CacheDev => Ok(()),
1464 DevSource::Cache => Err(Error::new(ErrorKind::Reason(reason::API_DeviceNotPolled))
1465 .with_description(format!("Cannot answer from the polling cache, this server \
1466 polls nothing yet ({what})"))),
1467 }
1468}
1469
1470#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1474pub enum Access {
1475 Shared,
1476 Exclusive,
1477}
1478
1479impl Access {
1480
1481 pub fn merge(self, other: Self) -> Self {
1485 match (self, other) {
1486 (Self::Shared, Self::Shared) => Self::Shared,
1487 _ => Self::Exclusive,
1488 }
1489 }
1490
1491}
1492
1493pub enum ModelGuard<'a, M> {
1496 Shared(RwLockReadGuard<'a, Option<Result<M>>>),
1497 Exclusive(RwLockWriteGuard<'a, Option<Result<M>>>),
1498}
1499
1500impl<M> ModelGuard<'_, M> {
1501
1502 #[inline]
1504 fn get(&self) -> &Option<Result<M>> {
1505 match self {
1506 Self::Shared(guard) => guard,
1507 Self::Exclusive(guard) => guard,
1508 }
1509 }
1510
1511 fn model(&mut self) -> ModelRef<'_, M> {
1517 match self {
1518 Self::Shared(guard) => match &**guard {
1519 Some(Ok(model)) => ModelRef::Shared(model),
1520 _ => panic!("the device has no model"),
1521 }
1522 Self::Exclusive(guard) => match &mut **guard {
1523 Some(Ok(model)) => ModelRef::Exclusive(model),
1524 _ => panic!("the device has no model"),
1525 }
1526 }
1527 }
1528
1529}
1530
1531pub enum ModelRef<'m, M> {
1535 Shared(&'m M),
1536 Exclusive(&'m mut M),
1537}
1538
1539impl<M> ModelRef<'_, M> {
1540
1541 #[inline]
1543 pub fn shared(&self) -> &M {
1544 match self {
1545 Self::Shared(model) => model,
1546 Self::Exclusive(model) => model,
1547 }
1548 }
1549
1550 #[inline]
1556 pub fn exclusive(&mut self) -> Result<&mut M> {
1557 match self {
1558 Self::Exclusive(model) => Ok(model),
1559 Self::Shared(_) => Err(Error::new(ErrorKind::Internal)
1560 .with_description("The model was borrowed shared while a handler of that \
1561 request needs it exclusively")),
1562 }
1563 }
1564
1565}
1566
1567fn make_failed_status(err: &Error) -> String {
1570 format!("The initialization of the device failed:\n{err}")
1571}
1572
1573fn init_err(failure: Option<&Error>, name: &str) -> Error {
1576 match failure {
1577 Some(err) => Error::new(ErrorKind::DevInitFailed)
1578 .with_append_description(format_args!("Device {name:?}"))
1579 .with_append_description(err),
1580 None => Error::new(ErrorKind::DevNotConnected)
1581 .with_description(format!("The device {name} is not initialized")),
1582 }
1583}
1584
1585pub fn some_doc(doc: &Cow<'static, str>) -> Option<Cow<'static, str>> {
1587 (!doc.is_empty()).then(|| doc.clone())
1588}
1589
1590#[repr(transparent)]
1599struct DevImpl<M: Model>(Dev<M>);
1600
1601impl<M: Model> DevImpl<M> {
1602
1603 #[inline]
1605 fn from_arc(dev: Arc<Dev<M>>) -> Arc<Self> {
1606 unsafe { Arc::from_raw(Arc::into_raw(dev) as *const Self) }
1609 }
1610
1611}
1612
1613impl<M: Model> Debug for DevImpl<M> {
1614 #[inline]
1615 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1616 Debug::fmt(&self.0, f)
1617 }
1618}
1619
1620impl<M: Model> BackendDev for DevImpl<M> {
1624
1625 fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
1626 Box::pin(ready(Ok(Arc::clone(&self.0.name))))
1627 }
1628
1629 fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
1630 Box::pin(ready(Ok(Arc::clone(&self.0.server().admin_dev_name))))
1631 }
1632
1633 fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
1634 Box::pin(ready(Ok(Arc::from(&*self.0.class.def().doc))))
1635 }
1636
1637 fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {
1641 let def = self.0.class.def();
1642 let server = self.0.server();
1643 let backend = match server.served_backend() {
1644 Ok(backend) => backend,
1645 Err(err) => return Box::pin(ready(Err(err))),
1646 };
1647 Box::pin(ready(Ok(DevInfo {
1648 class: def.name.to_string(),
1649 server_id: server.name().to_string(),
1650 server_host: backend.host().to_string(),
1651 server_version: backend.version(),
1652 doc_url: def.doc_url.clone(),
1653 dev_type: Some(def.doc.clone()),
1656 version_info: vec![
1657 (Cow::Borrowed("rstango"), Cow::Borrowed(env!("CARGO_PKG_VERSION"))),
1658 ],
1659 })))
1660 }
1661
1662 fn ping(&self) -> BoxFuture<'_, Result<()>> {
1663 Box::pin(ready(Ok(())))
1664 }
1665
1666 fn state(&self) -> BoxFuture<'_, Result<DevState>> {
1667 let dev = self.0.arc();
1668 let mut ctx = ReadCtx::new(&dev, true);
1669 pool::run(dev.state_may_block(&ctx), async move { dev.state(&mut ctx).await })
1670 }
1671
1672 fn status(&self) -> BoxFuture<'_, Result<String>> {
1673 let dev = self.0.arc();
1674 let mut ctx = ReadCtx::new(&dev, true);
1675 pool::run(dev.status_may_block(&ctx), async move { dev.status(&mut ctx).await })
1676 }
1677
1678 fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Vec<String>>> {
1679
1680 let mutable = self.0.mutable.lock().unwrap();
1681 let black_box = mutable.black_box.iter()
1682 .take(max.unwrap_or(usize::MAX))
1683 .cloned()
1684 .collect();
1685
1686 Box::pin(ready(Ok(black_box)))
1687
1688 }
1689
1690 fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {
1691
1692 if let Err(err) = check_polled(source, format_args!("the command {} of the device {}", name.value(), self.0.name)) {
1693 return Box::pin(ready(Err(err)));
1694 }
1695
1696 let dev = self.0.arc();
1697 let target = dev.resolve_command(name.value());
1698 let ctx = ReadCtx::new(&dev, target.as_ref().is_ok_and(CommandTarget::involves_state));
1699
1700 let blocking = target.as_ref().is_ok_and(|target| dev.command_may_block(target, &ctx));
1701 pool::run(blocking, async move {
1702 let res = dev.exec_command(target, input, ctx).await;
1703 if let Err(err) = &res {
1704 debug!(dev = %&*dev.name, command = %name.value(), %err, "Failed to execute the command");
1705 }
1706 res
1707 })
1708
1709 }
1710
1711 fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
1712
1713 if let Err(err) = check_polled(source, format_args!("the attributes of the device {}", self.0.name)) {
1714 return Box::pin(ready(Err(err)));
1715 }
1716
1717 let dev = self.0.arc();
1718 let targets = dev.attr_targets(names);
1719 let ctx = ReadCtx::new(&dev, targets.iter().flatten().any(AttrTarget::involves_state));
1720
1721 let blocking = targets.iter().flatten().any(|target| dev.read_may_block(target, &ctx));
1722 pool::run(blocking, async move { dev.exec_read_attr(targets, ctx).await })
1723
1724 }
1725
1726 fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
1727
1728 let dev = self.0.arc();
1729 let targets = values.into_iter()
1730 .map(|write| (dev.resolve_writable_attr(write.name.value()), write.value))
1731 .collect::<Vec<_>>();
1732
1733 let blocking = targets.iter()
1734 .filter_map(|(target, _)| target.as_ref().ok())
1735 .any(|attr| attr.write_may_block());
1736 pool::run(blocking, async move { dev.exec_write_attr(targets).await })
1737
1738 }
1739
1740 fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
1741
1742 let dev = self.0.arc();
1743 let targets = values.into_iter()
1744 .map(|write| {
1745 let target = dev.resolve_writable_attr(write.name.value());
1746 (resolved(write.name, target), write.value)
1747 })
1748 .collect::<Vec<_>>();
1749
1750 let blocking = targets.iter()
1751 .filter_map(|(target, _)| target.as_ref().ok())
1752 .any(|attr| attr.write_may_block() || attr.read_may_block());
1753 pool::run(blocking, async move { dev.exec_write_read_attr(targets).await })
1754
1755 }
1756
1757 fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {
1761
1762 let infos = self.0.command_targets(names).into_iter()
1763 .map(|target| match target {
1764 Ok(target) => Named::new(Scalar::from(target.name()), Ok(target.info())),
1765 Err(unresolved) => unresolved.map(Err),
1766 })
1767 .collect();
1768
1769 Box::pin(ready(Ok(infos)))
1770
1771 }
1772
1773 fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {
1774
1775 let infos = self.0.attr_targets(names).into_iter()
1776 .map(|target| match target {
1777 Ok(target) => Named::new(Scalar::from(target.name()), target.info()),
1778 Err(unresolved) => unresolved.map(Err),
1779 })
1780 .collect();
1781
1782 Box::pin(ready(Ok(infos)))
1783
1784 }
1785
1786 fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
1788 Box::pin(async move { Ok(self.0.exec_set_attr_config(configs).await) })
1789 }
1790
1791 fn command_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<(SystemTime, Result<Value>)>>> {
1792 Box::pin(ready(Err(Error::not_implemented()
1793 .with_append_description("The recorded history of a command"))))
1794 }
1795
1796 fn attr_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<Result<AttrRead<Value>>>>> {
1797 Box::pin(ready(Err(Error::not_implemented()
1798 .with_append_description("The recorded history of an attribute"))))
1799 }
1800
1801 fn subscribe_attr_value<'a>(&'a self, _admin: Arc<dyn BackendDev>, name: &'a str, event: AttrValueEvent) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
1807 Box::pin(ready(self.0.resolve_attr(name).and_then(|target| {
1808 target.attr().check_push_event(&self.0, event)?;
1809 Ok(self.0.server().subscribe_attr_value(&self.0.name, target.name(), event))
1810 })))
1811 }
1812
1813 fn subscribe_attr_ready<'a>(&'a self, _admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<()>>>> {
1814 Box::pin(ready(self.0.resolve_attr(name).map(|target| {
1815 self.0.server().subscribe_attr_ready(&self.0.name, target.name())
1816 })))
1817 }
1818
1819 fn subscribe_attr_conf<'a>(&'a self, _admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> {
1820 Box::pin(ready(self.0.resolve_attr(name).map(|target| {
1821 self.0.server().subscribe_attr_conf(&self.0.name, target.name())
1822 })))
1823 }
1824
1825}
1826
1827type Resolved<T> = std::result::Result<T, Named<Error>>;
1830
1831fn resolved<T>(name: Scalar<str>, res: Result<T>) -> Resolved<T> {
1832 res.map_err(|err| Named::new(name, err))
1833}
1834
1835fn scalars(names: &Spectrum<str>) -> impl Iterator<Item = Scalar<str>> + '_ {
1837 (0..names.len()).map(|index| names.get_scalar([index]).expect("a name at every index"))
1838}
1839
1840
1841#[cfg(test)]
1842mod tests {
1843
1844 use std::sync::Mutex;
1845 use std::thread;
1846
1847 use futures::channel::oneshot;
1848 use futures::executor::block_on;
1849 use futures::{FutureExt, StreamExt};
1850
1851 use crate::dev::{ToNames, WriteAttrSpec};
1852 use crate::value::FromValue;
1853
1854 use super::*;
1855 use super::super::def::{AttrDef, ClassDef, CommandDef};
1856 use super::super::server::ServerBuilder;
1857 use crate::testing::TestContext;
1858
1859 fn serve<M: Model>(def: ClassDef<M>) -> (TestContext, Arc<DevImpl<M>>) {
1862 let ctx = TestContext::new().unwrap();
1863 ctx.database().add_server("Test/test", &[]).unwrap();
1864 ctx.database().add_dev("test/dev/1", &def.name, "Test/test").unwrap();
1865 let server = ctx.serve(ServerBuilder::new("Test", "test")).unwrap();
1866 let class = server.add_class(def).unwrap();
1867 let dev = class.add_dev("test/dev/1").unwrap();
1868 let dev = DevImpl::from_arc(Arc::clone(&dev.inner));
1869 drop(class);
1870 (ctx, dev)
1871 }
1872
1873 #[test]
1874 fn push_event() {
1875
1876 let (ctx, dev) = serve(model_def());
1877 let server = ctx.server("Test/test").unwrap();
1878 server.sync_blocking().unwrap();
1879 let mut state = server.subscribe_attr_value("test/dev/1", "State", AttrValueEvent::Change);
1880 let mut count = server.subscribe_attr_value("test/dev/1", "Count", AttrValueEvent::Change);
1881
1882 thread::scope(|scope| {
1883 scope.spawn(|| dev.0.push_change_event("state", DevState::Moving).unwrap());
1884 });
1885
1886 let read = block_on(state.next()).unwrap().unwrap();
1887 assert_eq!(read.value, Some(Value::from(DevState::Moving)));
1888 assert_eq!(read.w_value, None);
1889 assert_eq!(read.quality, AttrQuality::Valid);
1890
1891 dev.0.set_attr_w_value("Count", 5).unwrap();
1893 dev.0.push_change_event("Count", 3).unwrap();
1894 let read = block_on(count.next()).unwrap().unwrap();
1895 assert_eq!(read.value, Some(Value::from(3)));
1896 assert_eq!(read.w_value, Some(Value::from(5)));
1897
1898 assert!(dev.0.push_change_event("Missing", 1.0).is_err());
1899
1900 let err = dev.0.push_archive_event("Count", 3).unwrap_err();
1902 assert_eq!(err.kind(), ErrorKind::Reason(reason::API_AttributePollingNotStarted));
1903 dev.0.push_user_event("Count", 3).unwrap();
1904
1905 }
1906
1907 #[test]
1908 fn set_attr_config() {
1909
1910 let (ctx, dev) = serve(model_def());
1911 let server = ctx.server("Test/test").unwrap();
1912 server.sync_blocking().unwrap();
1913 let mut conf = server.subscribe_attr_conf("test/dev/1", "Count");
1914
1915 let mut config = dev.0.get_attr("Count").unwrap().config();
1916 config.label = Some("Counter".into());
1917 dev.0.set_attr_config("count", &config).unwrap();
1918
1919 assert_eq!(dev.0.get_attr("Count").unwrap().config().label.as_deref(), Some("Counter"));
1920 let info = block_on(conf.next()).unwrap().unwrap();
1921 assert_eq!(info.config.label.as_deref(), Some("Counter"));
1922
1923 dev.0.set_attr_config("Count", &config).unwrap();
1925 assert!(conf.next().now_or_never().is_none());
1926
1927 }
1928
1929 fn thread_name() -> String {
1930 thread::current().name().unwrap_or_default().to_string()
1931 }
1932
1933 struct Motor {
1934 count: i32,
1935 }
1936
1937 impl Motor {
1938
1939 async fn init(_dev: &Dev<Self>) -> Result<Self> {
1940 Ok(Self { count: 0 })
1941 }
1942
1943 async fn incr(&mut self, _dev: &Dev<Self>, arg: i32) -> Result<i32> {
1944 self.count += arg;
1945 Ok(self.count)
1946 }
1947
1948 async fn count(&self, _dev: &Dev<Self>) -> Result<i32> {
1949 Ok(self.count)
1950 }
1951
1952 async fn set_count(&mut self, _dev: &Dev<Self>, arg: i32) -> Result<()> {
1953 self.count = arg;
1954 Ok(())
1955 }
1956
1957 async fn state(&self, _dev: &Dev<Self>) -> Result<DevState> {
1958 Ok(if self.count == 0 { DevState::On } else { DevState::Moving })
1959 }
1960
1961 fn thread(&self, _dev: &Dev<Self>) -> Result<String> {
1962 Ok(thread_name())
1963 }
1964
1965 async fn thread_async(&self, _dev: &Dev<Self>) -> Result<String> {
1966 Ok(thread_name())
1967 }
1968
1969 }
1970
1971 fn model_def() -> ClassDef<Motor> {
1972 ClassDef::new_async("Motor", Motor::init)
1973 .state_async(Motor::state)
1974 .push_state_events(Some(false), None)
1975 .command(CommandDef::new("Incr").exec_exclusive_async(Motor::incr))
1976 .attr(AttrDef::new("Count").read_async(Motor::count).write_exclusive_async(Motor::set_count)
1977 .push_change_event(false))
1978 .attr(AttrDef::new("Thread").read(Motor::thread))
1979 .attr(AttrDef::new("ThreadAsync").read_async(Motor::thread_async))
1980 }
1981
1982 fn read_one<T: FromValue>(dev: &DevImpl<Motor>, name: &str) -> T {
1983 let mut reads = block_on(dev.read_attr(name.to_names(), DevSource::Dev)).unwrap();
1984 let read = reads.pop().unwrap().value;
1985 T::from_value(read.unwrap().value.unwrap()).unwrap()
1986 }
1987
1988 #[test]
1991 fn async_handlers() {
1992
1993 let (_ctx, dev) = serve(model_def());
1994 dev.0.server().sync_blocking().unwrap();
1995
1996 assert_eq!(block_on(dev.state()).unwrap(), DevState::On);
1997 let count = block_on(dev.command(Scalar::from("Incr"), Value::from(3), DevSource::Dev)).unwrap();
1998 assert_eq!(i32::from_value(count).unwrap(), 3);
1999 assert_eq!(block_on(dev.state()).unwrap(), DevState::Moving);
2000
2001 let writes = block_on(dev.write_attr(("Count", 7).into_values())).unwrap();
2002 assert!(writes.into_iter().all(|write| write.is_ok()));
2003 assert_eq!(read_one::<i32>(&dev, "Count"), 7);
2004
2005 block_on(dev.command(Scalar::from("Init"), Value::Void, DevSource::Dev)).unwrap();
2007 assert_eq!(read_one::<i32>(&dev, "Count"), 0);
2008
2009 }
2010
2011 #[test]
2014 fn unknown_attr_name() {
2015
2016 let (_ctx, dev) = serve(model_def());
2017 dev.0.server().sync_blocking().unwrap();
2018
2019 let names = Spectrum::from(&["count", "Nothing"][..]);
2020 let reads = block_on(dev.read_attr(Some(names.clone()), DevSource::Dev)).unwrap();
2021
2022 assert_eq!(reads[0].name.value(), "Count");
2023 assert!(reads[0].value.is_ok());
2024
2025 assert_eq!(reads[1].name.value().as_ptr(), names.get([1]).unwrap().as_ptr());
2026 assert_eq!(reads[1].value.as_ref().unwrap_err().kind(), ErrorKind::AttrNotFound);
2027
2028 }
2029
2030 #[test]
2034 fn request_threads() {
2035
2036 let (_ctx, dev) = serve(model_def());
2037 dev.0.server().sync_blocking().unwrap();
2038
2039 let on_pool = |name: &str| name.starts_with("pool-");
2040 assert!(on_pool(&read_one::<String>(&dev, "Thread")));
2041 assert!(!on_pool(&read_one::<String>(&dev, "ThreadAsync")));
2042
2043 let reads = block_on(dev.read_attr(["ThreadAsync", "Thread"].to_names(), DevSource::Dev)).unwrap();
2044 for read in reads {
2045 assert!(on_pool(&String::from_value(read.value.unwrap().value.unwrap()).unwrap()));
2046 }
2047
2048 }
2049
2050 #[test]
2053 fn lock_across_await() {
2054
2055 let (tx, rx) = oneshot::channel::<()>();
2056 let gate = Mutex::new(Some(rx));
2057 let def = model_def()
2058 .command(CommandDef::new("Wait").exec_exclusive_async_with(move |_model: &mut Motor, _dev, _input| {
2059 let rx = gate.lock().unwrap().take();
2060 Box::pin(async move {
2061 if let Some(rx) = rx {
2062 let _ = rx.await;
2063 }
2064 Ok(Value::Void)
2065 })
2066 }));
2067
2068 let (_ctx, dev) = serve(def);
2069 dev.0.server().sync_blocking().unwrap();
2070
2071 let mut wait = dev.command(Scalar::from("Wait"), Value::Void, DevSource::Dev);
2072 assert!((&mut wait).now_or_never().is_none());
2073 let mut read = dev.read_attr("Count".to_names(), DevSource::Dev);
2074 assert!((&mut read).now_or_never().is_none());
2075
2076 tx.send(()).unwrap();
2077 block_on(wait).unwrap();
2078 assert!(block_on(read).unwrap().pop().unwrap().value.is_ok());
2079
2080 }
2081
2082 #[test]
2085 fn async_init_failed() {
2086
2087 let def = ClassDef::<Motor>::new_async_with("Motor", |_dev| Box::pin(async {
2088 Err(Error::new(ErrorKind::Internal).with_description("No hardware"))
2089 }));
2090
2091 let (_ctx, dev) = serve(def);
2092 assert_eq!(block_on(dev.state()).unwrap(), DevState::Unknown);
2093 dev.0.server().sync_blocking().unwrap();
2094
2095 assert_eq!(block_on(dev.state()).unwrap(), DevState::Fault);
2096 assert!(block_on(dev.status()).unwrap().contains("No hardware"));
2097 assert!(block_on(dev.command(Scalar::from("Init"), Value::Void, DevSource::Dev)).is_err());
2098
2099 }
2100
2101}