Skip to main content

tango/server/
dev.rs

1//! A device: its interface, its model, and every request a client makes of it.
2
3use 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
36/// The model-erased view of a [`Dev`], which is how a server refers to the devices it
37/// serves, whatever the model of the class they belong to.
38pub(super) trait AnyDev: Debug + Send + Sync {
39
40    /// The name of that device, shared so that it is cheaply copied out.
41    fn name(&self) -> &Arc<str>;
42
43    /// The class of that device.
44    fn class(&self) -> &dyn AnyClass;
45
46    /// Initialize that device: its model, if it has one, is deleted, and a new one is built
47    /// by the init function of its class, on the pool of the server when that may block.
48    /// This only fails when the initialization never ran to its end, its function having
49    /// panicked: one that failed is kept by the device instead.
50    fn init(&self) -> BoxFuture<'_, Result<()>>;
51
52    /// Stop serving that device, for good, and delete its model, if it has one, giving it
53    /// back to the delete function of its class so that it may release what it owns: what
54    /// a device goes through once it is removed from its server, or when the server shuts
55    /// down. Not to be confused with the deletion of the model alone, which is also what an
56    /// initialization starts with, see [`Dev::init`].
57    /// 
58    /// The backend is told first, so that no new request reaches a device that is being
59    /// deleted, see [`crate::backend::BackendServer::remove`]. A request it was answering
60    /// may still be running, and the model is only deleted once it is through with it,
61    /// which is why a request of that device must never wait for this.
62    fn withdraw(&self) -> BoxFuture<'_, Result<()>>;
63
64    /// Mark that device as withdrawn, see [`Dev::is_withdrawn`], and take it out of the
65    /// local registry, so that no client of this process finds it anymore: the first step
66    /// of [`Self::withdraw`], and all that becomes of the devices of a server that ended
67    /// without being stopped.
68    fn withdraw_now(&self);
69
70    /// Start serving that device through the backend of its server, giving back the ITR it
71    /// is reached at. This is quick and never blocks, see
72    /// [`crate::backend::BackendServer::add`].
73    fn serve(&self) -> Result<Arc<str>>;
74
75    /// That device as [`Any`], for it to be downcast to the [`Dev`] it is.
76    fn into_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
77
78    /// Poll those objects of that device once, which are due together, see
79    /// [`super::polling`]: read the attributes, or execute the commands, as a client asking
80    /// the device does, on the pool of the server when that may block, and record what
81    /// each gave in its history, valid for as long as its item says. A polled attribute is
82    /// also sent as a periodic event once its period has passed since the last one was
83    /// sent, see [`Dev::poll_attrs`], and when the last one was sent is given back for
84    /// each item, never for a command.
85    ///
86    /// An object the device does not have anymore is left out. This only fails when the
87    /// pool dropped the poll before it ran to its end, see [`pool::run`]: a read or an
88    /// execution that failed is recorded as such.
89    fn poll(&self, kind: PolledKind, items: &[PollItem]) -> BoxFuture<'static, Result<Vec<Option<Instant>>>>;
90
91    /// What the history of that object of that device holds, for its polling status, see
92    /// [`PolledHistory`], absent if the device has no such object.
93    fn polled_history(&self, kind: PolledKind, name: &str) -> Option<PolledHistory>;
94
95    /// The name of that object of that device, as its definition gives it, failing if the
96    /// device has no such object or if it cannot be polled, see [`Attr::is_pollable`] and
97    /// [`Command::is_pollable`].
98    fn check_pollable(&self, kind: PolledKind, name: &str) -> Result<Arc<str>>;
99
100}
101
102/// The handle to a device from within its own execution functions, giving access to the
103/// device outside of its model.
104/// 
105/// A device can be served while having no model: until it is initialized, or when its
106/// initialization failed. It is then in UNKNOWN or FAULT state, with a status describing
107/// why it has no model. The idea behind this is that the Init command, which deletes then
108/// initializes a device, might fail, however we don't want to stop serving the device
109/// for that. This also allows a server to start if any of its devices fails to init, the
110/// device will be available but it will not be usable in any way, it will just be possible
111/// to call Init and try to fix it, individually from the other devices that might be
112/// running.
113pub struct Dev<M: Model> {
114    /// Self reference, for what a device hands the backend of itself.
115    this: Weak<Self>,
116    /// The name of the device.
117    name: Arc<str>,
118    /// The class this device is an instance of, through which its definition is reached.
119    class: Arc<Class<M>>,
120    /// The interface of this device, see [`DevMutable`]. It is locked very shortly so 
121    /// there is no need for 
122    mutable: Mutex<DevMutable<M>>,
123    /// The internal model of the device, absent until the device is initialized.
124    /// 
125    /// An asynchronous lock, held across the awaits of the asynchronous handlers.
126    model: async_lock::RwLock<Option<Result<M>>>,
127    /// The device recorded as served by this process while it is served, which is how a
128    /// client of this process reaches it directly, see [`crate::local`].
129    local_handle: Mutex<Option<LocalDevHandle>>,
130    /// Set once this device is withdrawn from its server, see [`Self::is_withdrawn`].
131    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    /// The name of this device.
158    #[inline]
159    pub fn name(&self) -> &str {
160        &self.name
161    }
162
163    /// The class this device is an instance of.
164    #[inline]
165    pub fn class(&self) -> &Class<M> {
166        &self.class
167    }
168
169    /// The server serving this device.
170    #[inline]
171    pub fn server(&self) -> &Server {
172        self.class.server()
173    }
174
175    /// Whether this device is gone for good: removed from its server and deleted, or its
176    /// server ended. Nothing of it is served from then on, and pushing events or setting
177    /// anything of its attributes fails, which is how a thread of its own learns to stop. A
178    /// device added again under the same name, which is what a restart is, is another
179    /// device, which this one never becomes.
180    #[inline]
181    pub fn is_withdrawn(&self) -> bool {
182        self.withdrawn.load(Ordering::Acquire)
183    }
184
185    /// Fail once this device is withdrawn, see [`Self::is_withdrawn`].
186    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    // The properties of a device are the database's: everything below asks the database 
195    // of the server for what it holds of this device, see the very same methods on
196    // [`crate::database::DatabaseProxy`]. The blocking forms are for synchronous handlers, 
197    // which run on the pool, and must never run in an asynchronous one.
198
199    /// The names of the properties of this device that match the given filter.
200    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    /// The names of the properties of this device, see [`Self::list_prop_async`].
205    pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
206        self.server().database().list_dev_prop(&self.name, filter)
207    }
208
209    /// The properties of this device, parsed into the shape the given names ask for, see
210    /// [`crate::database::DatabaseProxy::get_dev_prop`].
211    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    /// The properties of this device, see [`Self::get_prop_async`].
219    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    /// Write the given properties of this device.
227    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    /// Write the given properties of this device, see [`Self::set_prop_async`].
235    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    /// Delete the given properties of this device.
243    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    /// Delete the given properties of this device, see [`Self::delete_prop_async`].
248    pub fn delete_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
249        self.server().database().delete_dev_prop(&self.name, names)
250    }
251
252    /// This device, shared, which is how a handler, only given a reference to it, hands it
253    /// to a thread of its own. Holding it does not keep it served, see
254    /// [`Self::is_withdrawn`].
255    #[inline]
256    pub fn arc(&self) -> Arc<Self> {
257        // A device is only ever reached through a reference to it, so it is alive.
258        self.this.upgrade().unwrap()
259    }
260
261    /// Add a command to that device only, from its definition, which is what a device
262    /// whose interface depends on its properties does at initialization; a command every
263    /// device of the class has is defined on the class instead, see [`ClassDef::command`].
264    /// 
265    /// A name the class already defines is refused rather than shadowed: the device was
266    /// given every command of its class before its initialization ran.
267    /// 
268    /// [`ClassDef::command`]: super::def::ClassDef::command
269    pub fn add_command(&self, def: CommandDef<M>) -> Result<()> {
270
271        // Refused rather than panicked on, unlike the same name given to a class, see
272        // [`ClassDef::command`]: what a device adds to itself it usually reads from its
273        // properties, so a reserved name here is a configuration to report, not a
274        // definition to fix.
275        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    /// Remove a command from that device, telling whether it was defined.
300    pub fn remove_command(&self, name: &str) -> bool {
301        self.mutable.lock().unwrap().commands.remove(Name::new(name)).is_some()
302    }
303
304    /// Add an attribute to that device only, from its definition, see
305    /// [`Self::add_command`] and [`ClassDef::attr`].
306    /// 
307    /// An attribute added by the initialization of the device is configured from the
308    /// database once the initialization returns. One added at any other time keeps the
309    /// configuration of its definition.
310    /// 
311    /// [`ClassDef::attr`]: super::def::ClassDef::attr
312    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    /// Give the attribute of that name a new set point without writing it, which every read
348    /// gives back as its write value from then on, and which the database remembers if the
349    /// attribute is memorized, see [`AttrMemorized`].
350    /// 
351    /// This is for a device whose set point changes by itself, a motor stopped short of its
352    /// target for instance, so that a client and the next initialization find the value
353    /// the device actually works towards. An attribute that cannot be written, the State
354    /// and the Status included, is refused.
355    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    /// The blocking form of [`Self::set_attr_w_value_async`], which must never run in an
362    /// asynchronous handler; it gives up after the timeout of the database of the server.
363    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    /// Change the configuration of the attribute of that name, given as the whole of it,
368    /// which is what a client changing it does: refused if it is incoherent, stored in the
369    /// database, and pushed to the subscribers of the configuration of that attribute. Its
370    /// current configuration is the one [`Attr::config`] gives.
371    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    /// The blocking form of [`Self::set_attr_config_async`], which must never run in an
377    /// asynchronous handler; it gives up after the timeout of the database of the server.
378    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    /// Set the states in which this device is put in ALARM when one of its attributes is in
383    /// alarm, which is only [`DevState::On`] unless set otherwise, an empty list leaving
384    /// the state of the device to its state function alone.
385    /// 
386    /// When the state function gives one of those states, the attributes that have any
387    /// alarm configured are read, and a single one of them in alarm or in warning makes
388    /// the device ALARM, its status telling which and why, the way cppTango does. Getting
389    /// the state of the device then borrows its model the way reading those attributes
390    /// does.
391    pub fn set_auto_alarm_states(&self, states: &[DevState]) {
392        self.mutable.lock().unwrap().auto_alarm_states = Arc::from(states);
393    }
394
395    /// Remove an attribute from that device, telling whether it was defined.
396    pub fn remove_attr(&self, name: &str) -> bool {
397        self.mutable.lock().unwrap().attrs.remove(Name::new(name)).is_some()
398    }
399
400    /// The command of that name on this device, absent when it has none of that name: the
401    /// State, the Status and the Init every device has, whatever its class defines, or else
402    /// one of its interface.
403    /// 
404    /// Every request on a command goes through this, and it is shared rather than borrowed
405    /// from the interface store: the interface store is unlocked as soon as the command is
406    /// out of it, because a handler is free to change the interface of its own device, and
407    /// that store is the lock taken *under* the model, never around it.
408    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    /// The attribute of that name on this device, the State and the Status included,
421    /// reached the way [`Self::get_command`] reaches a command.
422    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    /// Every command of this device: the three every device has first, then its whole
434    /// interface.
435    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    /// Every attribute of this device: the two every device has first, then its whole
449    /// interface.
450    /// 
451    /// TODO: The order of the interface is the one of the interface store, which is a hash
452    /// map and therefore arbitrary; a client reading a whole interface expects a stable one.
453    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    // The events of the attributes of this device are pushed to the subscriptions its
463    // server keeps of them, see [`Server::events`]. A push never blocks and needs no
464    // executor, so that it is made from anywhere, a thread of the device's own included.
465    // Nothing is sent of an event no client subscribed to.
466
467    /// Push an event of that kind for the attribute of that name, the state and the status
468    /// included: a reading, or the error reading it, failing if the device has no such
469    /// attribute, or if that attribute does not declare it pushes such events, see
470    /// [`AttrDef::push_change_event`]. See the methods below for what a device pushes most.
471    ///
472    /// A reading must have a read value, so an attribute that cannot be read pushes none,
473    /// and be of a valid quality for the state and the status. A reading with no write value
474    /// is given the set point of its attribute, and one with a write value makes it the set
475    /// point, the way a read does, so that an event and a read of a writable attribute
476    /// carry the same, as with cppTango.
477    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    /// Push a change event for the attribute of that name, a value of a valid quality read
490    /// now, see [`Self::push_attr_event`].
491    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    /// Push an archive event for the attribute of that name, see [`Self::push_change_event`].
496    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    /// Push a user event for the attribute of that name, see [`Self::push_change_event`].
501    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    /// Push an alarm event for the attribute of that name, see [`Self::push_change_event`].
506    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    /// Push that the attribute of that name is ready to be read, with a counter of the
511    /// device's own telling which of its readings it is, see [`AttrReady`]. This fails if
512    /// the device has no such attribute, or does not declare it announced, see
513    /// [`AttrDef::push_data_ready_event`].
514    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    /// Initialize this device: delete its model, if it has one, and build a new one with
524    /// the init function of its class. A failure is kept in place of the model, reported as
525    /// FAULT and in the status.
526    /// 
527    /// This runs when a synchronization picks the device up, and on every `Init` command.
528    /// 
529    /// The model is taken exclusively for the whole initialization, so that a request
530    /// arriving in the middle of it waits for it rather than seeing a half built device.
531    /// This is also why the init function of a class must not reach the model of the very
532    /// device it is building: it is that lock it would be asking for.
533    /// 
534    /// The attributes are configured from the database around the init function: those of
535    /// the class before it, so that it finds them configured, and those it added after it,
536    /// see [`Self::load_all_attr_config`].
537    pub(super) async fn init(&self) {
538
539        let def = self.class.def();
540        let mut model = self.model.write().await;
541
542        // A device withdrawn meanwhile had its model deleted for good, which is never
543        // built again: the Init command that asked for it is told so, see
544        // [`Self::check_model`].
545        if self.is_withdrawn() {
546            return;
547        }
548
549        // An initialization is a delete-then-init, which is what the Init command of a
550        // device is: a device that was running is deleted first, and a device whose
551        // previous initialization failed has only that error to drop.
552        self.delete_model(&mut model).await;
553
554        // The interface goes with the model it was built for: whatever the previous
555        // initialization added to this device alone is gone, and the next one is free to
556        // add it again, possibly differently.
557        //
558        // What the class defines is put back, so that the device starts again from the
559        // interface every device of its class has and its initialization only adds to it,
560        // see [`Self::add_attr`].
561        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        // We re-load attribute config afterward in case new attributes have been added,
570        // in which case only their config will be loaded. The memorized values are then
571        // written back, which needs the model.
572        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        // Initialize, we just log the result but keep the error for state/status.
580        *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        // Whatever the initialization gave, what the device has is polled, a device that
592        // failed recording the failure of every poll, as cppTango does.
593        self.setup_polling().await;
594
595    }
596
597    /// Give the polling of the server what this device is polled for, and the histories of
598    /// its attributes and commands their depth, from what the database holds for it and
599    /// what its definitions ask for, see [`super::polling`] for which has the last word.
600    ///
601    /// A property that cannot be read is taken as absent, and an entry of it that names
602    /// an object the device does not have, or one that cannot be polled, is left out, both
603    /// with a warning: the device is polled for the rest all the same.
604    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        // The depths of the histories, of every attribute and command whether polled or
645        // not, a device being free to record in them itself.
646        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        // The State and the Status have the history of their attribute, which their
665        // command only gives a depth of its own when the property names it.
666        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        // The periods, the database first, then the definitions for what it does not list.
675        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    /// Poll those attributes, see [`AnyDev::poll`]: read at once, as a client reading
717    /// them together does, see [`BackendDev::read_attr`], each reading recorded in the
718    /// history of its attribute and sent as a periodic event, see [`Self::record_poll`].
719    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        // A device withdrawn meanwhile is not polled anymore, see [`Self::is_withdrawn`].
726        if self.is_withdrawn() {
727            return Box::pin(ready(Ok(last_periodics)));
728        }
729
730        // Each attribute along with the index of its item.
731        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            // A request that failed as a whole, the device having no model for instance, is
753            // the failure of every reading.
754            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    /// Record that reading of that attribute polled in its history, valid for that long,
772    /// and send it as a periodic event once the period of that event has passed since the
773    /// last one was, see [`crate::dev::AttrPeriodicEventConfig`], giving back when the
774    /// last one was sent. A read that failed is both recorded and sent as the error it
775    /// failed with.
776    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    /// Poll those commands, see [`AnyDev::poll`]: executed together, with no argument,
801    /// see [`Self::exec_many_command`], each result recorded in the history of its command.
802    /// A command sends no periodic event, so none was ever sent.
803    fn poll_commands(self: Arc<Self>, items: &[PollItem]) -> BoxFuture<'static, Result<Vec<Option<Instant>>>> {
804
805        let none = vec![None; items.len()];
806
807        // A device withdrawn meanwhile is not polled anymore, see [`Self::is_withdrawn`].
808        if self.is_withdrawn() {
809            return Box::pin(ready(Ok(none)));
810        }
811
812        // Each command along with how long what it gives is valid for.
813        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            // A request that failed as a whole is the failure of every command.
835            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    /// Delete the model of this device, if it has one, giving it back to the delete
855    /// function of its class so that it may release what it owns, taking the model
856    /// already locked so that [`Self::init`] can delete and initialize under a single
857    /// lock.
858    /// 
859    /// The device itself keeps existing, with no model, until it is initialized again:
860    /// every request needing the model fails until then, but its state and its status
861    /// still answer, saying why it has none, see [`Self::no_model_answer`].
862    async fn delete_model(&self, model: &mut Option<Result<M>>) {
863
864        // Only a model that was actually built is deleted: the error of a failed
865        // initialization is dropped, there being nothing to give back.
866        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    /// Give this device the interface of its class and nothing else, dropping whatever it
882    /// had added to its own: what a device starts every initialization from, and therefore
883    /// what it has no part of until it is initialized once.
884    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    /// Configure every attribute of this device that is not yet from what the database holds
907    /// for its class and for this device, both asked at once, see [`Attr::load_config`].
908    /// 
909    /// A request that fails does not fail the device: it is kept as the error of every
910    /// attribute it was for, which a client finds out reading any of them.
911    async fn load_all_attr_config(&self) {
912
913        // Find all attributes that are not yet loaded from the database, the State and the
914        // Status included.
915        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    /// Configure those attributes from what the database holds for them, see
930    /// [`Self::load_all_attr_config`].
931    async fn load_given_attr_config(&self, attrs: &[Arc<Attr<M>>]) {
932
933        if attrs.is_empty() {
934            return;
935        }
936
937        // Now we collect the names of the attributes.
938        let names = attrs
939            .iter()
940            .map(|attr| attr.name())
941            .collect::<Vec<_>>();
942
943        // Waiting on both class attr properties and dev attr properties.
944        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    /// What a device that has no model answers for an attribute or a command of that role,
967    /// given the error its initialization failed with, if it failed at all: the State and
968    /// the Status say what the device is, the FAULT of a failed initialization along with
969    /// the error it failed with, which is what keeping that error is for, or the UNKNOWN of
970    /// a device that has not been initialized yet along with saying so. Everything else
971    /// fails with why the device has no model, see [`init_err`].
972    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    /// The state of the device derived from its model, which is already borrowed with the
983    /// access [`Self::state_access`] asked for, which is how its State attribute is read,
984    /// see [`Attr::read`], so that a request reading the state among other attributes
985    /// derives it under the same borrow as the rest of them.
986    /// 
987    /// The state the state function gives is ALARM when an attribute is in alarm, see
988    /// [`Self::set_auto_alarm_states`].
989    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    /// The counterpart of [`Self::model_state`] for the status of the device, which tells
999    /// every attribute in alarm after what the status function gives.
1000    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    /// The state the state function of the class gives, see [`Self::model_state`], called
1031    /// once for the whole request, see [`ReadCtx::state`].
1032    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    /// The attributes in alarm, with the alarm each is in, when the state function gave
1052    /// a state that alarms put the device in ALARM in, see [`Self::set_auto_alarm_states`].
1053    /// Only the first is looked for unless every one is asked for.
1054    /// 
1055    /// An attribute the request already read is not read again, see [`AlarmAttr::read`].
1056    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        // Taken out of the context while its attributes are read with it: none of them is
1064        // the State nor the Status, so none of their reads looks at it.
1065        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                    // The attribute hasn't been read in the request, so we read it here
1073                    // just to know its alarm state, we ignore the value! Like cppTango
1074                    // does for a polled attribute, its latest reading recorded is used
1075                    // when there is one, whatever the source of the request.
1076                    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    /// The access deriving the state of the device needs to its model, which reads the
1101    /// attributes that may put it in ALARM as well.
1102    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    /// The access deriving the status of the device needs to its model, which derives its
1111    /// state as well when the attributes may put it in ALARM.
1112    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    /// Whether initializing this device may block, its init function or its delete
1121    /// function being synchronous, see [`pool::run`].
1122    /// 
1123    /// The memorized values the initialization writes back count too, though only those of
1124    /// the attributes of the class are known beforehand.
1125    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    /// Whether deleting the model of this device may block, its delete function being
1134    /// synchronous, see [`pool::run`].
1135    fn delete_may_block(&self) -> bool {
1136        matches!(self.class.def().delete, Some(DeleteFn::Sync(_)))
1137    }
1138
1139    /// Whether deriving the state of this device may block, see [`pool::run`], reading
1140    /// the attributes that may put it in ALARM included.
1141    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    /// Whether deriving the status of this device may block, see [`pool::run`], deriving
1147    /// its state included when the attributes may put it in ALARM.
1148    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    /// Lock the model of this device with the given access, whether it has one or not.
1154    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    /// Lock the model of this device with the given access, which is how every request
1162    /// that needs the model reaches it.
1163    /// 
1164    /// A device that has no model fails instead: with the error of its failed
1165    /// initialization, so that the caller is told the actual cause, or with
1166    /// [`ErrorKind::DevNotConnected`] when it has not been initialized yet.
1167    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    /// Fail if this device has no model to serve a request with, with the error saying
1177    /// why: what a request that does not itself need the model answers with.
1178    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    /// The command of that name, see [`Self::get_command`], and the error a client is
1187    /// answered with when the device has no command of that name.
1188    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    /// The counterpart of [`Self::find_command`] for the attributes.
1195    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    /// The State attribute of this device, kept out of its interface, which reads the state
1202    /// through the state function of its class, see [`Self::model_state`].
1203    fn state_attr(&self) -> Arc<Attr<M>> {
1204        Arc::clone(&self.mutable.lock().unwrap().state_attr)
1205    }
1206
1207    /// The Status attribute of this device, see [`Self::state_attr`].
1208    fn status_attr(&self) -> Arc<Attr<M>> {
1209        Arc::clone(&self.mutable.lock().unwrap().status_attr)
1210    }
1211
1212    /// The attributes of the given names, or every one when none is given, resolved before
1213    /// the model is borrowed, since resolving one locks the interface store. A name that
1214    /// resolves to nothing only fails that one attribute: the others are still answered.
1215    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    /// The counterpart of [`Self::attr_targets`] for the commands.
1228    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    /// Execute a command, resolved beforehand, see [`BackendDev::command`]: under its own
1241    /// borrow of the model, as [`Self::exec_many_command`] executes several, a device that
1242    /// has no model answering the same, see [`Self::no_model_answer`]. The Init is executed
1243    /// apart, since it rebuilds the model rather than borrowing it, see [`Self::init`].
1244    async fn exec_command(&self, target: Result<Arc<Command<M>>>, input: Value, mut ctx: ReadCtx<M>, source: DevSource) -> Result<Value> {
1245
1246        // A device whose initialization failed only has what its class defines, so a name
1247        // it does not have may well be one it would have had: its own failure is what a
1248        // client needs to hear then, not that the name is unknown.
1249        let command = match target {
1250            Ok(command) => command,
1251            Err(err) => return Err(self.check_model().await.err().unwrap_or(err)),
1252        };
1253
1254        // The Init is never answered from the cache, and an initialization that failed is a
1255        // failed Init: the device keeps that error to report it in its status, and the
1256        // client that asked for the initialization is given it too.
1257        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    /// Execute commands, resolved beforehand, each with its argument, under a single borrow
1279    /// of the model, exclusive as soon as one of the execution functions asks for it: the
1280    /// counterpart for commands of [`Self::exec_read_attr`], which is how the commands
1281    /// polled together are executed, see [`Self::poll_commands`]. A single command is
1282    /// executed the same way, without the batch, see [`Self::exec_command`].
1283    ///
1284    /// The Init cannot be among them, see [`Self::exec_command`].
1285    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        // A device that has no model answers each command on its own, see
1294        // [`Self::no_model_answer`], as a multiple read does.
1295        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    /// Read attributes, resolved beforehand, see [`BackendDev::read_attr`].
1322    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        // They are all read under a single borrow of the model, which is what makes the
1325        // values of a multiple read consistent with each other, and that borrow is
1326        // exclusive as soon as one of the read functions asks for it.
1327        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        // A device that has no model answers each attribute on its own, see
1334        // [`Self::no_model_answer`]: a client reading the State of a broken device is told
1335        // it is broken, and why, not that the request failed.
1336        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        // The attributes are read first and the state and the status derived after them, so
1346        // that an attribute that may put the device in ALARM is read once, its reading being
1347        // what the state is derived from, see [`ReadCtx::record`]. Each is answered where it
1348        // was asked for all the same.
1349        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    /// Write attributes, resolved beforehand, see [`BackendDev::write_attr`].
1374    async fn exec_write_attr(&self, targets: Vec<(Result<Arc<Attr<M>>>, Value)>) -> Result<Vec<Result<()>>> {
1375
1376        // Written under a single borrow of the model, so that a client writing several
1377        // attributes at once has them written together.
1378        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    /// Change the configuration of attributes, each given as the whole configuration it is
1398    /// to have, see [`BackendDev::set_attr_config`].
1399    /// 
1400    /// Every attribute is changed on its own, one failing leaving the others to be changed:
1401    /// on the device first, and then in the database, the way cppTango does.
1402    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    /// Change the configuration of one attribute, see [`Self::exec_set_attr_config`].
1411    /// 
1412    /// A change the database fails to store is the one the device has all the same, until
1413    /// it is configured from the database again, and is therefore pushed all the same.
1414    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    /// Write then read back attributes, resolved beforehand, see
1448    /// [`BackendDev::write_read_attr`].
1449    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        // Every value is written and read back under a single borrow of the model, which is
1452        // the whole point of asking for the two together: what a client is given is what the
1453        // device had right after its write, with nothing of anyone else in between.
1454        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                // A write that failed is not read back: what a client gets of that
1465                // attribute is the reason its write was refused.
1466                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            // Out of reach of the clients of this process first, as of the others next.
1507            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        /// How many of the latest results the polling status tells the times of.
1554        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
1607/// The first value of that property parsed as a number, absent if it has none, or with a
1608/// warning if it is not one.
1609fn 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
1626/// The interface of a single device, which every device owns its own copy of because it
1627/// can be altered device by device.
1628struct DevMutable<M: Model> {
1629    /// The black box of the device where past commands, attribute reads/writes and other
1630    /// accesses to this device.
1631    black_box: VecDeque<String>,
1632    /// The commands the device currently registers.
1633    commands: HashMap<ArcName, Arc<Command<M>>>,
1634    attrs: HashMap<ArcName, Arc<Attr<M>>>,
1635    /// The State and the Status of the device, kept out of the interface since every device
1636    /// has them whatever its class, see [`Dev::state_attr`].
1637    state_attr: Arc<Attr<M>>,
1638    status_attr: Arc<Attr<M>>,
1639    /// The states in which the device is put in ALARM when one of its attributes is in
1640    /// alarm, see [`Dev::set_auto_alarm_states`].
1641    auto_alarm_states: Arc<[DevState]>,
1642}
1643
1644/// What a request reading a device learns of it as it goes, so that it reads nothing twice:
1645/// what may put the device in ALARM, along with what the request read of it, and the state
1646/// the state function gave, see [`Dev::set_auto_alarm_states`].
1647pub(super) struct ReadCtx<M: Model> {
1648    /// What may put the device in ALARM, which only a request involving the state or the
1649    /// status takes, and only when anything may, see [`ReadCtx::new`].
1650    alarm: Option<AlarmCtx<M>>,
1651    /// The state the state function gave, once called, so that the state and the status
1652    /// asked for together call it once.
1653    state: Option<DevState>,
1654}
1655
1656/// What may put a device in ALARM, taken out of the interface store once for a whole
1657/// request, so that what the request borrows the model for and what it reads are the same:
1658/// the states in which the device may be put in ALARM, and the attributes that have any
1659/// alarm configured, by name.
1660struct AlarmCtx<M: Model> {
1661    states: Arc<[DevState]>,
1662    attrs: Vec<AlarmAttr<M>>,
1663}
1664
1665/// An attribute that may put a device in ALARM, see [`AlarmCtx`].
1666struct AlarmAttr<M: Model> {
1667    attr: Arc<Attr<M>>,
1668    /// What the request found reading the attribute, so that it is read only once:
1669    /// - `None`: not read yet;
1670    /// - `Some(None)`: read, and in no alarm, or could not be read;
1671    /// - `Some(Some(alarm))`: read, and in that alarm.
1672    read: Option<Option<AttrAlarm>>,
1673}
1674
1675impl<M: Model> ReadCtx<M> {
1676
1677    /// The context of a request on that device, which takes what may put the device in
1678    /// ALARM out of its interface store only if the request involves its state or its
1679    /// status, and only keeps it if the device has states to be put in ALARM in and
1680    /// attributes that have any alarm configured.
1681    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    /// Whether anything may put the device in ALARM, in any state.
1718    fn has_alarm(&self) -> bool {
1719        self.alarm.is_some()
1720    }
1721
1722    /// Whether anything may put the device in ALARM when the state function gave that
1723    /// state, see [`Dev::set_auto_alarm_states`].
1724    fn may_alarm(&self, state: DevState) -> bool {
1725        self.alarm.as_ref().is_some_and(|alarm| alarm.states.contains(&state))
1726    }
1727
1728    /// Keep the alarm the request found that attribute in, if it may put the device in
1729    /// ALARM, so that it is not read again.
1730    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    /// The access reading the attributes that may put the device in ALARM needs to the
1743    /// model.
1744    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    /// Whether reading the attributes that may put the device in ALARM may block, see
1751    /// [`pool::run`].
1752    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
1760/// A reading of a value of a valid quality, made now, which is what a device pushes most.
1761fn 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/// The access a request needs to the model of a device: shared when every handler it runs
1771/// takes the model by reference, so that such requests run in parallel, and exclusive as
1772/// soon as one of them takes it mutably.
1773#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1774pub enum Access {
1775    Shared,
1776    Exclusive,
1777}
1778
1779impl Access {
1780
1781    /// The access covering both, which is exclusive as soon as either is: a request
1782    /// touching several attributes at once borrows the model once, for what the most
1783    /// demanding of its handlers needs.
1784    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
1793/// The lock of the model of a device, held for the duration of a request with the access
1794/// that request asked for.
1795pub 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    /// The model of the device if it has one, or the error of its failed initialization.
1803    #[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    /// Why the device has no model, if it has none: the error its initialization failed
1812    /// with, if it failed at all, see [`Dev::no_model_answer`].
1813    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    /// The model, borrowed with the access the lock was taken with.
1822    /// 
1823    /// # Panics
1824    /// 
1825    /// If the device has no model, which [`Dev::lock_model`] checks beforehand.
1826    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
1841/// The model of a device, borrowed for the duration of a request with the access that
1842/// request asked for, which is what lets every handler of a single request run under one
1843/// borrow of the model.
1844pub enum ModelRef<'m, M> {
1845    Shared(&'m M),
1846    Exclusive(&'m mut M),
1847}
1848
1849impl<M> ModelRef<'_, M> {
1850
1851    /// The model, shared, which either access gives.
1852    #[inline]
1853    pub fn shared(&self) -> &M {
1854        match self {
1855            Self::Shared(model) => model,
1856            Self::Exclusive(model) => model,
1857        }
1858    }
1859
1860    /// The model, exclusively, which only an exclusive access gives.
1861    /// 
1862    /// Failing here is a bug of the server and not of the device: the access is computed
1863    /// from the very handlers the request then runs, see [`Access::merge`], so a handler
1864    /// asking for the model mutably has been counted before it is called.
1865    #[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
1877/// The status of a device whose initialization failed: the error it failed with, which is
1878/// what a client reads to know why the device is not running.
1879fn make_failed_status(err: &Error) -> Scalar<str> {
1880    Scalar::from(format!("The initialization of the device failed:\n{err}").as_str())
1881}
1882
1883/// The error every request on a device that has no model fails with, given the error
1884/// its initialization failed with, if it failed at all.
1885fn 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
1895/// The documentation of a definition as a client is told it, an empty one being none.
1896pub fn some_doc(doc: &Cow<'static, str>) -> Option<Cow<'static, str>> {
1897    (!doc.is_empty()).then(|| doc.clone())
1898}
1899
1900// =================== //
1901//   Device backend    //
1902// =================== //
1903
1904/// The backend of a device, which is the device itself seen through a transparent wrapper,
1905/// so that the backend methods are not methods of [`Dev`] itself. The backend serving a
1906/// device holds it as such, and it is reached directly by a proxy of this process when the
1907/// server is local, see [`crate::local`].
1908#[repr(transparent)]
1909struct DevImpl<M: Model>(Dev<M>);
1910
1911impl<M: Model> DevImpl<M> {
1912
1913    /// See a device as its backend.
1914    #[inline]
1915    fn from_arc(dev: Arc<Dev<M>>) -> Arc<Self> {
1916        // SAFETY: DevImpl is a repr(transparent) wrapper around Dev, so both have the same
1917        // layout and an Arc of one is an Arc of the other.
1918        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
1930/// What touches the model is run on the pool of the server when one of the handlers it
1931/// runs may block, and polled by the executor of the backend otherwise, see [`pool::run`];
1932/// what only reads the definitions and the interface is answered on the call itself.
1933///
1934/// Every request is refused once the device is withdrawn, see [`Dev::is_withdrawn`], which
1935/// a client holding the device may still be asking it. A request withdrawn meanwhile finds
1936/// no model once it holds its lock, the withdrawal deleting it under that lock, and is
1937/// answered as a device that is not initialized.
1938impl<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    /// None of this is the device's own: the class it is an instance of, and the server
1971    /// running it, are what a client is told here, so all of it is read off the definition
1972    /// of its class and off the server it belongs to.
1973    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            // The description of the class, this crate having the one string where Tango
1993            // has two: what the device *is* and what it says of itself are the same here.
1994            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    /// The definitions and nothing else, so this borrows no model: a device that failed to
2133    /// initialize still describes the interface of its class, which is how a client knows
2134    /// to ask it for the state and the status that say it failed.
2135    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    /// The counterpart of [`Self::command_info`], and the same absence of any model.
2153    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    /// Change the configuration of each attribute on its own, see [`Dev::exec_set_attr_config`].
2171    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    /// What the device recorded itself, or its polling recorded, in the history of that
2182    /// command, see [`Command::history`]: a command nothing was recorded of fails rather
2183    /// than giving an empty history, whether it is polled or not.
2184    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    /// The counterpart of [`Self::command_history`] for the attributes, see [`Attr::history`].
2203    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    // The subscriptions are the server's, which keeps them for as long as the device is
2222    // served under that name, see [`Server::events`]. Only an attribute the device has can
2223    // be subscribed to, the state and the status included, as cppTango allows, and only to
2224    // an event something sends, see [`Attr::check_push_event`].
2225
2226    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
2266/// What a name a request gave resolved to, which names itself, or the name itself when it
2267/// resolved to nothing, along with the reason: the very name the request gave then.
2268type 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
2274/// Every name a request gave, each a scalar of its own sharing them.
2275fn 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    /// A server with a single device of that class, which is added but not synchronized
2300    /// yet, on a database listing no other device, the server being served by the context.
2301    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        // A writable attribute is pushed with its set point, as it is read.
2332        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        // A reading with no read value, or a State of another quality than valid, is refused.
2341        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        // An event the attribute does not declare it pushes is refused.
2348        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        // Nothing changed, nothing pushed.
2371        dev.0.set_attr_config("Count", &config).unwrap();
2372        assert!(conf.next().now_or_never().is_none());
2373
2374    }
2375
2376    /// A read from the cache is answered with the latest reading recorded while it is
2377    /// valid, the State command shares the history of its attribute, and a history is
2378    /// given oldest first.
2379    #[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        // An outdated reading is read again from the device, or refused from the cache alone.
2406        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        // The State is only recorded valid, and its command answers from that history.
2414        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    /// A command is answered from the cache the way an attribute is read from it, and only
2430    /// one taking no argument is recorded.
2431    #[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        // An outdated result is executed again, or refused from the cache alone.
2455        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        // The State command records in its attribute.
2465        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    /// A motor whose initialization always fails, which is served with no model.
2472    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    /// A device that has no model answers each attribute of a read on its own: the State
2481    /// and the Status saying what it is, the others failing with why it has no model.
2482    #[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        // Not initialized yet.
2490        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    /// Commands executed together under a single borrow, each answered in its order, as a
2505    /// multiple read is, a device that has no model included.
2506    #[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    /// Every asynchronous handler, from the initialization to the state, run as a client
2599    /// asks for it.
2600    #[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        // The Init command builds the model again.
2616        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    /// A name that resolves to nothing is answered with the very name the request gave,
2622    /// sharing its octets, and a known one with the name of its definition.
2623    #[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    /// A request runs on the pool as soon as one of its handlers is synchronous, the
2641    /// asynchronous ones being blocked on there, and is polled where it is awaited
2642    /// otherwise.
2643    #[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    /// The model stays locked across the awaits of an asynchronous handler: a request
2661    /// arriving meanwhile waits for it.
2662    #[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    /// A device whose asynchronous initialization failed keeps the error, and gives it
2693    /// instead of running anything.
2694    #[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}