1use std::collections::{HashMap, HashSet, VecDeque};
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::time::{Duration, Instant, SystemTime};
6use std::sync::{Arc, Mutex, Weak};
7use std::fmt::{self, Debug};
8use std::future::Future;
9use std::borrow::Cow;
10use std::any::Any;
11
12use futures::future::{join, ready, BoxFuture};
13use tracing::{debug, error, info, warn};
14use futures::stream::BoxStream;
15
16use crate::dev::{All, AttrConfig, AttrInfo, AttrMemorized, AttrQuality, AttrRead, AttrReady, AttrValueEvent, CommandInfo, DevInfo, Named, NamedResult, RecordedResult, ToNames, ToNamesNotAll,
17 DevSource, DevState};
18use crate::value::{IntoValue, Scalar, Spectrum, Value};
19use crate::error::{reason, Error, ErrorKind, Result};
20use crate::database::{GetPropSpec, SetPropSpec};
21use crate::local::{self, LocalDevHandle};
22use crate::name::{self, Name, ArcName};
23use crate::backend::BackendDev;
24use crate::pool;
25
26use super::def::{AttrDef, CommandDef, DeleteFn, InitFn, StateFn, StatusFn, WriteFn};
27use super::{reserved_name, Model, DEFAULT_HISTORY_DEPTH, DEFAULT_STATE, DEFAULT_STATUS, INIT_NAME,
28 MISSING_STATUS, RESERVED_ATTRS, RESERVED_COMMANDS, STATE_NAME, STATUS_NAME};
29use super::polling::{self, PollItem, PolledHistory, PolledKind, Polled, DEFAULT_VALIDITY_PERIODS};
30use super::class::{AnyClass, Class};
31use super::attr::{Attr, AttrAlarm, AttrRole};
32use super::command::Command;
33use super::server::Server;
34
35
36pub(super) trait AnyDev: Debug + Send + Sync {
39
40 fn name(&self) -> &Arc<str>;
42
43 fn class(&self) -> &dyn AnyClass;
45
46 fn init(&self) -> BoxFuture<'_, Result<()>>;
51
52 fn withdraw(&self) -> BoxFuture<'_, Result<()>>;
63
64 fn withdraw_now(&self);
69
70 fn serve(&self) -> Result<Arc<str>>;
74
75 fn into_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
77
78 fn poll(&self, kind: PolledKind, items: &[PollItem]) -> BoxFuture<'static, Result<Vec<Option<Instant>>>>;
90
91 fn polled_history(&self, kind: PolledKind, name: &str) -> Option<PolledHistory>;
94
95 fn check_pollable(&self, kind: PolledKind, name: &str) -> Result<Arc<str>>;
99
100}
101
102pub struct Dev<M: Model> {
114 this: Weak<Self>,
116 name: Arc<str>,
118 class: Arc<Class<M>>,
120 mutable: Mutex<DevMutable<M>>,
123 model: async_lock::RwLock<Option<Result<M>>>,
127 local_handle: Mutex<Option<LocalDevHandle>>,
130 withdrawn: AtomicBool,
132}
133
134impl<M: Model> Dev<M> {
135
136 pub(super) fn new(class: Arc<Class<M>>, name: Arc<str>) -> Arc<Self> {
137 let state_attr = Arc::new(Attr::new_with_role(Arc::clone(&class.def().state_attr), AttrRole::State));
138 let status_attr = Arc::new(Attr::new_with_role(Arc::clone(&class.def().status_attr), AttrRole::Status));
139 Arc::new_cyclic(|this| Self {
140 this: this.clone(),
141 name,
142 class,
143 mutable: Mutex::new(DevMutable {
144 black_box: VecDeque::new(),
145 commands: HashMap::new(),
146 attrs: HashMap::new(),
147 state_attr,
148 status_attr,
149 auto_alarm_states: Arc::from([DevState::On]),
150 }),
151 model: async_lock::RwLock::new(None),
152 local_handle: Mutex::new(None),
153 withdrawn: AtomicBool::new(false),
154 })
155 }
156
157 #[inline]
159 pub fn name(&self) -> &str {
160 &self.name
161 }
162
163 #[inline]
165 pub fn class(&self) -> &Class<M> {
166 &self.class
167 }
168
169 #[inline]
171 pub fn server(&self) -> &Server {
172 self.class.server()
173 }
174
175 #[inline]
181 pub fn is_withdrawn(&self) -> bool {
182 self.withdrawn.load(Ordering::Acquire)
183 }
184
185 fn check_withdrawn(&self) -> Result<()> {
187 if self.is_withdrawn() {
188 return Err(Error::new(ErrorKind::DevNotConnected)
189 .with_description(format!("Device {:?} is no longer served", self.name)));
190 }
191 Ok(())
192 }
193
194 pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
201 self.server().database().list_dev_prop_async(&self.name, filter).await
202 }
203
204 pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
206 self.server().database().list_dev_prop(&self.name, filter)
207 }
208
209 pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
212 where
213 Spec: GetPropSpec<Names>,
214 {
215 self.server().database().get_dev_prop_async::<Spec, Names>(&self.name, names)
216 }
217
218 pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
220 where
221 Spec: GetPropSpec<Names>,
222 {
223 self.server().database().get_dev_prop::<Spec, Names>(&self.name, names)
224 }
225
226 pub fn set_prop_async<'a, Spec>(&'a self, spec: Spec) -> impl Future<Output = Result<()>> + Send
228 where
229 Spec: SetPropSpec<'a>,
230 {
231 self.server().database().set_dev_prop_async(&self.name, spec)
232 }
233
234 pub fn set_prop<'a, Spec>(&'a self, spec: Spec) -> Result<()>
236 where
237 Spec: SetPropSpec<'a>,
238 {
239 self.server().database().set_dev_prop(&self.name, spec)
240 }
241
242 pub fn delete_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
244 self.server().database().delete_dev_prop_async(&self.name, names)
245 }
246
247 pub fn delete_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
249 self.server().database().delete_dev_prop(&self.name, names)
250 }
251
252 #[inline]
256 pub fn arc(&self) -> Arc<Self> {
257 self.this.upgrade().unwrap()
259 }
260
261 pub fn add_command(&self, def: CommandDef<M>) -> Result<()> {
270
271 if !name::check_tango_command_name(&def.name) {
276 return Err(Error::new(ErrorKind::CommandNameInvalid)
277 .with_description(format!("Command {:?} of device {:?} has an invalid name, it should \
278 be made of letters, digits, '_' and '-', at most 255 long", def.name, self.name)));
279 }
280
281 if let Some(reserved) = reserved_name(&def.name, RESERVED_COMMANDS) {
282 return Err(Error::new(ErrorKind::CommandNameInvalid)
283 .with_description(format!("Command {:?} cannot be defined on device {:?}, \
284 {reserved:?} is answered by the server itself", def.name, self.name)));
285 }
286
287 let mut mutable = self.mutable.lock().unwrap();
288 if mutable.commands.contains_key(Name::new(&def.name)) {
289 return Err(Error::new(ErrorKind::CommandAlreadyDefined)
290 .with_description(format!("Command {:?} is already defined on device {:?}", def.name, self.name)));
291 }
292
293 let name = ArcName::from(&*def.name);
294 mutable.commands.insert(name, Arc::new(Command::new(Arc::new(def))));
295 Ok(())
296
297 }
298
299 pub fn remove_command(&self, name: &str) -> bool {
301 self.mutable.lock().unwrap().commands.remove(Name::new(name)).is_some()
302 }
303
304 pub fn add_attr(&self, def: AttrDef<M>) -> Result<()> {
313
314 if !name::check_tango_attr_name(&def.name) {
315 return Err(Error::new(ErrorKind::AttrNameInvalid)
316 .with_description(format!("Attribute {:?} of device {:?} has an invalid name, it should \
317 be made of letters, digits, '_' and '-', at most 255 long",
318 def.name, self.name)));
319 }
320
321 if let Some(reserved) = reserved_name(&def.name, RESERVED_ATTRS) {
322 return Err(Error::new(ErrorKind::AttrNameInvalid)
323 .with_description(format!("Attribute {:?} cannot be defined on device {:?}, \
324 {reserved:?} is answered by the server itself",
325 def.name, self.name)));
326 }
327
328 if let Err(err) = def.check() {
329 return Err(Error::new(ErrorKind::Reason(reason::API_AttrWrongDefined))
330 .with_description(format!("Attribute {:?} cannot be defined on device {:?}, {err}",
331 def.name, self.name)));
332 }
333
334 let mut mutable = self.mutable.lock().unwrap();
335 if mutable.attrs.contains_key(Name::new(&def.name)) {
336 return Err(Error::new(ErrorKind::AttrAlreadyDefined)
337 .with_description(format!("Attribute {:?} is already defined on device {:?}",
338 def.name, self.name)));
339 }
340
341 let name = ArcName::from(&*def.name);
342 mutable.attrs.insert(name, Arc::new(Attr::new(Arc::new(def))));
343 Ok(())
344
345 }
346
347 pub async fn set_attr_w_value_async(&self, name: &str, value: impl IntoValue) -> Result<()> {
356 self.check_withdrawn()?;
357 let attr = self.find_attr(name)?;
358 attr.set_w_value(self, value.into_value()).await
359 }
360
361 pub fn set_attr_w_value(&self, name: &str, value: impl IntoValue) -> Result<()> {
364 self.server().database().block_on(self.set_attr_w_value_async(name, value))?
365 }
366
367 pub async fn set_attr_config_async(&self, name: &str, config: &AttrConfig) -> Result<()> {
372 self.check_withdrawn()?;
373 self.change_attr_config(name, config).await
374 }
375
376 pub fn set_attr_config(&self, name: &str, config: &AttrConfig) -> Result<()> {
379 self.server().database().block_on(self.set_attr_config_async(name, config))?
380 }
381
382 pub fn set_auto_alarm_states(&self, states: &[DevState]) {
392 self.mutable.lock().unwrap().auto_alarm_states = Arc::from(states);
393 }
394
395 pub fn remove_attr(&self, name: &str) -> bool {
397 self.mutable.lock().unwrap().attrs.remove(Name::new(name)).is_some()
398 }
399
400 pub fn get_command(&self, name: &str) -> Option<Arc<Command<M>>> {
409 if name.eq_ignore_ascii_case(STATE_NAME) {
410 Some(Arc::new(Command::new_with_attr(self.state_attr())))
411 } else if name.eq_ignore_ascii_case(STATUS_NAME) {
412 Some(Arc::new(Command::new_with_attr(self.status_attr())))
413 } else if name.eq_ignore_ascii_case(INIT_NAME) {
414 Some(Arc::new(Command::new_with_init()))
415 } else {
416 self.mutable.lock().unwrap().commands.get(Name::new(name)).map(Arc::clone)
417 }
418 }
419
420 pub fn get_attr(&self, name: &str) -> Option<Arc<Attr<M>>> {
423 let mutable = self.mutable.lock().unwrap();
424 if name.eq_ignore_ascii_case(STATE_NAME) {
425 Some(Arc::clone(&mutable.state_attr))
426 } else if name.eq_ignore_ascii_case(STATUS_NAME) {
427 Some(Arc::clone(&mutable.status_attr))
428 } else {
429 mutable.attrs.get(Name::new(name)).map(Arc::clone)
430 }
431 }
432
433 pub fn commands(&self) -> Vec<Arc<Command<M>>> {
436 let mutable = self.mutable.lock().unwrap();
437 [
438 Command::new_with_attr(Arc::clone(&mutable.state_attr)),
439 Command::new_with_attr(Arc::clone(&mutable.status_attr)),
440 Command::new_with_init(),
441 ]
442 .into_iter()
443 .map(Arc::new)
444 .chain(mutable.commands.values().map(Arc::clone))
445 .collect()
446 }
447
448 pub fn attrs(&self) -> Vec<Arc<Attr<M>>> {
454 let mutable = self.mutable.lock().unwrap();
455 [&mutable.state_attr, &mutable.status_attr]
456 .into_iter()
457 .chain(mutable.attrs.values())
458 .map(Arc::clone)
459 .collect()
460 }
461
462 pub fn push_attr_event(&self, name: &str, event: AttrValueEvent, mut read: Result<AttrRead<Value>>) -> Result<()> {
478 self.check_withdrawn()?;
479 let attr = self.find_attr(name)?;
480 attr.check_push_event(self, event)?;
481 if let Ok(read) = &mut read {
482 attr.check_pushed(read)?;
483 attr.merge_w_value(read);
484 }
485 self.server().push_attr_value(&self.name, attr.name(), event, read);
486 Ok(())
487 }
488
489 pub fn push_change_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
492 self.push_attr_event(name, AttrValueEvent::Change, Ok(read_now(value)))
493 }
494
495 pub fn push_archive_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
497 self.push_attr_event(name, AttrValueEvent::Archive, Ok(read_now(value)))
498 }
499
500 pub fn push_user_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
502 self.push_attr_event(name, AttrValueEvent::User, Ok(read_now(value)))
503 }
504
505 pub fn push_alarm_event(&self, name: &str, value: impl IntoValue) -> Result<()> {
507 self.push_attr_event(name, AttrValueEvent::Alarm, Ok(read_now(value)))
508 }
509
510 pub fn push_data_ready_event(&self, name: &str, counter: i32) -> Result<()> {
515 self.check_withdrawn()?;
516 let attr = self.find_attr(name)?;
517 attr.check_data_ready(self)?;
518 let ready = AttrReady { data_kind: attr.data_kind()?, counter };
519 self.server().push_attr_ready(&self.name, attr.name(), Ok(ready));
520 Ok(())
521 }
522
523 pub(super) async fn init(&self) {
538
539 let def = self.class.def();
540 let mut model = self.model.write().await;
541
542 if self.is_withdrawn() {
546 return;
547 }
548
549 self.delete_model(&mut model).await;
553
554 self.reset_interface();
562 self.load_all_attr_config().await;
563
564 let mut res = match &def.init {
565 InitFn::Sync(init) => init(self),
566 InitFn::Async(init) => init(self).await,
567 };
568
569 if let Ok(new_model) = &mut res {
573 self.load_all_attr_config().await;
574 for attr in self.attrs() {
575 attr.restore(self, new_model).await;
576 }
577 }
578
579 *model = Some(match res {
581 Ok(new_model) => {
582 info!(dev = %self.name, class = %def.name, "Initialized the device");
583 Ok(new_model)
584 }
585 Err(err) => {
586 error!(dev = %self.name, class = %def.name, %err, "Failed to initialize the device");
587 Err(err)
588 }
589 });
590
591 self.setup_polling().await;
594
595 }
596
597 async fn setup_polling(&self) {
605
606 const PROPS: [&str; 6] = [
607 "polled_attr",
608 "polled_cmd",
609 "poll_ring_depth",
610 "attr_poll_ring_depth",
611 "cmd_poll_ring_depth",
612 "poll_old_factor",
613 ];
614
615 let Some(poller) = self.server().poller() else {
616 return;
617 };
618
619 let props = match self.server().database().get_dev_prop_async::<Spectrum<str>, _>(&self.name, PROPS).await {
620 Ok(props) => props,
621 Err(err) => {
622 warn!(dev = %self.name, %err, "Failed to read the polling properties of the device, polling it as defined");
623 PROPS.map(|_| Ok(Spectrum::<str>::from(&[] as &[&str])))
624 }
625 };
626
627 let [
628 polled_attr,
629 polled_cmd,
630 ring_depth,
631 attr_depths,
632 cmd_depths,
633 validity_periods,
634 ] = props.into_iter()
635 .zip(PROPS)
636 .map(|(prop, name)| prop.unwrap_or_else(|err| {
637 warn!(dev = %self.name, prop = %name, %err, "Failed to read a polling property of the device, ignoring it");
638 Spectrum::<str>::from(&[] as &[&str])
639 }))
640 .collect::<Vec<_>>()
641 .try_into()
642 .expect("one value per property");
643
644 let ring_depth = first_number(&self.name, PROPS[2], &ring_depth).unwrap_or(DEFAULT_HISTORY_DEPTH);
647 let validity_periods = first_number(&self.name, PROPS[5], &validity_periods)
648 .filter(|&factor| factor > 0)
649 .unwrap_or(DEFAULT_VALIDITY_PERIODS);
650
651 let attr_depths = polling::parse_depths(&self.name, PROPS[3], attr_depths.iter()).into_iter()
652 .map(|(name, depth)| (ArcName::from(name), depth))
653 .collect::<HashMap<_, _>>();
654 let cmd_depths = polling::parse_depths(&self.name, PROPS[4], cmd_depths.iter()).into_iter()
655 .map(|(name, depth)| (ArcName::from(name), depth))
656 .collect::<HashMap<_, _>>();
657
658 let (attrs, commands) = (self.attrs(), self.commands());
659
660 for attr in &attrs {
661 attr.set_history_depth(attr_depths.get(Name::new(attr.name())).copied().unwrap_or(ring_depth));
662 }
663
664 for command in &commands {
667 match cmd_depths.get(Name::new(command.name())) {
668 Some(&depth) => command.set_history_depth(depth),
669 None if !command.involves_state() => command.set_history_depth(ring_depth),
670 None => {}
671 }
672 }
673
674 let mut polled = Vec::new();
676 for (kind, prop) in [(PolledKind::Attr, &polled_attr), (PolledKind::Command, &polled_cmd)] {
677
678 let prop_name = match kind {
679 PolledKind::Attr => PROPS[0],
680 PolledKind::Command => PROPS[1],
681 };
682
683 let mut listed = HashSet::new();
684 for (name, period) in polling::parse_periods(&self.name, prop_name, prop.iter()) {
685 match self.check_pollable(kind, name) {
686 Ok(name) => {
687 listed.insert(ArcName::from(Arc::clone(&name)));
688 polled.push(Polled { kind, name, period });
689 }
690 Err(err) => warn!(dev = %self.name, prop = %prop_name, %name, %err, "Ignored a polled object"),
691 }
692 }
693
694 let defined = match kind {
695 PolledKind::Attr => attrs.iter()
696 .filter_map(|attr| attr.poll_period().map(|period| (attr.name(), period)))
697 .collect::<Vec<_>>(),
698 PolledKind::Command => commands.iter()
699 .filter_map(|command| command.poll_period().map(|period| (command.name(), period)))
700 .collect::<Vec<_>>(),
701 };
702
703 for (name, period) in defined {
704 if !listed.contains(Name::new(name)) {
705 polled.push(Polled { kind, name: Arc::from(name), period: period.max(polling::MIN_PERIOD) });
706 }
707 }
708
709 }
710
711 debug!(dev = %self.name, polled = polled.len(), "Polling the device");
712 poller.set_dev(&(self.arc() as Arc<dyn AnyDev>), validity_periods, polled);
713
714 }
715
716 fn poll_attrs(self: Arc<Self>, items: &[PollItem]) -> BoxFuture<'static, Result<Vec<Option<Instant>>>> {
720
721 let mut last_periodics = items.iter()
722 .map(|item| item.last_periodic)
723 .collect::<Vec<_>>();
724
725 if self.is_withdrawn() {
727 return Box::pin(ready(Ok(last_periodics)));
728 }
729
730 let mut polled = Vec::new();
732 for (index, item) in items.iter().enumerate() {
733 match self.find_attr(&item.name) {
734 Ok(attr) => polled.push((index, attr, item.valid_for)),
735 Err(err) => debug!(dev = %self.name, attr = %item.name, %err, "Failed to poll the attribute"),
736 }
737 }
738
739 if polled.is_empty() {
740 return Box::pin(ready(Ok(last_periodics)));
741 }
742
743 let targets = polled.iter()
744 .map(|(_, attr, _)| Ok(Arc::clone(attr)))
745 .collect::<Vec<_>>();
746
747 let ctx = ReadCtx::new(&self, polled.iter().any(|(_, attr, _)| attr.role() != AttrRole::Interface));
748 let blocking = polled.iter().any(|(_, attr, _)| attr.read_may_block(&self, &ctx));
749
750 pool::run(blocking, async move {
751
752 let mut reads = self.exec_read_attr(targets, ctx, DevSource::Dev).await
755 .map(Vec::into_iter);
756
757 for (index, attr, valid_for) in polled {
758 let read = match &mut reads {
759 Ok(reads) => reads.next().expect("every attribute is read").value,
760 Err(err) => Err(err.clone()),
761 };
762 last_periodics[index] = self.record_poll(&attr, read, valid_for, last_periodics[index]);
763 }
764
765 Ok(last_periodics)
766
767 })
768
769 }
770
771 fn record_poll(&self, attr: &Attr<M>, read: Result<AttrRead<Value>>, valid_for: Duration, last_periodic: Option<Instant>) -> Option<Instant> {
777
778 let read = read.and_then(|read| attr.check_pushed(&read).map(|()| read));
779
780 if let Err(err) = attr.push_history(read.clone(), Some(valid_for)) {
781 debug!(dev = %self.name, attr = %attr.name(), %err, "Failed to record the attribute polled");
782 }
783
784 let now = Instant::now();
785 let period = attr.config().periodic_event.period.unwrap_or(polling::DEFAULT_PERIODIC_EVENT_PERIOD);
786 if last_periodic.is_some_and(|last| now.duration_since(last) < period.mul_f64(polling::PERIODIC_EVENT_TOLERANCE)) {
787 return last_periodic;
788 }
789
790 let read = read.map(|mut read| {
791 attr.merge_w_value(&mut read);
792 read
793 });
794
795 self.server().push_attr_value(&self.name, attr.name(), AttrValueEvent::Periodic, read);
796 Some(now)
797
798 }
799
800 fn poll_commands(self: Arc<Self>, items: &[PollItem]) -> BoxFuture<'static, Result<Vec<Option<Instant>>>> {
804
805 let none = vec![None; items.len()];
806
807 if self.is_withdrawn() {
809 return Box::pin(ready(Ok(none)));
810 }
811
812 let mut polled = Vec::new();
814 for item in items {
815 match self.find_command(&item.name) {
816 Ok(command) => polled.push((command, item.valid_for)),
817 Err(err) => debug!(dev = %self.name, command = %item.name, %err, "Failed to poll the command"),
818 }
819 }
820
821 if polled.is_empty() {
822 return Box::pin(ready(Ok(none)));
823 }
824
825 let targets = polled.iter()
826 .map(|(command, _)| (Ok(Arc::clone(command)), Value::Void))
827 .collect::<Vec<_>>();
828
829 let ctx = ReadCtx::new(&self, polled.iter().any(|(command, _)| command.involves_state()));
830 let blocking = polled.iter().any(|(command, _)| command.may_block(&self, &ctx));
831
832 pool::run(blocking, async move {
833
834 let mut results = self.exec_many_command(targets, ctx, DevSource::Dev).await
836 .map(Vec::into_iter);
837
838 for (command, valid_for) in polled {
839 let res = match &mut results {
840 Ok(results) => results.next().expect("every command is executed").value,
841 Err(err) => Err(err.clone()),
842 };
843 if let Err(err) = command.push_history(res, Some(valid_for)) {
844 debug!(dev = %self.name, command = %command.name(), %err, "Failed to record the command polled");
845 }
846 }
847
848 Ok(none)
849
850 })
851
852 }
853
854 async fn delete_model(&self, model: &mut Option<Result<M>>) {
863
864 let Some(Ok(prev)) = model.take() else {
867 return;
868 };
869
870 let def = self.class.def();
871 debug!(dev = %self.name, class = %def.name, "Deleting the model of the device");
872
873 match &def.delete {
874 Some(DeleteFn::Sync(delete)) => delete(prev, self),
875 Some(DeleteFn::Async(delete)) => delete(prev, self).await,
876 None => {}
877 }
878
879 }
880
881 fn reset_interface(&self) {
885
886 let def = self.class.def();
887 let mut mutable = self.mutable.lock().unwrap();
888
889 mutable.commands.clear();
890 mutable.attrs.clear();
891 mutable.state_attr = Arc::new(Attr::new_with_role(Arc::clone(&def.state_attr), AttrRole::State));
892 mutable.status_attr = Arc::new(Attr::new_with_role(Arc::clone(&def.status_attr), AttrRole::Status));
893
894 for def in &def.commands {
895 let name = ArcName::from(&*def.name);
896 mutable.commands.insert(name, Arc::new(Command::new(Arc::clone(def))));
897 }
898
899 for def in &def.attrs {
900 let name = ArcName::from(&*def.name);
901 mutable.attrs.insert(name, Arc::new(Attr::new(Arc::clone(def))));
902 }
903
904 }
905
906 async fn load_all_attr_config(&self) {
912
913 let attrs = {
916 let mutable = self.mutable.lock().unwrap();
917 [&mutable.state_attr, &mutable.status_attr]
918 .into_iter()
919 .chain(mutable.attrs.values())
920 .filter(|attr| !attr.is_loaded())
921 .map(Arc::clone)
922 .collect::<Vec<_>>()
923 };
924
925 self.load_given_attr_config(&attrs).await;
926
927 }
928
929 async fn load_given_attr_config(&self, attrs: &[Arc<Attr<M>>]) {
932
933 if attrs.is_empty() {
934 return;
935 }
936
937 let names = attrs
939 .iter()
940 .map(|attr| attr.name())
941 .collect::<Vec<_>>();
942
943 let (
945 class_props,
946 dev_props,
947 ) = join(
948 self.class.attr_props(&names),
949 self.server()
950 .database()
951 .get_dev_attr_prop_async::<Spectrum<str>, _, _>(&self.name, names.as_slice(), All),
952 ).await;
953
954 let dev_props = dev_props.and_then(|props| props
955 .into_iter()
956 .collect::<Result<Vec<_>>>());
957
958 for (index, attr) in attrs.iter().enumerate() {
959 attr.load_config(self,
960 class_props.as_ref().map(|props| &*props[index]),
961 dev_props.as_ref().map(|props| &*props[index]));
962 }
963
964 }
965
966 fn no_model_answer(&self, role: AttrRole, failure: Option<&Error>) -> Result<Value> {
973 match (role, failure) {
974 (AttrRole::State, Some(_)) => Ok(DevState::Fault.into()),
975 (AttrRole::State, None) => Ok(DevState::Unknown.into()),
976 (AttrRole::Status, Some(err)) => Ok(make_failed_status(err).into()),
977 (AttrRole::Status, None) => Ok(Scalar::from(MISSING_STATUS).into()),
978 (AttrRole::Interface, _) => Err(init_err(failure, &self.name)),
979 }
980 }
981
982 pub(super) async fn model_state(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<DevState> {
990 let state = self.model_state_inner(model, ctx).await?;
991 if self.attrs_in_alarm(ctx, state, model, false).await.is_empty() {
992 Ok(state)
993 } else {
994 Ok(DevState::Alarm)
995 }
996 }
997
998 pub(super) async fn model_status(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<Scalar<str>> {
1001
1002 let status = match &self.class.def().status {
1003 Some(StatusFn::SharedSync(func)) => func(model.shared(), self),
1004 Some(StatusFn::ExclusiveSync(func)) => func(model.exclusive()?, self),
1005 Some(StatusFn::SharedAsync(func)) => func(model.shared(), self).await,
1006 Some(StatusFn::ExclusiveAsync(func)) => func(model.exclusive()?, self).await,
1007 None => Ok(Scalar::from(DEFAULT_STATUS)),
1008 }?;
1009
1010 if !ctx.has_alarm() {
1011 return Ok(status);
1012 }
1013
1014 let state = self.model_state_inner(model, ctx).await?;
1015 let alarms = self.attrs_in_alarm(ctx, state, model, true).await;
1016 if alarms.is_empty() {
1017 return Ok(status);
1018 }
1019
1020 let mut status = status.value().to_owned();
1021 for (attr, alarm) in alarms {
1022 status.push('\n');
1023 status.push_str(&alarm.describe(attr.name()));
1024 }
1025
1026 Ok(Scalar::from(status.as_str()))
1027
1028 }
1029
1030 async fn model_state_inner(&self, model: &mut ModelRef<'_, M>, ctx: &mut ReadCtx<M>) -> Result<DevState> {
1033
1034 if let Some(state) = ctx.state {
1035 return Ok(state);
1036 }
1037
1038 let state = match &self.class.def().state {
1039 Some(StateFn::SharedSync(func)) => func(model.shared(), self),
1040 Some(StateFn::ExclusiveSync(func)) => func(model.exclusive()?, self),
1041 Some(StateFn::SharedAsync(func)) => func(model.shared(), self).await,
1042 Some(StateFn::ExclusiveAsync(func)) => func(model.exclusive()?, self).await,
1043 None => Ok(DEFAULT_STATE),
1044 }?;
1045
1046 ctx.state = Some(state);
1047 Ok(state)
1048
1049 }
1050
1051 async fn attrs_in_alarm(&self, ctx: &mut ReadCtx<M>, state: DevState, model: &mut ModelRef<'_, M>, every: bool) -> Vec<(Arc<Attr<M>>, AttrAlarm)> {
1057
1058 let mut found = Vec::new();
1059 if !ctx.may_alarm(state) {
1060 return found;
1061 }
1062
1063 let mut alarm_ctx = ctx.alarm.take();
1066 for alarm_attr in alarm_ctx.iter_mut().flat_map(|alarm| &mut alarm.attrs) {
1067
1068 let alarm = match alarm_attr.read {
1069 Some(alarm) => alarm,
1070 None => {
1071
1072 let alarm = alarm_attr.attr.read_with_alarm(self, model, ctx, DevSource::CacheDev).await
1077 .ok()
1078 .and_then(|(_, alarm)| alarm);
1079
1080 alarm_attr.read = Some(alarm);
1081 alarm
1082
1083 }
1084 };
1085
1086 if let Some(alarm) = alarm {
1087 found.push((Arc::clone(&alarm_attr.attr), alarm));
1088 if !every {
1089 break;
1090 }
1091 }
1092
1093 }
1094
1095 ctx.alarm = alarm_ctx;
1096 found
1097
1098 }
1099
1100 pub(super) fn state_access(&self, ctx: &ReadCtx<M>) -> Access {
1103 let access = match &self.class.def().state {
1104 Some(StateFn::ExclusiveSync(_) | StateFn::ExclusiveAsync(_)) => Access::Exclusive,
1105 _ => Access::Shared,
1106 };
1107 access.merge(ctx.access(self))
1108 }
1109
1110 pub(super) fn status_access(&self, ctx: &ReadCtx<M>) -> Access {
1113 let access = match &self.class.def().status {
1114 Some(StatusFn::ExclusiveSync(_) | StatusFn::ExclusiveAsync(_)) => Access::Exclusive,
1115 _ => Access::Shared,
1116 };
1117 if ctx.has_alarm() { access.merge(self.state_access(ctx)) } else { access }
1118 }
1119
1120 pub(super) fn init_may_block(&self) -> bool {
1126 let def = self.class.def();
1127 matches!(def.init, InitFn::Sync(_)) || self.delete_may_block() || def.attrs.iter().any(|attr| {
1128 attr.memorized == Some(AttrMemorized::RestoreAndWrite)
1129 && matches!(attr.write, Some(WriteFn::SharedSync(_) | WriteFn::ExclusiveSync(_)))
1130 })
1131 }
1132
1133 fn delete_may_block(&self) -> bool {
1136 matches!(self.class.def().delete, Some(DeleteFn::Sync(_)))
1137 }
1138
1139 pub(super) fn state_may_block(&self, ctx: &ReadCtx<M>) -> bool {
1142 matches!(&self.class.def().state, Some(StateFn::SharedSync(_) | StateFn::ExclusiveSync(_)))
1143 || ctx.may_block(self)
1144 }
1145
1146 pub(super) fn status_may_block(&self, ctx: &ReadCtx<M>) -> bool {
1149 matches!(&self.class.def().status, Some(StatusFn::SharedSync(_) | StatusFn::ExclusiveSync(_)))
1150 || (ctx.has_alarm() && self.state_may_block(ctx))
1151 }
1152
1153 async fn lock(&self, access: Access) -> ModelGuard<'_, M> {
1155 match access {
1156 Access::Shared => ModelGuard::Shared(self.model.read().await),
1157 Access::Exclusive => ModelGuard::Exclusive(self.model.write().await),
1158 }
1159 }
1160
1161 pub(super) async fn lock_model(&self, access: Access) -> Result<ModelGuard<'_, M>> {
1168 let guard = self.lock(access).await;
1169 match guard.get() {
1170 Some(Ok(_)) => Ok(guard),
1171 Some(Err(err)) => Err(init_err(Some(err), &self.name)),
1172 None => Err(init_err(None, &self.name)),
1173 }
1174 }
1175
1176 pub(super) async fn check_model(&self) -> Result<()> {
1179 match &*self.model.read().await {
1180 Some(Ok(_)) => Ok(()),
1181 Some(Err(err)) => Err(init_err(Some(err), &self.name)),
1182 None => Err(init_err(None, &self.name)),
1183 }
1184 }
1185
1186 fn find_command(&self, name: &str) -> Result<Arc<Command<M>>> {
1189 self.get_command(name)
1190 .ok_or_else(|| Error::new(ErrorKind::CommandNotFound)
1191 .with_description(format!("Command {name:?} is not defined on device {:?}", self.name)))
1192 }
1193
1194 fn find_attr(&self, name: &str) -> Result<Arc<Attr<M>>> {
1196 self.get_attr(name)
1197 .ok_or_else(|| Error::new(ErrorKind::AttrNotFound)
1198 .with_description(format!("Attribute {name:?} is not defined on device {:?}", self.name)))
1199 }
1200
1201 fn state_attr(&self) -> Arc<Attr<M>> {
1204 Arc::clone(&self.mutable.lock().unwrap().state_attr)
1205 }
1206
1207 fn status_attr(&self) -> Arc<Attr<M>> {
1209 Arc::clone(&self.mutable.lock().unwrap().status_attr)
1210 }
1211
1212 fn attr_targets(&self, names: Option<Spectrum<str>>) -> Vec<Resolved<Arc<Attr<M>>>> {
1216 match names {
1217 Some(names) => scalars(&names)
1218 .map(|name| {
1219 let target = self.find_attr(name.value());
1220 resolved(name, target)
1221 })
1222 .collect(),
1223 None => self.attrs().into_iter().map(Ok).collect(),
1224 }
1225 }
1226
1227 fn command_targets(&self, names: Option<Spectrum<str>>) -> Vec<Resolved<Arc<Command<M>>>> {
1229 match names {
1230 Some(names) => scalars(&names)
1231 .map(|name| {
1232 let target = self.find_command(name.value());
1233 resolved(name, target)
1234 })
1235 .collect(),
1236 None => self.commands().into_iter().map(Ok).collect(),
1237 }
1238 }
1239
1240 async fn exec_command(&self, target: Result<Arc<Command<M>>>, input: Value, mut ctx: ReadCtx<M>, source: DevSource) -> Result<Value> {
1245
1246 let command = match target {
1250 Ok(command) => command,
1251 Err(err) => return Err(self.check_model().await.err().unwrap_or(err)),
1252 };
1253
1254 if command.is_init() {
1258
1259 if source == DevSource::Cache {
1260 return Err(command.not_polled_err(self));
1261 }
1262
1263 self.init().await;
1264 self.check_model().await?;
1265 return Ok(Value::Void);
1266
1267 }
1268
1269 let mut guard = self.lock(command.exec_access(self, &ctx)).await;
1270 if let Some(failure) = guard.failure() {
1271 return self.no_model_answer(command.attr_role(), failure.as_ref());
1272 }
1273
1274 command.exec(self, &mut guard.model(), input, &mut ctx, source).await
1275
1276 }
1277
1278 async fn exec_many_command(&self, targets: Vec<(Resolved<Arc<Command<M>>>, Value)>, mut ctx: ReadCtx<M>, source: DevSource) -> Result<Vec<NamedResult<Value>>> {
1286
1287 let access = targets.iter()
1288 .filter_map(|(target, _)| target.as_ref().ok())
1289 .fold(Access::Shared, |access, command| access.merge(command.exec_access(self, &ctx)));
1290
1291 let mut guard = self.lock(access).await;
1292
1293 if let Some(failure) = guard.failure() {
1296 return Ok(targets.into_iter()
1297 .map(|(target, _)| match target {
1298 Ok(command) => Named::new(Scalar::from(command.name()), self.no_model_answer(command.attr_role(), failure.as_ref())),
1299 Err(unresolved) => unresolved.map(Err),
1300 })
1301 .collect());
1302 }
1303
1304 let mut model = guard.model();
1305 let mut results = Vec::with_capacity(targets.len());
1306
1307 for (target, input) in targets {
1308 results.push(match target {
1309 Ok(command) => {
1310 let res = command.exec(self, &mut model, input, &mut ctx, source).await;
1311 Named::new(Scalar::from(command.name()), res)
1312 }
1313 Err(unresolved) => unresolved.map(Err),
1314 });
1315 }
1316
1317 Ok(results)
1318
1319 }
1320
1321 async fn exec_read_attr(&self, targets: Vec<Resolved<Arc<Attr<M>>>>, mut ctx: ReadCtx<M>, source: DevSource) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
1323
1324 let access = targets.iter()
1328 .filter_map(|target| target.as_ref().ok())
1329 .fold(Access::Shared, |access, attr| access.merge(attr.read_access(self, &ctx)));
1330
1331 let mut guard = self.lock(access).await;
1332
1333 if let Some(failure) = guard.failure() {
1337 return Ok(targets.into_iter()
1338 .map(|target| match target {
1339 Ok(attr) => Named::new(Scalar::from(attr.name()), self.no_model_answer(attr.role(), failure.as_ref()).map(read_now)),
1340 Err(unresolved) => unresolved.map(Err),
1341 })
1342 .collect());
1343 }
1344
1345 let mut model = guard.model();
1350 let mut reads = vec![const { None }; targets.len()];
1351
1352 for interface in [true, false] {
1353 for (target, slot) in targets.iter().zip(&mut reads) {
1354 if let Ok(attr) = target
1355 && (attr.role() == AttrRole::Interface) == interface {
1356 let read = attr.read_with_alarm(self, &mut model, &mut ctx, source).await;
1357 ctx.record(attr, read.as_ref().ok().and_then(|(_, alarm)| *alarm));
1358 *slot = Some(Named::new(Scalar::from(attr.name()), read.map(|(read, _)| read)));
1359 }
1360 }
1361 }
1362
1363 for (target, slot) in targets.into_iter().zip(&mut reads) {
1364 if let Err(unresolved) = target {
1365 *slot = Some(unresolved.map(Err));
1366 }
1367 }
1368
1369 Ok(reads.into_iter().map(|read| read.expect("every target is read")).collect())
1370
1371 }
1372
1373 async fn exec_write_attr(&self, targets: Vec<(Result<Arc<Attr<M>>>, Value)>) -> Result<Vec<Result<()>>> {
1375
1376 let access = targets.iter()
1379 .filter_map(|(target, _)| target.as_ref().ok())
1380 .fold(Access::Shared, |access, attr| access.merge(attr.write_access()));
1381
1382 let mut guard = self.lock_model(access).await?;
1383 let mut model = guard.model();
1384
1385 let mut writes = Vec::with_capacity(targets.len());
1386 for (target, value) in targets {
1387 writes.push(match target {
1388 Ok(attr) => attr.write(self, &mut model, value).await,
1389 Err(err) => Err(err),
1390 });
1391 }
1392
1393 Ok(writes)
1394
1395 }
1396
1397 async fn exec_set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> Vec<Result<()>> {
1403 let mut results = Vec::with_capacity(configs.len());
1404 for Named { name, value } in configs {
1405 results.push(self.change_attr_config(name.value(), &value).await);
1406 }
1407 results
1408 }
1409
1410 async fn change_attr_config(&self, name: &str, config: &AttrConfig) -> Result<()> {
1415
1416 let attr = self.find_attr(name)?;
1417 let class_props = self.class.attr_props(&[attr.name()]).await?;
1418 let changes = attr.change_config(&class_props[0], config)
1419 .map_err(|err| Error::new(ErrorKind::Reason(reason::API_IncoherentValues))
1420 .with_description(format!("Configuration of attribute {:?} of device {:?} \
1421 cannot be changed, {err}", attr.name(), self.name)))?;
1422
1423 if changes.is_empty() {
1424 return Ok(());
1425 }
1426
1427 self.server().push_attr_config(&self.name, attr.name(), attr.info());
1428
1429 let database = self.server().database();
1430
1431 if !changes.put.is_empty() {
1432 let put = changes.put.into_iter()
1433 .map(|(prop, values)| (prop.name(), values))
1434 .collect::<Vec<_>>();
1435 database.set_dev_attr_prop_async(&self.name, (attr.name(), put)).await?;
1436 }
1437
1438 if !changes.del.is_empty() {
1439 let del = changes.del.iter().map(|prop| prop.name()).collect::<Vec<_>>();
1440 database.delete_dev_attr_prop_async(&self.name, attr.name(), del.as_slice()).await?;
1441 }
1442
1443 Ok(())
1444
1445 }
1446
1447 async fn exec_write_read_attr(&self, targets: Vec<(Resolved<Arc<Attr<M>>>, Value)>, mut ctx: ReadCtx<M>) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
1450
1451 let access = targets.iter()
1455 .filter_map(|(target, _)| target.as_ref().ok())
1456 .fold(Access::Shared, |access, attr| access.merge(attr.write_access()).merge(attr.read_access(self, &ctx)));
1457
1458 let mut guard = self.lock_model(access).await?;
1459 let mut model = guard.model();
1460
1461 let mut reads = Vec::with_capacity(targets.len());
1462 for (target, value) in targets {
1463 reads.push(match target {
1464 Ok(attr) => {
1467 let read = match attr.write(self, &mut model, value).await {
1468 Ok(()) => attr.read(self, &mut model, &mut ctx, DevSource::Dev).await,
1469 Err(err) => Err(err),
1470 };
1471 Named::new(Scalar::from(attr.name()), read)
1472 }
1473 Err(unresolved) => unresolved.map(Err),
1474 });
1475 }
1476
1477 Ok(reads)
1478
1479 }
1480}
1481
1482impl<M: Model> AnyDev for Dev<M> {
1483
1484 #[inline]
1485 fn name(&self) -> &Arc<str> {
1486 &self.name
1487 }
1488
1489 #[inline]
1490 fn class(&self) -> &dyn AnyClass {
1491 &*self.class
1492 }
1493
1494 fn init(&self) -> BoxFuture<'_, Result<()>> {
1495 let dev = self.arc();
1496 pool::run(self.init_may_block(), async move {
1497 dev.init().await;
1498 Ok(())
1499 })
1500 }
1501
1502 fn withdraw(&self) -> BoxFuture<'_, Result<()>> {
1503 let dev = self.arc();
1504 Box::pin(async move {
1505
1506 dev.withdraw_now();
1508 if let Some(backend) = dev.server().backend() {
1509 backend.remove(&dev.name);
1510 }
1511
1512 pool::run(dev.delete_may_block(), async move {
1513 dev.delete_model(&mut *dev.model.write().await).await;
1514 Ok(())
1515 }).await
1516
1517 })
1518 }
1519
1520 fn withdraw_now(&self) {
1521 self.withdrawn.store(true, Ordering::Release);
1522 self.local_handle.lock().unwrap().take();
1523 if let Some(poller) = self.server().poller() {
1524 poller.remove_dev(self);
1525 }
1526 }
1527
1528 fn serve(&self) -> Result<Arc<str>> {
1529 let backend = self.server().backend_checked()?;
1530 let dev = DevImpl::from_arc(self.arc()) as Arc<dyn BackendDev>;
1531 let itr = backend.add(&self.name, Arc::clone(&dev))?;
1532 if self.server().local {
1533 let local_handle = local::register(&dev, Arc::clone(&itr), backend.host(), backend.port(), Arc::clone(&self.name));
1534 *self.local_handle.lock().unwrap() = Some(local_handle);
1535 }
1536 Ok(itr)
1537 }
1538
1539 #[inline]
1540 fn into_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
1541 self
1542 }
1543
1544 fn poll(&self, kind: PolledKind, items: &[PollItem]) -> BoxFuture<'static, Result<Vec<Option<Instant>>>> {
1545 match kind {
1546 PolledKind::Attr => self.arc().poll_attrs(items),
1547 PolledKind::Command => self.arc().poll_commands(items),
1548 }
1549 }
1550
1551 fn polled_history(&self, kind: PolledKind, name: &str) -> Option<PolledHistory> {
1552
1553 const RECORDS: usize = 5;
1555
1556 let (depth, entries) = match kind {
1557 PolledKind::Attr => {
1558 let attr = self.get_attr(name)?;
1559 let entries = attr.history(RECORDS).into_iter()
1560 .map(|entry| (entry.time, entry.value.err()))
1561 .collect::<Vec<_>>();
1562 (attr.history_depth(), entries)
1563 }
1564 PolledKind::Command => {
1565 let command = self.get_command(name)?;
1566 let entries = command.history(RECORDS).into_iter()
1567 .map(|entry| (entry.time, entry.value.err()))
1568 .collect::<Vec<_>>();
1569 (command.history_depth(), entries)
1570 }
1571 };
1572
1573 let failure = entries.last().and_then(|(_, err)| err.clone());
1574 Some(PolledHistory {
1575 depth,
1576 times: entries.into_iter().map(|(time, _)| time).collect(),
1577 failure,
1578 })
1579
1580 }
1581
1582 fn check_pollable(&self, kind: PolledKind, name: &str) -> Result<Arc<str>> {
1583 match kind {
1584 PolledKind::Attr => {
1585 let attr = self.find_attr(name)?;
1586 if !attr.is_pollable() {
1587 return Err(Error::new(ErrorKind::Reason(reason::API_NotSupported))
1588 .with_description(format!("Attribute {:?} of device {:?} cannot be polled, it is write-only",
1589 attr.name(), self.name)));
1590 }
1591 Ok(Arc::from(attr.name()))
1592 }
1593 PolledKind::Command => {
1594 let command = self.find_command(name)?;
1595 if !command.is_pollable() {
1596 return Err(Error::new(ErrorKind::Reason(reason::API_NotSupported))
1597 .with_description(format!("Command {:?} of device {:?} cannot be polled, only one taking no argument can",
1598 command.name(), self.name)));
1599 }
1600 Ok(Arc::from(command.name()))
1601 }
1602 }
1603 }
1604
1605}
1606
1607fn first_number<T: std::str::FromStr>(dev: &str, prop: &str, values: &Spectrum<str>) -> Option<T> {
1610 let value = values.iter().next()?;
1611 match value.trim().parse() {
1612 Ok(number) => Some(number),
1613 Err(_) => {
1614 warn!(%dev, %prop, %value, "Ignored an invalid polling property");
1615 None
1616 }
1617 }
1618}
1619
1620impl<M: Model> Debug for Dev<M> {
1621 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1622 f.debug_struct("Dev").field("name", &self.name).finish_non_exhaustive()
1623 }
1624}
1625
1626struct DevMutable<M: Model> {
1629 black_box: VecDeque<String>,
1632 commands: HashMap<ArcName, Arc<Command<M>>>,
1634 attrs: HashMap<ArcName, Arc<Attr<M>>>,
1635 state_attr: Arc<Attr<M>>,
1638 status_attr: Arc<Attr<M>>,
1639 auto_alarm_states: Arc<[DevState]>,
1642}
1643
1644pub(super) struct ReadCtx<M: Model> {
1648 alarm: Option<AlarmCtx<M>>,
1651 state: Option<DevState>,
1654}
1655
1656struct AlarmCtx<M: Model> {
1661 states: Arc<[DevState]>,
1662 attrs: Vec<AlarmAttr<M>>,
1663}
1664
1665struct AlarmAttr<M: Model> {
1667 attr: Arc<Attr<M>>,
1668 read: Option<Option<AttrAlarm>>,
1673}
1674
1675impl<M: Model> ReadCtx<M> {
1676
1677 fn new(dev: &Dev<M>, involves_state_status: bool) -> Self {
1682
1683 let alarm = involves_state_status.then(|| {
1684
1685 let mutable = dev.mutable.lock().unwrap();
1686 if mutable.auto_alarm_states.is_empty() {
1687 return None;
1688 }
1689
1690 let mut attrs = mutable.attrs.values()
1691 .filter(|attr| attr.has_alarms())
1692 .map(|attr| AlarmAttr {
1693 attr: Arc::clone(attr),
1694 read: None,
1695 })
1696 .collect::<Vec<_>>();
1697
1698 if attrs.is_empty() {
1699 return None;
1700 }
1701
1702 attrs.sort_by(|a, b| a.attr.name().cmp(b.attr.name()));
1703 Some(AlarmCtx {
1704 states: Arc::clone(&mutable.auto_alarm_states),
1705 attrs,
1706 })
1707
1708 }).flatten();
1709
1710 Self {
1711 alarm,
1712 state: None,
1713 }
1714
1715 }
1716
1717 fn has_alarm(&self) -> bool {
1719 self.alarm.is_some()
1720 }
1721
1722 fn may_alarm(&self, state: DevState) -> bool {
1725 self.alarm.as_ref().is_some_and(|alarm| alarm.states.contains(&state))
1726 }
1727
1728 fn record(&mut self, attr: &Arc<Attr<M>>, alarm: Option<AttrAlarm>) {
1731
1732 let alarm_attr = self.alarm.iter_mut()
1733 .flat_map(|alarm| &mut alarm.attrs)
1734 .find(|alarm_attr| Arc::ptr_eq(&alarm_attr.attr, attr));
1735
1736 if let Some(alarm_attr) = alarm_attr {
1737 alarm_attr.read = Some(alarm);
1738 }
1739
1740 }
1741
1742 fn access(&self, dev: &Dev<M>) -> Access {
1745 self.alarm.iter()
1746 .flat_map(|alarm| &alarm.attrs)
1747 .fold(Access::Shared, |access, alarm_attr| access.merge(alarm_attr.attr.read_access(dev, self)))
1748 }
1749
1750 fn may_block(&self, dev: &Dev<M>) -> bool {
1753 self.alarm.iter()
1754 .flat_map(|alarm| &alarm.attrs)
1755 .any(|alarm_attr| alarm_attr.attr.read_may_block(dev, self))
1756 }
1757
1758}
1759
1760fn read_now(value: impl IntoValue) -> AttrRead<Value> {
1762 AttrRead {
1763 value: Some(value.into_value()),
1764 w_value: None,
1765 quality: AttrQuality::Valid,
1766 time: SystemTime::now(),
1767 }
1768}
1769
1770#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1774pub enum Access {
1775 Shared,
1776 Exclusive,
1777}
1778
1779impl Access {
1780
1781 pub fn merge(self, other: Self) -> Self {
1785 match (self, other) {
1786 (Self::Shared, Self::Shared) => Self::Shared,
1787 _ => Self::Exclusive,
1788 }
1789 }
1790
1791}
1792
1793pub enum ModelGuard<'a, M> {
1796 Shared(async_lock::RwLockReadGuard<'a, Option<Result<M>>>),
1797 Exclusive(async_lock::RwLockWriteGuard<'a, Option<Result<M>>>),
1798}
1799
1800impl<M> ModelGuard<'_, M> {
1801
1802 #[inline]
1804 fn get(&self) -> &Option<Result<M>> {
1805 match self {
1806 Self::Shared(guard) => guard,
1807 Self::Exclusive(guard) => guard,
1808 }
1809 }
1810
1811 fn failure(&self) -> Option<Option<Error>> {
1814 match self.get() {
1815 Some(Ok(_)) => None,
1816 Some(Err(err)) => Some(Some(err.clone())),
1817 None => Some(None),
1818 }
1819 }
1820
1821 pub(super) fn model(&mut self) -> ModelRef<'_, M> {
1827 match self {
1828 Self::Shared(guard) => match &**guard {
1829 Some(Ok(model)) => ModelRef::Shared(model),
1830 _ => panic!("the device has no model"),
1831 }
1832 Self::Exclusive(guard) => match &mut **guard {
1833 Some(Ok(model)) => ModelRef::Exclusive(model),
1834 _ => panic!("the device has no model"),
1835 }
1836 }
1837 }
1838
1839}
1840
1841pub enum ModelRef<'m, M> {
1845 Shared(&'m M),
1846 Exclusive(&'m mut M),
1847}
1848
1849impl<M> ModelRef<'_, M> {
1850
1851 #[inline]
1853 pub fn shared(&self) -> &M {
1854 match self {
1855 Self::Shared(model) => model,
1856 Self::Exclusive(model) => model,
1857 }
1858 }
1859
1860 #[inline]
1866 pub fn exclusive(&mut self) -> Result<&mut M> {
1867 match self {
1868 Self::Exclusive(model) => Ok(model),
1869 Self::Shared(_) => Err(Error::new(ErrorKind::Internal)
1870 .with_description("The model was borrowed shared while a handler of that \
1871 request needs it exclusively")),
1872 }
1873 }
1874
1875}
1876
1877fn make_failed_status(err: &Error) -> Scalar<str> {
1880 Scalar::from(format!("The initialization of the device failed:\n{err}").as_str())
1881}
1882
1883fn init_err(failure: Option<&Error>, name: &str) -> Error {
1886 match failure {
1887 Some(err) => Error::new(ErrorKind::DevInitFailed)
1888 .with_description(format!("Device {name:?} failed to initialize"))
1889 .with_parent(err.clone()),
1890 None => Error::new(ErrorKind::DevNotConnected)
1891 .with_description(format!("Device {name:?} is not initialized")),
1892 }
1893}
1894
1895pub fn some_doc(doc: &Cow<'static, str>) -> Option<Cow<'static, str>> {
1897 (!doc.is_empty()).then(|| doc.clone())
1898}
1899
1900#[repr(transparent)]
1909struct DevImpl<M: Model>(Dev<M>);
1910
1911impl<M: Model> DevImpl<M> {
1912
1913 #[inline]
1915 fn from_arc(dev: Arc<Dev<M>>) -> Arc<Self> {
1916 unsafe { Arc::from_raw(Arc::into_raw(dev) as *const Self) }
1919 }
1920
1921}
1922
1923impl<M: Model> Debug for DevImpl<M> {
1924 #[inline]
1925 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1926 Debug::fmt(&self.0, f)
1927 }
1928}
1929
1930impl<M: Model> BackendDev for DevImpl<M> {
1939
1940 fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
1941
1942 if let Err(err) = self.0.check_withdrawn() {
1943 return Box::pin(ready(Err(err)));
1944 }
1945
1946 Box::pin(ready(Ok(Arc::clone(&self.0.name))))
1947
1948 }
1949
1950 fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
1951
1952 if let Err(err) = self.0.check_withdrawn() {
1953 return Box::pin(ready(Err(err)));
1954 }
1955
1956 Box::pin(ready(Ok(Arc::clone(&self.0.server().admin_dev_name))))
1957
1958 }
1959
1960 fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
1961
1962 if let Err(err) = self.0.check_withdrawn() {
1963 return Box::pin(ready(Err(err)));
1964 }
1965
1966 Box::pin(ready(Ok(Arc::from(&*self.0.class.def().doc))))
1967
1968 }
1969
1970 fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {
1974
1975 if let Err(err) = self.0.check_withdrawn() {
1976 return Box::pin(ready(Err(err)));
1977 }
1978
1979 let def = self.0.class.def();
1980 let server = self.0.server();
1981 let backend = match server.backend_checked() {
1982 Ok(backend) => backend,
1983 Err(err) => return Box::pin(ready(Err(err))),
1984 };
1985
1986 Box::pin(ready(Ok(DevInfo {
1987 class: def.name.to_string(),
1988 server_id: server.name().to_string(),
1989 server_host: backend.host().to_string(),
1990 server_version: backend.version(),
1991 doc_url: def.doc_url.clone(),
1992 dev_type: Some(def.doc.clone()),
1995 version_info: vec![
1996 (Cow::Borrowed("rstango"), Cow::Borrowed(env!("CARGO_PKG_VERSION"))),
1997 ],
1998 })))
1999
2000 }
2001
2002 fn ping(&self) -> BoxFuture<'_, Result<()>> {
2003
2004 if let Err(err) = self.0.check_withdrawn() {
2005 return Box::pin(ready(Err(err)));
2006 }
2007
2008 Box::pin(ready(Ok(())))
2009
2010 }
2011
2012 fn state(&self) -> BoxFuture<'_, Result<DevState>> {
2013
2014 if let Err(err) = self.0.check_withdrawn() {
2015 return Box::pin(ready(Err(err)));
2016 }
2017
2018 let dev = self.0.arc();
2019 let command = Arc::new(Command::new_with_attr(dev.state_attr()));
2020 let ctx = ReadCtx::new(&dev, true);
2021 pool::run(command.may_block(&dev, &ctx), async move {
2022 DevState::try_from(dev.exec_command(Ok(command), Value::Void, ctx, DevSource::CacheDev).await?)
2023 })
2024
2025 }
2026
2027 fn status(&self) -> BoxFuture<'_, Result<Scalar<str>>> {
2028
2029 if let Err(err) = self.0.check_withdrawn() {
2030 return Box::pin(ready(Err(err)));
2031 }
2032
2033 let dev = self.0.arc();
2034 let command = Arc::new(Command::new_with_attr(dev.status_attr()));
2035 let ctx = ReadCtx::new(&dev, true);
2036 pool::run(command.may_block(&dev, &ctx), async move {
2037 Scalar::<str>::try_from(dev.exec_command(Ok(command), Value::Void, ctx, DevSource::CacheDev).await?)
2038 })
2039
2040 }
2041
2042 fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Spectrum<str>>> {
2043
2044 if let Err(err) = self.0.check_withdrawn() {
2045 return Box::pin(ready(Err(err)));
2046 }
2047
2048 let mutable = self.0.mutable.lock().unwrap();
2049 let black_box = Spectrum::from_strs(mutable.black_box.iter().take(max.unwrap_or(usize::MAX)))
2050 .expect("strings of more than 4 GiB");
2051
2052 Box::pin(ready(Ok(black_box)))
2053
2054 }
2055
2056 fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {
2057
2058 if let Err(err) = self.0.check_withdrawn() {
2059 return Box::pin(ready(Err(err)));
2060 }
2061
2062 let dev = self.0.arc();
2063 let target = dev.find_command(name.value());
2064 let ctx = ReadCtx::new(&dev, target.as_ref().is_ok_and(|command| command.involves_state()));
2065
2066 let blocking = target.as_ref().is_ok_and(|command| command.may_block(&dev, &ctx));
2067 pool::run(blocking, async move {
2068 let res = dev.exec_command(target, input, ctx, source).await;
2069 if let Err(err) = &res {
2070 debug!(dev = %dev.name, command = %name.value(), %err, "Failed to execute the command");
2071 }
2072 res
2073 })
2074
2075 }
2076
2077 fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
2078
2079 if let Err(err) = self.0.check_withdrawn() {
2080 return Box::pin(ready(Err(err)));
2081 }
2082
2083 let dev = self.0.arc();
2084 let targets = dev.attr_targets(names);
2085 let ctx = ReadCtx::new(&dev, targets.iter().flatten().any(|attr| attr.role() != AttrRole::Interface));
2086
2087 let blocking = targets.iter().flatten().any(|attr| attr.read_may_block(&dev, &ctx));
2088 pool::run(blocking, async move { dev.exec_read_attr(targets, ctx, source).await })
2089
2090 }
2091
2092 fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
2093
2094 if let Err(err) = self.0.check_withdrawn() {
2095 return Box::pin(ready(Err(err)));
2096 }
2097
2098 let dev = self.0.arc();
2099 let targets = values.into_iter()
2100 .map(|write| (dev.find_attr(write.name.value()), write.value))
2101 .collect::<Vec<_>>();
2102
2103 let blocking = targets.iter()
2104 .filter_map(|(target, _)| target.as_ref().ok())
2105 .any(|attr| attr.write_may_block());
2106 pool::run(blocking, async move { dev.exec_write_attr(targets).await })
2107
2108 }
2109
2110 fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
2111
2112 if let Err(err) = self.0.check_withdrawn() {
2113 return Box::pin(ready(Err(err)));
2114 }
2115
2116 let dev = self.0.arc();
2117 let targets = values.into_iter()
2118 .map(|write| {
2119 let target = dev.find_attr(write.name.value());
2120 (resolved(write.name, target), write.value)
2121 })
2122 .collect::<Vec<_>>();
2123
2124 let ctx = ReadCtx::new(&dev, false);
2125 let blocking = targets.iter()
2126 .filter_map(|(target, _)| target.as_ref().ok())
2127 .any(|attr| attr.write_may_block() || attr.read_may_block(&dev, &ctx));
2128 pool::run(blocking, async move { dev.exec_write_read_attr(targets, ctx).await })
2129
2130 }
2131
2132 fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {
2136
2137 if let Err(err) = self.0.check_withdrawn() {
2138 return Box::pin(ready(Err(err)));
2139 }
2140
2141 let infos = self.0.command_targets(names).into_iter()
2142 .map(|target| match target {
2143 Ok(target) => Named::new(Scalar::from(target.name()), Ok(target.info())),
2144 Err(unresolved) => unresolved.map(Err),
2145 })
2146 .collect();
2147
2148 Box::pin(ready(Ok(infos)))
2149
2150 }
2151
2152 fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {
2154
2155 if let Err(err) = self.0.check_withdrawn() {
2156 return Box::pin(ready(Err(err)));
2157 }
2158
2159 let infos = self.0.attr_targets(names).into_iter()
2160 .map(|target| match target {
2161 Ok(target) => Named::new(Scalar::from(target.name()), target.info()),
2162 Err(unresolved) => unresolved.map(Err),
2163 })
2164 .collect();
2165
2166 Box::pin(ready(Ok(infos)))
2167
2168 }
2169
2170 fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
2172
2173 if let Err(err) = self.0.check_withdrawn() {
2174 return Box::pin(ready(Err(err)));
2175 }
2176
2177 Box::pin(async move { Ok(self.0.exec_set_attr_config(configs).await) })
2178
2179 }
2180
2181 fn command_history(&self, name: Scalar<str>, max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<Value>>>> {
2185
2186 if let Err(err) = self.0.check_withdrawn() {
2187 return Box::pin(ready(Err(err)));
2188 }
2189
2190 Box::pin(ready(self.0.find_command(name.value()).and_then(|command| {
2191 let history = command.history(max);
2192 if history.is_empty() {
2193 return Err(Error::new(ErrorKind::CommandNotPolled)
2194 .with_description(format!("Command {:?} of device {:?} has no history recorded",
2195 command.name(), self.0.name)));
2196 }
2197 Ok(history)
2198 })))
2199
2200 }
2201
2202 fn attr_history(&self, name: Scalar<str>, max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<AttrRead<Value>>>>> {
2204
2205 if let Err(err) = self.0.check_withdrawn() {
2206 return Box::pin(ready(Err(err)));
2207 }
2208
2209 Box::pin(ready(self.0.find_attr(name.value()).and_then(|attr| {
2210 let history = attr.history(max);
2211 if history.is_empty() {
2212 return Err(Error::new(ErrorKind::AttrNotPolled)
2213 .with_description(format!("Attribute {:?} of device {:?} has no history recorded",
2214 attr.name(), self.0.name)));
2215 }
2216 Ok(history)
2217 })))
2218
2219 }
2220
2221 fn subscribe_attr_value(&self, _admin: Arc<dyn BackendDev>, name: Scalar<str>, event: AttrValueEvent) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
2227
2228 if let Err(err) = self.0.check_withdrawn() {
2229 return Box::pin(ready(Err(err)));
2230 }
2231
2232 Box::pin(ready(self.0.find_attr(name.value()).and_then(|target| {
2233 target.check_push_event(&self.0, event)?;
2234 Ok(self.0.server().subscribe_attr_value(&self.0.name, target.name(), event))
2235 })))
2236
2237 }
2238
2239 fn subscribe_attr_ready(&self, _admin: Arc<dyn BackendDev>, name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrReady>>>> {
2240
2241 if let Err(err) = self.0.check_withdrawn() {
2242 return Box::pin(ready(Err(err)));
2243 }
2244
2245 Box::pin(ready(self.0.find_attr(name.value()).and_then(|target| {
2246 target.check_data_ready(&self.0)?;
2247 Ok(self.0.server().subscribe_attr_ready(&self.0.name, target.name()))
2248 })))
2249
2250 }
2251
2252 fn subscribe_attr_config(&self, _admin: Arc<dyn BackendDev>, name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrInfo>>>> {
2253
2254 if let Err(err) = self.0.check_withdrawn() {
2255 return Box::pin(ready(Err(err)));
2256 }
2257
2258 Box::pin(ready(self.0.find_attr(name.value()).map(|target| {
2259 self.0.server().subscribe_attr_config(&self.0.name, target.name())
2260 })))
2261
2262 }
2263
2264}
2265
2266type Resolved<T> = std::result::Result<T, Named<Error>>;
2269
2270fn resolved<T>(name: Scalar<str>, res: Result<T>) -> Resolved<T> {
2271 res.map_err(|err| Named::new(name, err))
2272}
2273
2274fn scalars(names: &Spectrum<str>) -> impl Iterator<Item = Scalar<str>> + '_ {
2276 (0..names.len()).map(|index| names.get_scalar([index]).expect("a name at every index"))
2277}
2278
2279
2280#[cfg(test)]
2281mod tests {
2282
2283 use std::sync::Mutex;
2284 use std::thread;
2285 use std::time::Duration;
2286
2287 use futures::channel::oneshot;
2288 use futures::executor::block_on;
2289 use futures::{FutureExt, StreamExt};
2290
2291 use crate::dev::{ToNames, WriteAttrSpec};
2292 use crate::value::FromValue;
2293
2294 use super::*;
2295 use super::super::def::{AttrDef, ClassDef, CommandDef};
2296 use super::super::server::ServerBuilder;
2297 use crate::testing::TestContext;
2298
2299 fn serve<M: Model>(def: ClassDef<M>) -> (TestContext, Arc<DevImpl<M>>) {
2302 let ctx = TestContext::new().unwrap();
2303 ctx.database().add_server("Test/test", &[]).unwrap();
2304 ctx.database().add_dev("test/dev/1", &def.name, "Test/test").unwrap();
2305 let server = ctx.serve(ServerBuilder::new("Test", "test")).unwrap();
2306 let class = server.add_class(def).unwrap();
2307 let dev = class.add_dev("test/dev/1").unwrap();
2308 let dev = DevImpl::from_arc(dev);
2309 drop(class);
2310 (ctx, dev)
2311 }
2312
2313 #[test]
2314 fn push_event() {
2315
2316 let (ctx, dev) = serve(model_def());
2317 let server = ctx.server("Test/test").unwrap();
2318 server.sync().unwrap();
2319 let mut state = server.subscribe_attr_value("test/dev/1", "State", AttrValueEvent::Change);
2320 let mut count = server.subscribe_attr_value("test/dev/1", "Count", AttrValueEvent::Change);
2321
2322 thread::scope(|scope| {
2323 scope.spawn(|| dev.0.push_change_event("state", DevState::Moving).unwrap());
2324 });
2325
2326 let read = block_on(state.next()).unwrap().unwrap();
2327 assert_eq!(read.value, Some(Value::from(DevState::Moving)));
2328 assert_eq!(read.w_value, None);
2329 assert_eq!(read.quality, AttrQuality::Valid);
2330
2331 dev.0.set_attr_w_value("Count", 5).unwrap();
2333 dev.0.push_change_event("Count", 3).unwrap();
2334 let read = block_on(count.next()).unwrap().unwrap();
2335 assert_eq!(read.value, Some(Value::from(3)));
2336 assert_eq!(read.w_value, Some(Value::from(5)));
2337
2338 assert!(dev.0.push_change_event("Missing", 1.0).is_err());
2339
2340 let mut empty = read_now(3);
2342 empty.value = None;
2343 assert_eq!(dev.0.push_attr_event("Count", AttrValueEvent::Change, Ok(empty)).unwrap_err().kind(), ErrorKind::InvalidData);
2344 let alarm = read_now(DevState::Moving).with_quality(AttrQuality::Alarm);
2345 assert_eq!(dev.0.push_attr_event("State", AttrValueEvent::Change, Ok(alarm)).unwrap_err().kind(), ErrorKind::InvalidData);
2346
2347 let err = dev.0.push_archive_event("Count", 3).unwrap_err();
2349 assert_eq!(err.kind(), ErrorKind::Reason(reason::API_AttributePollingNotStarted));
2350 dev.0.push_user_event("Count", 3).unwrap();
2351
2352 }
2353
2354 #[test]
2355 fn set_attr_config() {
2356
2357 let (ctx, dev) = serve(model_def());
2358 let server = ctx.server("Test/test").unwrap();
2359 server.sync().unwrap();
2360 let mut conf = server.subscribe_attr_config("test/dev/1", "Count");
2361
2362 let mut config = dev.0.get_attr("Count").unwrap().config();
2363 config.label = Some("Counter".into());
2364 dev.0.set_attr_config("count", &config).unwrap();
2365
2366 assert_eq!(dev.0.get_attr("Count").unwrap().config().label.as_deref(), Some("Counter"));
2367 let info = block_on(conf.next()).unwrap().unwrap();
2368 assert_eq!(info.config.label.as_deref(), Some("Counter"));
2369
2370 dev.0.set_attr_config("Count", &config).unwrap();
2372 assert!(conf.next().now_or_never().is_none());
2373
2374 }
2375
2376 #[test]
2380 fn attr_history() {
2381
2382 let (_ctx, dev) = serve(model_def());
2383 dev.0.server().sync().unwrap();
2384
2385 let read = |name: &str, source| block_on(dev.read_attr(name.to_names(), source)).unwrap().pop().unwrap().value;
2386
2387 let err = read("Count", DevSource::Cache).unwrap_err();
2388 assert_eq!(err.kind(), ErrorKind::AttrNotPolled);
2389 assert!(read("Count", DevSource::CacheDev).is_ok());
2390 assert_eq!(block_on(dev.attr_history(Scalar::from("Count"), 10)).unwrap_err().kind(), ErrorKind::AttrNotPolled);
2391
2392 let count = dev.0.get_attr("Count").unwrap();
2393 count.push_history(Ok(read_now(5)), None).unwrap();
2394 count.push_history(Ok(read_now(6)), None).unwrap();
2395 assert_eq!(read("Count", DevSource::Cache).unwrap().value, Some(Value::from(6)));
2396 assert_eq!(read("Count", DevSource::Dev).unwrap().value, Some(Value::from(0)));
2397
2398 let history = block_on(dev.attr_history(Scalar::from("Count"), 10)).unwrap();
2399 let values = history.into_iter().map(|entry| entry.value.unwrap().value.unwrap()).collect::<Vec<_>>();
2400 assert_eq!(values, [Value::from(5), Value::from(6)]);
2401
2402 count.set_history_depth(1);
2403 assert_eq!(count.history(10).len(), 1);
2404
2405 count.push_history(Ok(read_now(7)), Some(Duration::ZERO)).unwrap();
2407 assert_eq!(read("Count", DevSource::Cache).unwrap_err().kind(), ErrorKind::ValueOutdated);
2408 assert_eq!(read("Count", DevSource::CacheDev).unwrap().value, Some(Value::from(0)));
2409
2410 count.clear_history();
2411 assert!(count.history(10).is_empty());
2412
2413 let state = dev.0.get_attr("State").unwrap();
2415 let mut alarm = read_now(DevState::Fault);
2416 alarm.quality = AttrQuality::Alarm;
2417 assert!(state.push_history(Ok(alarm), None).is_err());
2418 let mut empty = read_now(DevState::Fault);
2419 empty.value = None;
2420 assert!(state.push_history(Ok(empty), None).is_err());
2421 state.push_history(Ok(read_now(DevState::Fault)), None).unwrap();
2422 let value = block_on(dev.command(Scalar::from("State"), Value::Void, DevSource::Cache)).unwrap();
2423 assert_eq!(DevState::try_from(value).unwrap(), DevState::Fault);
2424 let history = block_on(dev.command_history(Scalar::from("state"), 10)).unwrap();
2425 assert_eq!(history.len(), 1);
2426
2427 }
2428
2429 #[test]
2432 fn command_history() {
2433
2434 let (_ctx, dev) = serve(model_def());
2435 dev.0.server().sync().unwrap();
2436
2437 let exec = |name: &str, source| block_on(dev.command(Scalar::from(name), Value::Void, source));
2438
2439 assert_eq!(exec("GetCount", DevSource::Cache).unwrap_err().kind(), ErrorKind::CommandNotPolled);
2440 assert_eq!(exec("GetCount", DevSource::CacheDev).unwrap(), Value::from(0));
2441 assert_eq!(block_on(dev.command_history(Scalar::from("GetCount"), 10)).unwrap_err().kind(), ErrorKind::CommandNotPolled);
2442
2443 let get_count = dev.0.get_command("GetCount").unwrap();
2444 get_count.push_history(Ok(Value::from(5)), None).unwrap();
2445 get_count.push_history(Err(Error::new(ErrorKind::TimedOut)), None).unwrap();
2446 assert_eq!(exec("GetCount", DevSource::Cache).unwrap_err().kind(), ErrorKind::TimedOut);
2447 assert_eq!(exec("GetCount", DevSource::Dev).unwrap(), Value::from(0));
2448
2449 let history = block_on(dev.command_history(Scalar::from("getcount"), 10)).unwrap();
2450 assert_eq!(history.len(), 2);
2451 assert_eq!(history[0].value.as_ref().unwrap(), &Value::from(5));
2452 assert!(history[1].value.is_err());
2453
2454 get_count.push_history(Ok(Value::from(6)), Some(Duration::ZERO)).unwrap();
2456 assert_eq!(exec("GetCount", DevSource::Cache).unwrap_err().kind(), ErrorKind::ValueOutdated);
2457 assert_eq!(exec("GetCount", DevSource::CacheDev).unwrap(), Value::from(0));
2458
2459 let incr = dev.0.get_command("Incr").unwrap();
2460 assert!(incr.push_history(Ok(Value::from(1)), None).is_err());
2461 assert!(dev.0.get_command("Init").unwrap().push_history(Ok(Value::Void), None).is_err());
2462 assert_eq!(exec("Init", DevSource::Cache).unwrap_err().kind(), ErrorKind::CommandNotPolled);
2463
2464 dev.0.get_command("State").unwrap().push_history(Ok(DevState::Alarm.into()), None).unwrap();
2466 let read = dev.0.get_attr("State").unwrap().history(1).pop().unwrap().value.unwrap();
2467 assert_eq!(read.value, Some(Value::from(DevState::Alarm)));
2468
2469 }
2470
2471 fn failing_def() -> ClassDef<Motor> {
2473 ClassDef::<Motor>::new_async_with("Motor", |_dev| Box::pin(async {
2474 Err(Error::new(ErrorKind::Internal).with_description("No hardware"))
2475 }))
2476 .command(CommandDef::new("GetCount").exec_async(Motor::get_count))
2477 .attr(AttrDef::new("Count").read_async(Motor::count))
2478 }
2479
2480 #[test]
2483 fn no_model_reads() {
2484
2485 let (_ctx, dev) = serve(failing_def());
2486 let read = |names: &[&str]| block_on(dev.read_attr(names.to_names(), DevSource::Dev)).unwrap();
2487 let value = |read: &NamedResult<AttrRead<Value>>| read.value.as_ref().unwrap().value.clone().unwrap();
2488
2489 let reads = read(&["State", "Status"]);
2491 assert_eq!(DevState::try_from(value(&reads[0])).unwrap(), DevState::Unknown);
2492 assert_eq!(Scalar::<str>::try_from(value(&reads[1])).unwrap().value(), MISSING_STATUS);
2493
2494 dev.0.server().sync().unwrap();
2495
2496 let reads = read(&["State", "Status", "Count", "Missing"]);
2497 assert_eq!(DevState::try_from(value(&reads[0])).unwrap(), DevState::Fault);
2498 assert!(Scalar::<str>::try_from(value(&reads[1])).unwrap().value().contains("No hardware"));
2499 assert_eq!(reads[2].value.as_ref().unwrap_err().kind(), ErrorKind::DevInitFailed);
2500 assert_eq!(reads[3].value.as_ref().unwrap_err().kind(), ErrorKind::AttrNotFound);
2501
2502 }
2503
2504 #[test]
2507 fn exec_many_command() {
2508
2509 let exec = |dev: &DevImpl<Motor>, names: &[&str]| {
2510 let targets = names.iter()
2511 .map(|&name| (resolved(Scalar::from(name), dev.0.find_command(name)), Value::Void))
2512 .collect();
2513 block_on(dev.0.exec_many_command(targets, ReadCtx::new(&dev.0, true), DevSource::Dev)).unwrap()
2514 };
2515
2516 let (_ctx, dev) = serve(model_def());
2517 dev.0.server().sync().unwrap();
2518
2519 let results = exec(&dev, &["GetCount", "State", "Missing"]);
2520 assert_eq!(results[0].value.as_ref().unwrap(), &Value::from(0));
2521 assert_eq!(DevState::try_from(results[1].value.clone().unwrap()).unwrap(), DevState::On);
2522 assert_eq!(results[2].value.as_ref().unwrap_err().kind(), ErrorKind::CommandNotFound);
2523
2524 let (_ctx, dev) = serve(failing_def());
2525 dev.0.server().sync().unwrap();
2526
2527 let results = exec(&dev, &["GetCount", "State", "Status"]);
2528 assert_eq!(results[0].value.as_ref().unwrap_err().kind(), ErrorKind::DevInitFailed);
2529 assert_eq!(DevState::try_from(results[1].value.clone().unwrap()).unwrap(), DevState::Fault);
2530 assert!(Scalar::<str>::try_from(results[2].value.clone().unwrap()).unwrap().value().contains("No hardware"));
2531
2532 }
2533
2534 fn thread_name() -> String {
2535 thread::current().name().unwrap_or_default().to_string()
2536 }
2537
2538 struct Motor {
2539 count: i32,
2540 }
2541
2542 impl Motor {
2543
2544 async fn init(_dev: &Dev<Self>) -> Result<Self> {
2545 Ok(Self { count: 0 })
2546 }
2547
2548 async fn incr(&mut self, _dev: &Dev<Self>, arg: i32) -> Result<i32> {
2549 self.count += arg;
2550 Ok(self.count)
2551 }
2552
2553 async fn count(&self, _dev: &Dev<Self>) -> Result<i32> {
2554 Ok(self.count)
2555 }
2556
2557 async fn get_count(&self, _dev: &Dev<Self>, _: ()) -> Result<i32> {
2558 Ok(self.count)
2559 }
2560
2561 async fn set_count(&mut self, _dev: &Dev<Self>, arg: i32) -> Result<()> {
2562 self.count = arg;
2563 Ok(())
2564 }
2565
2566 async fn state(&self, _dev: &Dev<Self>) -> Result<DevState> {
2567 Ok(if self.count == 0 { DevState::On } else { DevState::Moving })
2568 }
2569
2570 fn thread(&self, _dev: &Dev<Self>) -> Result<String> {
2571 Ok(thread_name())
2572 }
2573
2574 async fn thread_async(&self, _dev: &Dev<Self>) -> Result<String> {
2575 Ok(thread_name())
2576 }
2577
2578 }
2579
2580 fn model_def() -> ClassDef<Motor> {
2581 ClassDef::new_async("Motor", Motor::init)
2582 .state_async(Motor::state)
2583 .push_state_events(Some(false), None)
2584 .command(CommandDef::new("Incr").exec_exclusive_async(Motor::incr))
2585 .command(CommandDef::new("GetCount").exec_async(Motor::get_count))
2586 .attr(AttrDef::new("Count").read_async(Motor::count).write_exclusive_async(Motor::set_count)
2587 .push_change_event(false))
2588 .attr(AttrDef::new("Thread").read(Motor::thread))
2589 .attr(AttrDef::new("ThreadAsync").read_async(Motor::thread_async))
2590 }
2591
2592 fn read_one<T: FromValue>(dev: &DevImpl<Motor>, name: &str) -> T {
2593 let mut reads = block_on(dev.read_attr(name.to_names(), DevSource::Dev)).unwrap();
2594 let read = reads.pop().unwrap().value;
2595 T::from_value(read.unwrap().value.unwrap()).unwrap()
2596 }
2597
2598 #[test]
2601 fn async_handlers() {
2602
2603 let (_ctx, dev) = serve(model_def());
2604 dev.0.server().sync().unwrap();
2605
2606 assert_eq!(block_on(dev.state()).unwrap(), DevState::On);
2607 let count = block_on(dev.command(Scalar::from("Incr"), Value::from(3), DevSource::Dev)).unwrap();
2608 assert_eq!(i32::from_value(count).unwrap(), 3);
2609 assert_eq!(block_on(dev.state()).unwrap(), DevState::Moving);
2610
2611 let writes = block_on(dev.write_attr(("Count", 7).into_values())).unwrap();
2612 assert!(writes.into_iter().all(|write| write.is_ok()));
2613 assert_eq!(read_one::<i32>(&dev, "Count"), 7);
2614
2615 block_on(dev.command(Scalar::from("Init"), Value::Void, DevSource::Dev)).unwrap();
2617 assert_eq!(read_one::<i32>(&dev, "Count"), 0);
2618
2619 }
2620
2621 #[test]
2624 fn unknown_attr_name() {
2625
2626 let (_ctx, dev) = serve(model_def());
2627 dev.0.server().sync().unwrap();
2628
2629 let names = Spectrum::from(&["count", "Nothing"][..]);
2630 let reads = block_on(dev.read_attr(Some(names.clone()), DevSource::Dev)).unwrap();
2631
2632 assert_eq!(reads[0].name.value(), "Count");
2633 assert!(reads[0].value.is_ok());
2634
2635 assert_eq!(reads[1].name.value().as_ptr(), names.get([1]).unwrap().as_ptr());
2636 assert_eq!(reads[1].value.as_ref().unwrap_err().kind(), ErrorKind::AttrNotFound);
2637
2638 }
2639
2640 #[test]
2644 fn request_threads() {
2645
2646 let (_ctx, dev) = serve(model_def());
2647 dev.0.server().sync().unwrap();
2648
2649 let on_pool = |name: &str| name.starts_with("pool-");
2650 assert!(on_pool(&read_one::<String>(&dev, "Thread")));
2651 assert!(!on_pool(&read_one::<String>(&dev, "ThreadAsync")));
2652
2653 let reads = block_on(dev.read_attr(["ThreadAsync", "Thread"].to_names(), DevSource::Dev)).unwrap();
2654 for read in reads {
2655 assert!(on_pool(&String::from_value(read.value.unwrap().value.unwrap()).unwrap()));
2656 }
2657
2658 }
2659
2660 #[test]
2663 fn lock_across_await() {
2664
2665 let (tx, rx) = oneshot::channel::<()>();
2666 let gate = Mutex::new(Some(rx));
2667 let def = model_def()
2668 .command(CommandDef::new("Wait").exec_exclusive_async_with(move |_model: &mut Motor, _dev, _input| {
2669 let rx = gate.lock().unwrap().take();
2670 Box::pin(async move {
2671 if let Some(rx) = rx {
2672 let _ = rx.await;
2673 }
2674 Ok(Value::Void)
2675 })
2676 }));
2677
2678 let (_ctx, dev) = serve(def);
2679 dev.0.server().sync().unwrap();
2680
2681 let mut wait = dev.command(Scalar::from("Wait"), Value::Void, DevSource::Dev);
2682 assert!((&mut wait).now_or_never().is_none());
2683 let mut read = dev.read_attr("Count".to_names(), DevSource::Dev);
2684 assert!((&mut read).now_or_never().is_none());
2685
2686 tx.send(()).unwrap();
2687 block_on(wait).unwrap();
2688 assert!(block_on(read).unwrap().pop().unwrap().value.is_ok());
2689
2690 }
2691
2692 #[test]
2695 fn async_init_failed() {
2696
2697 let def = ClassDef::<Motor>::new_async_with("Motor", |_dev| Box::pin(async {
2698 Err(Error::new(ErrorKind::Internal).with_description("No hardware"))
2699 }));
2700
2701 let (_ctx, dev) = serve(def);
2702 assert_eq!(block_on(dev.state()).unwrap(), DevState::Unknown);
2703 dev.0.server().sync().unwrap();
2704
2705 assert_eq!(block_on(dev.state()).unwrap(), DevState::Fault);
2706 assert!(block_on(dev.status()).unwrap().value().contains("No hardware"));
2707 assert!(block_on(dev.command(Scalar::from("Init"), Value::Void, DevSource::Dev)).is_err());
2708
2709 }
2710
2711}