Skip to main content

tango/
dev.rs

1//! The client side: [`DevProxy`], the handle to a device, and what it reads and writes.
2
3use std::sync::{Arc, OnceLock, RwLock};
4use std::time::{Duration, SystemTime};
5use std::fmt::{self, Debug, Display};
6use std::task::{Context, Poll};
7use std::hash::{Hash, Hasher};
8use std::future::Future;
9use std::cmp::Ordering;
10use std::str::FromStr;
11use std::borrow::Cow;
12use std::{env, mem};
13use std::pin::Pin;
14
15use futures::stream::{BoxStream, Stream, StreamExt};
16use futures::ready;
17use tracing::{debug, instrument, warn};
18use futures::future::{poll_fn, BoxFuture};
19
20use crate::value::{AttrDataFormat, AttrDataKind, CommandDataKind, FromValue, IntoValue, Scalar, Spectrum, Value};
21use crate::database::{prop_params, set_prop_params, DatabaseProxy, GetPropSpec, PropCommands, SetPropSpec};
22use crate::backend::{default_backend, Backend, BackendDev};
23use crate::error::{Error, ErrorKind, Result, reason};
24use crate::name::{ArcTrl, IntoTrl, TrlObject};
25use crate::future::block_on_timeout;
26use crate::local;
27
28
29/// The timeout of a new proxy, the same as cppTango's, see [`DevProxy::set_timeout`].
30const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);
31
32/// The statically known device name for the database device in its server.
33pub(crate) const DATABASE_DEV_NAME: &str = "database";
34
35/// A device proxy is an object that allows querying a remote device.
36/// 
37/// Two device proxies are equal, and hash the same, when they use the same backend (compared
38/// by pointer) and have the same internal target, regardless of their `trl`, their connection
39/// or their configuration.
40#[derive(Debug)]
41pub struct DevProxy {
42    /// The inner structure, boxed to reduce exposed size.
43    inner: Box<ProxyInner>,
44    /// The state of this proxy, behind a read-write lock.
45    state: RwLock<ProxyState>,
46}
47
48#[derive(Debug, Clone)]
49struct ProxyInner {
50    // The backend, shared with a reference-counted pointer.
51    backend: Arc<dyn Backend>,
52    /// The source configured for this proxy.
53    source: DevSource,
54    /// The timeout configured for this proxy.
55    timeout: Duration,
56    /// The real target for connection of this proxy.
57    target: ProxyTarget,
58}
59
60#[derive(Debug, Clone)]
61struct ProxyState {
62    /// The actual backend in use, none if the last call errored.
63    backend: Option<Arc<dyn BackendDev>>,
64    /// In case the target is directly a database, this contains the host and port of the
65    /// database connected to, this is used to construct the TRL.
66    database_host_and_port: Option<(Arc<str>, u16)>,
67}
68
69/// Target for a device proxy, this is used so that the DevProxy can also be responsible
70/// for connecting to a database as well as regular devices. The device name might be an
71/// alias!
72#[derive(Debug, Clone, PartialEq, Eq, Hash)]
73pub(crate) enum ProxyTarget {
74    /// Database device from environment.
75    DatabaseFromEnv,
76    /// Database device from host and port.
77    DatabaseAt {
78        host: Arc<str>,
79        port: u16,
80    },
81    /// Regular device from a database.
82    RegularFromDatabase {
83        database: DatabaseProxy,
84        name: Arc<str>,
85    },
86    /// A regular device accessible directly from a given service on a host, no database
87    /// is involved!
88    RegularAt {
89        host: Arc<str>,
90        port: u16,
91        name: Arc<str>,
92    },
93}
94
95impl DevProxy {
96
97    /// Same as [`Self::try_new`] but panics if the TRL is invalid for a device proxy.
98    pub fn new<'a>(trl: impl IntoTrl<'a>) -> Self {
99        Self::new_with_backend(default_backend(), trl)
100    }
101
102    /// Create a new device proxy located by the given Tango Resource Locator, the given
103    /// TRL should not target an attribute or property, or else this function will return
104    /// None.
105    pub fn try_new<'a>(trl: impl IntoTrl<'a>) -> Option<Self> {
106        Self::try_new_with_backend(default_backend(), trl)
107    }
108
109    /// Create a new device proxy from a database and the device's name.
110    pub fn new_from_database(database: &DatabaseProxy, name: &str) -> Self {
111        Self::new_from_database_with_backend(default_backend(), database, name)
112    }
113
114    /// Same as [`Self::try_new_with_backend`] but panics if the TRL is invalid for a device
115    /// proxy.
116    pub fn new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Self {
117        Self::try_new_with_backend(backend, trl).expect("The given TRL does not target a device")
118    }
119
120    /// Create a new device proxy located by the given Tango Resource Locator, the given
121    /// TRL should not target an attribute or property, or else this function will return
122    /// None.
123    pub fn try_new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Option<Self> {
124
125        let trl = trl.into_trl().ok()?;
126
127        // We can't target a property!
128        if trl.property().is_some() {
129            return None;
130        }
131
132        let name = match trl.object() {
133            TrlObject::Alias(name) => name,
134            TrlObject::Dev(name) => name,
135            TrlObject::Attr(_, _) => return None,
136        };
137
138        let target = match (trl.db(), trl.host_and_port()) {
139            (true, Some((host, port))) => ProxyTarget::RegularFromDatabase { 
140                database: DatabaseProxy::new_with_backend(Arc::clone(&backend), host, port),
141                name: Arc::from(name),
142            },
143            (true, None) => ProxyTarget::RegularFromDatabase { 
144                database: DatabaseProxy::new_from_env_with_backend(Arc::clone(&backend)), 
145                name: Arc::from(name),
146            },
147            (false, Some((host, port))) => ProxyTarget::RegularAt { 
148                host: Arc::from(host), 
149                port, 
150                name: Arc::from(name),
151            },
152            (false, None) => return None,
153        };
154
155        Some(Self::new_from_target_with_backend(backend, target))
156        
157    }
158
159    /// Create a new device proxy from a database and the device's name, with backend.
160    pub fn new_from_database_with_backend(backend: Arc<dyn Backend>, database: &DatabaseProxy, name: &str) -> Self {
161        Self::new_from_target_with_backend(backend, ProxyTarget::RegularFromDatabase { 
162            database: database.clone(), 
163            name: Arc::from(name),
164        })
165    }
166
167    /// Internal function to create a device proxy for the given target, with backend.
168    pub(crate) fn new_from_target_with_backend(backend: Arc<dyn Backend>, target: ProxyTarget) -> Self {
169        Self {
170            inner: Box::new(ProxyInner {
171                backend,
172                target,
173                source: DevSource::CacheDev,
174                timeout: DEFAULT_TIMEOUT,
175            }),
176            state: RwLock::new(ProxyState {
177                backend: None,
178                database_host_and_port: None,
179            }),
180        }
181    }
182
183    pub fn backend(&self) -> &Arc<dyn Backend> {
184        &self.inner.backend
185    }
186
187    /// Retrieve the timeout duration currently used for all requests.
188    pub fn timeout(&self) -> Duration {
189        self.inner.timeout
190    }
191
192    /// Set the timeout duration and return the previous one. See [`Self::timeout`].
193    pub fn set_timeout(&mut self, timeout: Duration) -> Duration {
194        mem::replace(&mut self.inner.timeout, timeout)
195    }
196
197    /// Retrieve the source of data currently used for attribute readings and commands.
198    pub fn source(&self) -> DevSource {
199        self.inner.source
200    }
201
202    /// Set the source of data and return the previous one. See [`Self::source`].
203    pub fn set_source(&mut self, source: DevSource) -> DevSource {
204        mem::replace(&mut self.inner.source, source)
205    }
206
207    /// Internal function to initialize a brand new connection to this device proxy.
208    /// 
209    /// Boxed, because connecting a device goes through the database, itself a device
210    /// connected by this very function.
211    fn connect_backend(&self) -> BoxFuture<'_, Result<(Arc<dyn BackendDev>, Option<(Arc<str>, u16)>)>> {
212        Box::pin(async move {
213
214            match self.inner.target {
215                ProxyTarget::DatabaseFromEnv => {
216
217                    debug!("Connecting to the database of TANGO_HOST");
218
219                    let databases = env_databases();
220                    if databases.is_empty() {
221                        return Err(match env::var("TANGO_HOST") {
222                            Ok(var) => Error::new(ErrorKind::TangoHostNotSet)
223                                .with_description("TANGO_HOST is not a comma separated list of host:port")
224                                .with_append_description(format_args!("Got: {var:?}")),
225                            Err(_) => Error::new(ErrorKind::TangoHostNotSet)
226                                .with_description("TANGO_HOST is not defined in the environment"),
227                        });
228                    }
229
230                    let mut last_err = None;
231                    for (host, port) in databases {
232
233                        if let Some(dev_backend) = local::find_at(host, *port, DATABASE_DEV_NAME) {
234                            debug!(%host, port, "Connected to the database, served by this process");
235                            return Ok((dev_backend, Some((host.clone(), *port))));
236                        }
237
238                        match self.inner.backend.connect_at(host, *port, DATABASE_DEV_NAME).await {
239                            Ok(dev_backend) => {
240                                debug!(%host, port, "Connected to the database");
241                                return Ok((dev_backend, Some((host.clone(), *port))));
242                            }
243                            Err(err) => {
244                                warn!(%host, port, %err, "Failed to connect to the database, trying the next one");
245                                last_err = Some(err);
246                            }
247                        }
248                    }
249
250                    // Every database in TANGO_HOST has been tried, the last failure being the
251                    // one kept as the cause: naming them all is what tells the reader that the
252                    // other ones were tried too.
253                    let mut err = Error::new(ErrorKind::DatabaseNotConnected)
254                        .with_description("Failed to connect to any database")
255                        .with_append_description(format_args!("Tried, from TANGO_HOST: {}", DisplayHostAndPorts(databases)));
256                    if let Some(last_err) = last_err {
257                        err.set_parent(last_err);
258                    }
259                    Err(err)
260
261                }
262                ProxyTarget::DatabaseAt { ref host, port } => {
263                
264                    debug!(%host, port, "Connecting to the database");
265
266                    if let Some(dev_backend) = local::find_at(host, port, DATABASE_DEV_NAME) {
267                        debug!(%host, port, "Connected to the database, served by this process");
268                        return Ok((dev_backend, Some((host.clone(), port))));
269                    }
270
271                    match self.inner.backend.connect_at(host, port, DATABASE_DEV_NAME).await {
272                        Ok(dev_backend) => {
273                            debug!(%host, port, "Connected to the database");
274                            Ok((dev_backend, Some((host.clone(), port))))
275                        }
276                        Err(err) => Err(Error::new(ErrorKind::DatabaseNotConnected)
277                            .with_description("Failed to connect to the database")
278                            .with_append_description(format_args!("Tried: {host}:{port}"))
279                            .with_parent(err)),
280                    }
281
282                }
283                ProxyTarget::RegularFromDatabase { ref database, ref name } => {
284
285                    debug!(dev = %name, "Connecting to the device, through the database");
286                    
287                    let info = match database.import_dev_async(name).await {
288                        Ok(info) if !info.exported =>
289                            return Err(Error::new(ErrorKind::DevNotExported)
290                                .with_description(format!("Device {name:?} is defined in the database but its server {:?} does not export it",
291                                    info.server))),
292                        Ok(info) => info,
293                        Err(err) if let ErrorKind::DevNotConnected = err.kind() =>
294                            return Err(Error::new(ErrorKind::DatabaseNotConnected)
295                                .with_description(format!("Failed to import device {name:?} from the database"))
296                                .with_parent(err)),
297                        Err(err) => return Err(err),
298                    };
299
300                    // A device this very process serves is reached directly, whatever the
301                    // backend, the reference telling it apart from any other of that name.
302                    if let Some(dev_backend) = local::find(info.itr.value()) {
303                        debug!(dev = %name, "Connected to the device, served by this process");
304                        return Ok((dev_backend, None));
305                    }
306
307                    // The kind is the one the backend found, 'with_backend' keying the
308                    // reconnection on it, the failure of the backend being the cause.
309                    match self.inner.backend.connect(info.itr.value()).await {
310                        Ok(dev_backend) => {
311                            debug!(dev = %name, "Connected to the device");
312                            Ok((dev_backend, None))
313                        }
314                        Err(err) => Err(context_err(err, format!(
315                            "Failed to connect to device {name:?} of server {:?}, exported on host {:?}",
316                            info.server, info.host))),
317                    }
318
319                }
320                ProxyTarget::RegularAt { ref host, port, ref name } => {
321                
322                    debug!(%host, port, dev = %name, "Connecting to the device");
323
324                    if let Some(dev_backend) = local::find_at(host, port, name) {
325                        debug!(%host, port, dev = %name, "Connected to the device, served by this process");
326                        return Ok((dev_backend, None));
327                    }
328
329                    match self.inner.backend.connect_at(host, port, name).await {
330                        Ok(dev_backend) => {
331                            debug!(%host, port, dev = %name, "Connected to the device");
332                            Ok((dev_backend, None))
333                        }
334                        Err(err) => Err(context_err(err, format!(
335                            "Failed to connect to device {name:?} at {host}:{port}"))),
336                    }
337
338                }
339            }
340
341        })
342    }
343
344    /// Give back the backend this proxy reaches its device through, connecting it if that
345    /// has not happened yet.
346    ///
347    /// The connection is awaited with no lock of the proxy held: a guard cannot travel
348    /// across an await, and keeping one here would hold every other reader of the proxy
349    /// for the whole of a round trip. Two callers landing here at once therefore both
350    /// connect, and the one that stores its backend first is the one that is kept, the
351    /// other connection being dropped on the spot.
352    async fn ensure_backend(&self) -> Result<Arc<dyn BackendDev>> {
353
354        if let Some(backend) = self.state.read().unwrap().backend.as_ref() {
355            return Ok(Arc::clone(backend));
356        }
357
358        let (backend, database_host_and_port) = self.connect_backend().await?;
359
360        let mut state = self.state.write().unwrap();
361        match state.backend {
362            Some(ref backend) => Ok(Arc::clone(backend)),
363            None => {
364                state.backend = Some(Arc::clone(&backend));
365                state.database_host_and_port = database_host_and_port;
366                Ok(backend)
367            }
368        }
369
370    }
371
372    fn reset_backend(&self) {
373        let mut state = self.state.write().unwrap();
374        state.backend = None;
375        state.database_host_and_port = None;
376    }
377
378    fn check_backend<T>(&self, res: Result<T>) -> Result<T> {
379        if let Err(err) = &res && err.kind() == ErrorKind::DevNotConnected {
380            self.reset_backend();
381        }
382        res
383    }
384
385    pub(crate) fn block_on<F: Future>(&self, fut: F) -> Result<F::Output> {
386        block_on_timeout(fut, self.inner.timeout)
387    }
388
389    fn database_host_and_port(&self) -> Result<Option<(Arc<str>, u16)>> {
390        self.block_on(self.database_host_and_port_async())?
391    }
392
393    /// The host and port of the database this proxy reached its device through, connecting
394    /// it if that has not happened yet, see [`Self::database_host_and_port`].
395    pub(crate) async fn database_host_and_port_async(&self) -> Result<Option<(Arc<str>, u16)>> {
396        self.ensure_backend().await?;
397        Ok(self.state.read().unwrap().database_host_and_port.clone())
398    }
399
400    /// Return the [`DatabaseProxy`] this device has been requested from, or this device as a
401    /// database if it is one. A device reached without a database fails with
402    /// [`ErrorKind::DatabaseNeeded`].
403    pub fn database(&self) -> Result<&DatabaseProxy> {
404        match self.inner.target {
405            ProxyTarget::DatabaseFromEnv |
406            ProxyTarget::DatabaseAt { .. } => 
407                Ok(DatabaseProxy::from_dev(self)),
408            ProxyTarget::RegularFromDatabase { ref database, .. } => 
409                Ok(database),
410            ProxyTarget::RegularAt { ref host, port, ref name } => 
411                Err(Error::new(ErrorKind::DatabaseNeeded)
412                    .with_description(format!("Device {name:?} is reached at {host}:{port} rather than through a database"))),
413        }
414    }
415
416    /// Resolve this device's fully qualified device name.
417    #[instrument(skip(self), fields(target = %self.inner.target))]
418    pub fn trl(&self) -> Result<ArcTrl> {
419        match self.inner.target {
420            ProxyTarget::DatabaseFromEnv => {
421                let name = self.name()?;
422                let (host, port) = self.database_host_and_port()?
423                    .expect("This should be defined if the target is a database");
424                Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
425            }
426            ProxyTarget::DatabaseAt { ref host, port } => {
427                let name = self.name()?;
428                Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
429            }
430            ProxyTarget::RegularFromDatabase { ref database, ref name } => {
431                let (host, port) = database.database_host_and_port()?
432                    .expect("This should be defined if the target is a database");
433                Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
434            }
435            ProxyTarget::RegularAt { ref host, port, ref name } => {
436                Ok(format!("tango://{host}:{port}/{name}#dbase=no").parse::<ArcTrl>().unwrap())
437            }
438        }
439    }
440
441    /// Connect this proxy to its device if that has not happened yet, so that its clones
442    /// share that connection instead of each making its own.
443    #[instrument(skip(self), fields(target = %self.inner.target))]
444    pub fn connect(&self) -> Result<()> {
445        self.block_on(self.connect_async())?
446    }
447
448    /// Connect this proxy to its device if that has not happened yet, see [`Self::connect`].
449    pub async fn connect_async(&self) -> Result<()> {
450        self.ensure_backend().await.map(drop)
451    }
452
453    /// Resolve this device's name.
454    #[instrument(skip(self), fields(target = %self.inner.target))]
455    pub fn name(&self) -> Result<Arc<str>> {
456        self.block_on(self.name_async())?
457    }
458
459    /// Resolve this device's name, see [`Self::name`].
460    pub async fn name_async(&self) -> Result<Arc<str>> {
461        let backend = self.ensure_backend().await?;
462        self.check_backend(backend.name().await)
463    }
464
465    /// Resolve this device's admin device name.
466    #[instrument(skip(self), fields(target = %self.inner.target))]
467    pub fn admin_dev_name(&self) -> Result<Arc<str>> {
468        self.block_on(self.admin_dev_name_async())?
469    }
470
471    /// Resolve this device's admin device name, see [`Self::admin_dev_name`].
472    pub async fn admin_dev_name_async(&self) -> Result<Arc<str>> {
473        let backend = self.ensure_backend().await?;
474        self.check_backend(backend.admin_dev_name().await)
475    }
476
477    /// Resolve this device's admin device and return it!
478    #[instrument(skip(self), fields(target = %self.inner.target))]
479    pub fn admin_dev(&self) -> Result<DevProxy> {
480        Ok(self.admin_dev_of(self.admin_dev_name()?))
481    }
482
483    /// Resolve this device's admin device and return it, see [`Self::admin_dev`].
484    pub async fn admin_dev_async(&self) -> Result<DevProxy> {
485        Ok(self.admin_dev_of(self.admin_dev_name_async().await?))
486    }
487
488    /// The proxy to the admin device of this one, from the name of it: asking the device
489    /// for that name is the only part of this that reaches it, so it is what the two
490    /// methods above do and this is everything they share.
491    fn admin_dev_of(&self, name: Arc<str>) -> DevProxy {
492
493        let backend = Arc::clone(&self.inner.backend);
494
495        DevProxy::new_from_target_with_backend(backend, match &self.inner.target {
496            ProxyTarget::DatabaseFromEnv |
497            ProxyTarget::DatabaseAt { .. } => {
498                // Self can be a DatabaseProxy, so we use it!
499                ProxyTarget::RegularFromDatabase { 
500                    database: DatabaseProxy::with_dev(self.clone()), 
501                    name,
502                }
503            }
504            ProxyTarget::RegularFromDatabase { database, .. } => {
505                ProxyTarget::RegularFromDatabase { 
506                    database: database.clone(), 
507                    name,
508                }
509            }
510            ProxyTarget::RegularAt { host, port, .. } => {
511                ProxyTarget::RegularAt { 
512                    host: Arc::clone(host), 
513                    port: *port, 
514                    name,
515                }
516            }
517        })
518
519    }
520
521    /// Get what the device says it is: the class it is an instance of, the server running
522    /// it and where, and the versions of whatever that server is built out of.
523    ///
524    /// All of it is the server's own description of itself and none of it is asked of the
525    /// database, so it answers for a device served without one.
526    #[instrument(skip(self), fields(target = %self.inner.target))]
527    pub fn info(&self) -> Result<DevInfo> {
528        self.block_on(self.info_async())?
529    }
530
531    /// Get what the device says it is, see [`Self::info`].
532    pub async fn info_async(&self) -> Result<DevInfo> {
533        let backend = self.ensure_backend().await?;
534        self.check_backend(backend.info().await)
535    }
536
537    /// Ping this device.
538    #[instrument(skip(self), fields(target = %self.inner.target))]
539    pub fn ping(&self) -> Result<()> {
540        self.block_on(self.ping_async())?
541    }
542
543    /// Ping this device, see [`Self::ping`].
544    pub async fn ping_async(&self) -> Result<()> {
545        let backend = self.ensure_backend().await?;
546        self.check_backend(backend.ping().await)
547    }
548
549    /// Get the device state.
550    #[instrument(skip(self), fields(target = %self.inner.target))]
551    pub fn state(&self) -> Result<DevState> {
552        self.block_on(self.state_async())?
553    }
554
555    /// Get the device state, see [`Self::state`].
556    pub async fn state_async(&self) -> Result<DevState> {
557        let backend = self.ensure_backend().await?;
558        self.check_backend(backend.state().await)
559    }
560
561    /// Get the device status.
562    #[instrument(skip(self), fields(target = %self.inner.target))]
563    pub fn status(&self) -> Result<Scalar<str>> {
564        self.block_on(self.status_async())?
565    }
566
567    /// Get the device status, see [`Self::status`].
568    pub async fn status_async(&self) -> Result<Scalar<str>> {
569        let backend = self.ensure_backend().await?;
570        self.check_backend(backend.status().await)
571    }
572
573    /// Initialize the device again, executing its `Init` command.
574    pub fn init(&self) -> Result<()> {
575        self.command("Init", ())
576    }
577
578    /// Initialize the device again, see [`Self::init`].
579    pub fn init_async(&self) -> impl Future<Output = Result<()>> + Send {
580        self.command_async("Init", ())
581    }
582
583    /// Request the device black box.
584    #[instrument(skip(self), fields(target = %self.inner.target))]
585    pub fn black_box(&self, max: Option<usize>) -> Result<Spectrum<str>> {
586        self.block_on(self.black_box_async(max))?
587    }
588
589    /// Request the device black box, see [`Self::black_box`].
590    pub async fn black_box_async(&self, max: Option<usize>) -> Result<Spectrum<str>> {
591        let backend = self.ensure_backend().await?;
592        self.check_backend(backend.black_box(max).await)
593    }
594
595    /// Execute a remote command, giving back what it answered with, converted into the type
596    /// it is expected to be, see [`FromValue`].
597    ///
598    /// The value shares the octets the device answered with, so asking for a [`Value`] or
599    /// for one of the buffers it holds — a [`crate::value::Spectrum`] of a million points
600    /// above all — copies nothing, where the usual owned types are copied out of it.
601    #[instrument(skip(self, input), fields(target = %self.inner.target))]
602    pub fn command_async<I, O>(&self, name: &str, input: I) -> impl Future<Output = Result<O>> + Send
603    where
604        I: IntoValue,
605        O: FromValue,
606    {
607
608        // TODO: Support for API_CommandTimedOut
609        //   It happens when we execute a command on a server with device-level
610        //   serialization, which prevents it from executing our command because another
611        //   action is holding the lock on the device monitor thread (another long
612        //   command or attribute).
613
614        let command = Scalar::from(name);
615        let input = input.into_value();
616        async move {
617            let backend = self.ensure_backend().await?;
618            let res = backend.command(command, input, self.inner.source).await;
619            O::from_value(self.check_backend(res).map_err(command_err)?)
620        }
621
622    }
623
624    /// Execute a remote command, giving back what it answered with, see
625    /// [`Self::command_async`].
626    pub fn command<I, O>(&self, name: &str, input: I) -> Result<O>
627    where
628        I: IntoValue,
629        O: FromValue,
630    {
631        self.block_on(self.command_async(name, input))?
632    }
633
634    /// Read attributes given a specification of attributes. This specification is really
635    /// flexible and allows various signatures for the function:
636    /// 1. `read_attr<T>(&str) -> Result<AttrRead<T>>`
637    /// 2. `read_attr<T>(&[&str]) -> Result<Vec<Result<AttrRead<T>>>>`
638    /// 3. `read_attr<T>([&str; N]) -> Result<[Result<AttrRead<T>>; N]>`
639    /// 4. `read_attr<(A, B, ...)>((&str, &str, ...)) -> Result<(Result<AttrRead<A>>, Result<AttrRead<B>>, ...)>`
640    /// 5. `read_attr<T>(All) -> Result<Vec<NamedResult<AttrRead<T>>>>`
641    ///
642    /// Forms 1 to 3 and 5 convert every attribute into the same type, whereas form 4 gives
643    /// each attribute its own type. Nothing in the names constrains the specification, so
644    /// it has to come either from a turbofish, leaving the names inferred as in
645    /// `read_attr::<i16, _>("enum_scalar")`, or from the type of the binding it is assigned
646    /// to. Reading into a [`Value`] keeps whatever the device answered with, and reading
647    /// into a buffer copies nothing, see [`Self::command_async`].
648    ///
649    /// Each reading comes with the [`Result`] of that one reading, so that a per-attribute
650    /// failure is seen without the whole request failing. Note that, unlike the
651    /// many-attributes forms, the single-attribute form has no per-attribute slot in its
652    /// output and therefore returns that attribute's error directly. The fixed-size forms
653    /// 3 and 4 additionally fail with [`ErrorKind::Internal`] if the device does not answer
654    /// with exactly as many readings as were requested.
655    ///
656    /// All requested attributes are read within a single request, which matters beyond
657    /// saving round trips: if the device enforces serialization, it is held for the whole
658    /// batch and only released at the end, so the readings are consistent with each other
659    /// instead of being sampled at instants another client could have written between.
660    #[instrument(skip(self, names), fields(target = %self.inner.target))]
661    pub fn read_attr_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
662    where
663        Spec: ReadAttrSpec<Names>,
664    {
665        let names = names.to_names();
666        async move {
667            let backend = self.ensure_backend().await?;
668            let res = backend.read_attr(names, self.inner.source).await;
669            Spec::collect(self.check_backend(res)?)
670        }
671    }
672
673    /// Read attributes, see [`Self::read_attr_async`].
674    pub fn read_attr<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
675    where
676        Spec: ReadAttrSpec<Names>,
677    {
678        self.block_on(self.read_attr_async::<Spec, Names>(names))?
679    }
680
681    /// Write attributes given a specification of attributes and their values, giving back
682    /// the outcome of each individual write, in the order they were given. In Tango a write
683    /// only carries a value: quality and timestamp cannot be written. Like for
684    /// [`Self::read_attr`], this specification is really flexible and allows various
685    /// signatures for the function:
686    /// 1. `write_attr((&str, T)) -> Result<()>`
687    /// 2. `write_attr(Vec<(&str, T)>) -> Result<Vec<Result<()>>>`
688    /// 3. `write_attr([(&str, T); N]) -> Result<[Result<()>; N]>`
689    /// 4. `write_attr(((&str, A), (&str, B), ...)) -> Result<(Result<()>, Result<()>, ...)>`
690    ///
691    /// Each written value only needs to implement [`IntoValue`], so unlike for
692    /// [`Self::read_attr`] the value types are inferred from the specification and never
693    /// have to be named. Note that, unlike the many-attributes forms, the single-attribute
694    /// form has no per-attribute slot in its output and therefore returns that attribute's
695    /// error directly. The fixed-size forms 3 and 4 fail with [`ErrorKind::Internal`] if
696    /// the device does not answer for exactly as many attributes as were written.
697    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
698    pub fn write_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send
699    where
700        Spec: WriteAttrSpec<'a>,
701    {
702
703        let values = spec.into_values();
704
705        async move {
706
707            let backend = self.ensure_backend().await?;
708            let res = backend.write_attr(values).await;
709
710            // A refused write is given the kind that fits it rather than the reason code
711            // the device raised with, the same way the request failing as a whole is.
712            let outcomes = self.check_backend(res)?.into_iter()
713                .map(|res| res.map_err(write_attr_err))
714                .collect();
715
716            Spec::collect(outcomes)
717
718        }
719
720    }
721
722    /// Write attributes, see [`Self::write_attr_async`].
723    pub fn write_attr<'a, Spec>(&self, spec: Spec) -> Result<Spec::Output>
724    where
725        Spec: WriteAttrSpec<'a>,
726    {
727        self.block_on(self.write_attr_async(spec))?
728    }
729
730    /// Write attributes and read them back, given the same flexible specification as
731    /// [`Self::write_attr`], but returning the read-back values instead of unit results.
732    /// Each value is converted back into the very type that was written:
733    /// 1. `write_read_attr((&str, T)) -> Result<AttrRead<T>>`
734    /// 2. `write_read_attr(Vec<(&str, T)>) -> Result<Vec<Result<AttrRead<T>>>>`
735    /// 3. `write_read_attr([(&str, T); N]) -> Result<[Result<AttrRead<T>>; N]>`
736    /// 4. `write_read_attr(((&str, A), (&str, B), ...)) -> Result<(Result<AttrRead<A>>, Result<AttrRead<B>>, ...)>`
737    ///
738    /// The read is done by the server right after the write, so this saves a round trip
739    /// compared to a [`Self::write_attr`] followed by a [`Self::read_attr`], and no other
740    /// request can interleave between the two when the device enforces serialization.
741    ///
742    /// Because the read-back is converted back into the written type, each value type must
743    /// also implement [`FromValue`] here, whereas [`Self::write_attr`] only requires
744    /// [`IntoValue`].
745    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
746    pub fn write_read_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>> + Send
747    where
748        Spec: WriteAttrSpec<'a>,
749        Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
750    {
751        let values = spec.into_values();
752        async move {
753            let backend = self.ensure_backend().await?;
754            let res = backend.write_read_attr(values).await;
755            Spec::ReadSpec::collect(self.check_backend(res)?)
756        }
757    }
758
759    /// Write attributes and read them back, see [`Self::write_read_attr_async`].
760    pub fn write_read_attr<'a, Spec>(&self, spec: Spec) -> Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>
761    where
762        Spec: WriteAttrSpec<'a>,
763        Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
764    {
765        self.block_on(self.write_read_attr_async(spec))?
766    }
767
768    /// Get the static information of commands, given the same flexible specification of
769    /// names as [`Self::read_attr`] takes:
770    /// 1. `command_info(&str) -> Result<CommandInfo>`
771    /// 2. `command_info(&[&str]) -> Result<Vec<Result<CommandInfo>>>`
772    /// 3. `command_info([&str; N]) -> Result<[Result<CommandInfo>; N]>`
773    /// 4. `command_info(All) -> Result<Vec<NamedResult<CommandInfo>>>`
774    ///
775    /// Every command comes with the [`Result`] of that one command, an unknown command name
776    /// and a data type that no [`CommandDataKind`] has a variant for being the two the
777    /// device answers with.
778    ///
779    /// Note that, unlike the attributes, the Tango interface has no operation taking a list
780    /// of command names, so asking for several of them costs one request each, where [`All`]
781    /// only costs one for the whole listing.
782    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
783    pub fn command_info_async<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
784        let names = spec.to_names();
785        async move {
786            let backend = self.ensure_backend().await?;
787            let res = backend.command_info(names).await;
788            Spec::collect(self.check_backend(res)?)
789        }
790    }
791
792    /// Get the static information of commands, see [`Self::command_info_async`].
793    pub fn command_info<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
794        self.block_on(self.command_info_async(spec))?
795    }
796
797    /// Get the static information of attributes, their runtime-modifiable configuration
798    /// included, given the same flexible specification of names as [`Self::read_attr`]
799    /// takes:
800    /// 1. `attr_info(&str) -> Result<AttrInfo>`
801    /// 2. `attr_info(&[&str]) -> Result<Vec<Result<AttrInfo>>>`
802    /// 3. `attr_info([&str; N]) -> Result<[Result<AttrInfo>; N]>`
803    /// 4. `attr_info(All) -> Result<Vec<NamedResult<AttrInfo>>>`
804    ///
805    /// Every attribute comes with the [`Result`] of that one attribute, a data type that no
806    /// [`AttrDataKind`] has a variant for being what the device answers with. Unlike the
807    /// commands, an unknown attribute name may fail the whole request, all of them travelling
808    /// in a single one, depending on the backend.
809    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
810    pub fn attr_info_async<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
811        let names = spec.to_names();
812        async move {
813            let backend = self.ensure_backend().await?;
814            let res = backend.attr_info(names).await;
815            Spec::collect(self.check_backend(res)?)
816        }
817    }
818
819    /// Get the static information of attributes, see [`Self::attr_info_async`].
820    pub fn attr_info<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
821        self.block_on(self.attr_info_async(spec))?
822    }
823
824    /// Change the configuration of the given attributes, giving back the outcome of each
825    /// individual change, in the order they were given.
826    ///
827    /// Only the configuration is writable, which is the part of an [`AttrInfo`] that a
828    /// client is allowed to set: what an attribute *is* — its mode, its data kind and its
829    /// format — is what its definition says and is never written from the outside.
830    ///
831    /// The whole configuration of an attribute goes out every time, so changing one
832    /// property of it is reading the configuration with [`Self::attr_info`] and writing
833    /// back what was read with that one property changed, which is what the protocol
834    /// carries.
835    ///
836    /// A field with no value brings it back to what the attribute inherits from its class,
837    /// if anything.
838    #[instrument(skip(self, values), fields(target = %self.inner.target))]
839    pub async fn set_attr_config_async(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
840        let configs = values.iter()
841            .map(|(name, config)| Named::new(Scalar::from(*name), config.clone()))
842            .collect();
843        let backend = self.ensure_backend().await?;
844        let res = backend.set_attr_config(configs).await;
845        self.check_backend(res).map_err(write_attr_err)
846    }
847
848    /// Change the configuration of the given attributes, see
849    /// [`Self::set_attr_config_async`].
850    pub fn set_attr_config(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
851        self.block_on(self.set_attr_config_async(values))?
852    }
853
854    /// The recorded history of a command, oldest entry first and at most `max` of them:
855    /// what the polling of the device recorded, or the device itself, which may manage the
856    /// history of a command it does not poll. Every entry is what the command answered
857    /// then, converted into the type it is expected to be, a run that failed carrying the
858    /// failure it failed with, see [`Recorded`].
859    ///
860    /// A command nothing was recorded of has no history to give, which is a failure and
861    /// not an empty history, see [`ErrorKind::CommandNotPolled`] and
862    /// [`ErrorKind::AttrNotPolled`].
863    #[instrument(skip(self), fields(target = %self.inner.target))]
864    pub async fn command_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<T>>> {
865        let backend = self.ensure_backend().await?;
866        let res = backend.command_history(Scalar::from(name), max).await;
867        Ok(self.check_backend(res).map_err(command_err)?.into_iter()
868            .map(|entry| entry.map(|value| value.and_then(T::from_value)))
869            .collect())
870    }
871
872    /// The recorded history of a command, see [`Self::command_history_async`].
873    pub fn command_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<T>>> {
874        self.block_on(self.command_history_async(name, max))?
875    }
876
877    /// The recorded history of an attribute, see [`Self::command_history_async`], a reading
878    /// carrying the time it was recorded at as its own.
879    #[instrument(skip(self), fields(target = %self.inner.target))]
880    pub async fn attr_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<AttrRead<T>>>> {
881        let backend = self.ensure_backend().await?;
882        let res = backend.attr_history(Scalar::from(name), max).await;
883        Ok(self.check_backend(res)?.into_iter()
884            .map(|entry| entry.map(|read| read.and_then(AttrRead::parse)))
885            .collect())
886    }
887
888    /// The recorded history of an attribute, see [`Self::attr_history_async`].
889    pub fn attr_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<AttrRead<T>>>> {
890        self.block_on(self.attr_history_async(name, max))?
891    }
892
893    // The polling of a device is its server's, so everything below asks the admin device
894    // of its server, which writes what it changes to the database along, so that the
895    // device is polled the same once restarted.
896
897    /// Poll that attribute of this device at that period, or at another one if it is
898    /// polled already, see the `AddObjPolling` and `UpdObjPollingPeriod` commands of the
899    /// admin device.
900    pub async fn poll_attr_async(&self, name: &str, period: Duration) -> Result<()> {
901        self.poll_async("attribute", name, period).await
902    }
903
904    /// Poll that attribute of this device, see [`Self::poll_attr_async`].
905    pub fn poll_attr(&self, name: &str, period: Duration) -> Result<()> {
906        self.block_on(self.poll_attr_async(name, period))?
907    }
908
909    /// Poll that command of this device, see [`Self::poll_attr_async`]: only a command
910    /// taking no argument can be polled.
911    pub async fn poll_command_async(&self, name: &str, period: Duration) -> Result<()> {
912        self.poll_async("command", name, period).await
913    }
914
915    /// Poll that command of this device, see [`Self::poll_command_async`].
916    pub fn poll_command(&self, name: &str, period: Duration) -> Result<()> {
917        self.block_on(self.poll_command_async(name, period))?
918    }
919
920    /// Stop polling that attribute of this device, failing if it is not polled, see the
921    /// `RemObjPolling` command of the admin device.
922    pub async fn stop_poll_attr_async(&self, name: &str) -> Result<()> {
923        self.stop_poll_async("attribute", name).await
924    }
925
926    /// Stop polling that attribute of this device, see [`Self::stop_poll_attr_async`].
927    pub fn stop_poll_attr(&self, name: &str) -> Result<()> {
928        self.block_on(self.stop_poll_attr_async(name))?
929    }
930
931    /// Stop polling that command of this device, see [`Self::stop_poll_attr_async`].
932    pub async fn stop_poll_command_async(&self, name: &str) -> Result<()> {
933        self.stop_poll_async("command", name).await
934    }
935
936    /// Stop polling that command of this device, see [`Self::stop_poll_command_async`].
937    pub fn stop_poll_command(&self, name: &str) -> Result<()> {
938        self.block_on(self.stop_poll_command_async(name))?
939    }
940
941    /// The period that attribute of this device is polled at, absent if it is not polled,
942    /// out of its polling status, see [`Self::polling_status_async`].
943    pub async fn attr_poll_period_async(&self, name: &str) -> Result<Option<Duration>> {
944        Ok(status_period(&self.polling_status_async().await?, "attribute", name))
945    }
946
947    /// The period that attribute of this device is polled at, see
948    /// [`Self::attr_poll_period_async`].
949    pub fn attr_poll_period(&self, name: &str) -> Result<Option<Duration>> {
950        self.block_on(self.attr_poll_period_async(name))?
951    }
952
953    /// The period that command of this device is polled at, see
954    /// [`Self::attr_poll_period_async`].
955    pub async fn command_poll_period_async(&self, name: &str) -> Result<Option<Duration>> {
956        Ok(status_period(&self.polling_status_async().await?, "command", name))
957    }
958
959    /// The period that command of this device is polled at, see
960    /// [`Self::command_poll_period_async`].
961    pub fn command_poll_period(&self, name: &str) -> Result<Option<Duration>> {
962        self.block_on(self.command_poll_period_async(name))?
963    }
964
965    /// The polling status of this device, one text per object polled, as its admin device
966    /// gives it with its `DevPollStatus` command, none for a device that is not polled.
967    pub async fn polling_status_async(&self) -> Result<Vec<String>> {
968        let name = self.name_async().await?;
969        self.admin_dev_async().await?.command_async("DevPollStatus", &*name).await
970    }
971
972    /// The polling status of this device, see [`Self::polling_status_async`].
973    pub fn polling_status(&self) -> Result<Vec<String>> {
974        self.block_on(self.polling_status_async())?
975    }
976
977    /// Poll that object of this device, of that type as the admin device names it, at that
978    /// period, whether it is polled already or not, see [`Self::poll_attr_async`].
979    async fn poll_async(&self, kind: &str, name: &str, period: Duration) -> Result<()> {
980
981        let dev_name = self.name_async().await?;
982        let admin = self.admin_dev_async().await?;
983        let period = i32::try_from(period.as_millis()).map_err(|_| Error::new(ErrorKind::InvalidData)
984            .with_description(format!("Polling period {period:?} is too long")))?;
985
986        let input = Value::LongString(
987            Spectrum::from_slice(&[period], [1]).expect("one period"), 
988            Spectrum::from(&[&*dev_name, kind, name][..]));
989
990        // Only an object that is not polled can be added, whose period is otherwise
991        // updated, as cppTango does.
992        match admin.command_async::<_, ()>("AddObjPolling", input.clone()).await {
993            Err(err) if err.reason() == reason::API_AlreadyPolled => {
994                admin.command_async("UpdObjPollingPeriod", input).await
995            }
996            res => res,
997        }
998
999    }
1000
1001    /// Stop polling that object of this device, see [`Self::stop_poll_attr_async`].
1002    async fn stop_poll_async(&self, kind: &str, name: &str) -> Result<()> {
1003        let dev_name = self.name_async().await?;
1004        let admin = self.admin_dev_async().await?;
1005        admin.command_async("RemObjPolling", Spectrum::<str>::from(&[&*dev_name, kind, name][..])).await
1006    }
1007
1008    // The properties of a device are the database's and not the device's, so everything
1009    // below asks the device for its name and then the database for what it holds of it,
1010    // see the very same methods on [`DatabaseProxy`].
1011
1012    /// The names of the properties of this device that match the given filter.
1013    pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
1014        let dev_name = self.name_async().await?;
1015        self.database()?.list_dev_prop_async(&dev_name, filter).await
1016    }
1017
1018    /// The names of the properties of this device, see [`Self::list_prop_async`].
1019    pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
1020        self.block_on(self.list_prop_async(filter))?
1021    }
1022
1023    /// The properties of this device, parsed into the shape the given names ask for. Like
1024    /// for [`Self::read_attr`], this specification is really flexible and allows various
1025    /// signatures for the function:
1026    /// 1. `get_prop<T>(&str) -> Result<T>`
1027    /// 2. `get_prop<T>(&[&str]) -> Result<Vec<Result<T>>>`
1028    /// 3. `get_prop<T>([&str; N]) -> Result<[Result<T>; N]>`
1029    /// 4. `get_prop<(A, B, ...)>((&str, &str, ...)) -> Result<(Result<A>, Result<B>, ...)>`
1030    /// 5. `get_prop<Spectrum<str>>(All) -> Result<Vec<Named<Spectrum<str>>>>`
1031    ///
1032    /// See [`DatabaseProxy::get_dev_prop_async`] for what the values may be parsed into
1033    /// and what a property the database holds nothing for gives.
1034    pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
1035    where
1036        Spec: GetPropSpec<Names>,
1037    {
1038        // The name of the device, which goes first, is only known once it is asked for, so
1039        // the names are taken out as a buffer of their own first.
1040        let names = names.to_names();
1041        async move {
1042            let dev_name = self.name_async().await?;
1043            let params = names.map(|names| prop_params(&dev_name, names.iter()));
1044            let mut reply = self.database()?.get_raw_prop_async(PropCommands::DEV, &dev_name, params).await?;
1045            Spec::collect(&mut reply)
1046        }
1047    }
1048
1049    /// The properties of this device, see [`Self::get_prop_async`].
1050    pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
1051    where
1052        Spec: GetPropSpec<Names>,
1053    {
1054        self.block_on(self.get_prop_async::<Spec, Names>(names))?
1055    }
1056
1057    /// Write the given properties of this device, given as pairs of a name and a value,
1058    /// which allows various signatures for the function:
1059    /// 1. `set_prop((&str, T)) -> Result<()>`
1060    /// 2. `set_prop([(&str, T); N]) -> Result<()>`
1061    /// 3. `set_prop(Vec<(&str, T)>) -> Result<()>`
1062    /// 4. `set_prop(((&str, A), (&str, B), ...)) -> Result<()>`
1063    ///
1064    /// See [`DatabaseProxy::set_dev_prop_async`] for what the values may be.
1065    pub fn set_prop_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<()>> + Send
1066    where
1067        Spec: SetPropSpec<'a>,
1068    {
1069        let params = set_prop_params(spec);
1070        async move {
1071            let dev_name = self.name_async().await?;
1072            self.database()?.prop_command_async("DbPutDeviceProperty", prop_params(&dev_name, params.iter().map(|param| &**param))).await
1073        }
1074    }
1075
1076    /// Write the given properties of this device, see [`Self::set_prop_async`].
1077    pub fn set_prop<'a, Spec>(&self, spec: Spec) -> Result<()>
1078    where
1079        Spec: SetPropSpec<'a>,
1080    {
1081        self.block_on(self.set_prop_async(spec))?
1082    }
1083
1084    /// Delete the given properties of this device, which allows various signatures for the
1085    /// function:
1086    /// 1. `delete_prop(&str) -> Result<()>`
1087    /// 2. `delete_prop(&[&str]) -> Result<()>`
1088    /// 3. `delete_prop([&str; N]) -> Result<()>`
1089    /// 4. `delete_prop((&str, &str, ...)) -> Result<()>`
1090    ///
1091    /// See [`DatabaseProxy::delete_dev_prop_async`].
1092    pub fn delete_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
1093        let names = names.to_names_not_all();
1094        async move {
1095            let dev_name = self.name_async().await?;
1096            self.database()?.prop_command_async("DbDeleteDeviceProperty", prop_params(&dev_name, names.iter())).await
1097        }
1098    }
1099
1100    /// Delete the given properties of this device, see [`Self::delete_prop_async`].
1101    pub fn delete_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
1102        self.block_on(self.delete_prop_async(names))?
1103    }
1104
1105    /// Subscribe to the readings of an attribute, of one of the kinds a device pushes them
1106    /// under, the channel giving the latest of them, see [`DevChannel`].
1107    ///
1108    /// Nothing is subscribed until the channel is polled, or its subscription ensured, see
1109    /// [`DevChannel::ensure_subscribed_async`]. There is no initial value: the first event
1110    /// is the next one the device pushes, so read the attribute once subscribed to know
1111    /// where it stands.
1112    pub fn subscribe_attr_value<T: FromValue>(&self, name: &str, event: AttrValueEvent) -> DevChannel<AttrRead<T>> {
1113        DevChannel::new(self.clone(), name, ChannelKind::Value(event), ChannelEvents::poll_value)
1114    }
1115
1116    /// Subscribe to an attribute being announced as ready to be read, which carries nothing
1117    /// of the reading: making it is the subscriber's to do, see [`AttrReady`]. See
1118    /// [`Self::subscribe_attr_value`].
1119    pub fn subscribe_attr_ready(&self, name: &str) -> DevChannel<AttrReady> {
1120        DevChannel::new(self.clone(), name, ChannelKind::Ready, ChannelEvents::poll_ready)
1121    }
1122
1123    /// Subscribe to the configuration of an attribute changing, every event carrying the
1124    /// configuration as it stands after the change. See [`Self::subscribe_attr_value`].
1125    pub fn subscribe_attr_config(&self, name: &str) -> DevChannel<AttrInfo> {
1126        DevChannel::new(self.clone(), name, ChannelKind::Config, ChannelEvents::poll_config)
1127    }
1128
1129}
1130
1131/// The clone shares the connection of this proxy if it can read it right away, and connects
1132/// on its own otherwise.
1133/// The period an object is polled at, out of the polling status of its device, as the
1134/// `DevPollStatus` command of its admin device gives it: the text of each object polled
1135/// starts with `Polled <kind> name = <name>` and tells its period in a later line, which is
1136/// what cppTango gives and its clients parse.
1137pub(crate) fn status_period(status: &[impl AsRef<str>], kind: &str, name: &str) -> Option<Duration> {
1138    let prefix = format!("Polled {kind} name = ");
1139    status.iter().find_map(|entry| {
1140        let mut lines = entry.as_ref().lines();
1141        let polled = lines.next()?.strip_prefix(&prefix)?;
1142        if !polled.trim().eq_ignore_ascii_case(name) {
1143            return None;
1144        }
1145        lines.find_map(|line| line.strip_prefix("Polling period (mS) = "))
1146            .and_then(|ms| ms.trim().parse().ok())
1147            .map(Duration::from_millis)
1148    })
1149}
1150
1151impl Clone for DevProxy {
1152    fn clone(&self) -> Self {
1153        Self {
1154            inner: Box::new(ProxyInner {
1155                backend: self.inner.backend.clone(),
1156                target: self.inner.target.clone(),
1157                source: self.inner.source,
1158                timeout: self.inner.timeout,
1159            }),
1160            state: self.state.try_read().ok()
1161                .map(|guard| RwLock::new((*guard).clone()))
1162                .unwrap_or_else(|| RwLock::new(ProxyState {
1163                    backend: None,
1164                    database_host_and_port: None,
1165                })),
1166        }
1167    }
1168}
1169
1170impl PartialEq for DevProxy {
1171    #[inline]
1172    fn eq(&self, other: &Self) -> bool {
1173        Arc::ptr_eq(&self.inner.backend, &other.inner.backend) && self.inner.target == other.inner.target
1174    }
1175}
1176
1177impl Eq for DevProxy { }
1178
1179impl Hash for DevProxy {
1180    #[inline]
1181    fn hash<H: Hasher>(&self, state: &mut H) {
1182        Arc::as_ptr(&self.inner.backend).hash(state);
1183        self.inner.target.hash(state);
1184    }
1185}
1186
1187/// The enum is internal so we just impl Display on it, in order to properly display it
1188/// in the tracing spans!
1189impl fmt::Display for ProxyTarget {
1190    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1191        match self {
1192            ProxyTarget::DatabaseFromEnv => 
1193                write!(f, "db"),
1194            ProxyTarget::DatabaseAt { host, port } => 
1195                write!(f, "db@{host}:{port}"),
1196            ProxyTarget::RegularFromDatabase { name, .. } => 
1197                write!(f, "{name}"),
1198            ProxyTarget::RegularAt { host, port, name } => 
1199                write!(f, "{name}@{host}:{port}"),
1200        }
1201    }
1202}
1203
1204// =========================== //
1205//          To names           //
1206// =========================== //
1207
1208/// Supporting trait for being generic over the different types that can be interpreted 
1209/// as names, either single or many names.
1210///
1211/// The names are given to a backend as one buffer of strings, which a served device shares
1212/// in what it answers with, and which a buffer of names already given passes through
1213/// untouched: names read back from a device, or listed by the database, are given again
1214/// without being copied.
1215///
1216/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
1217/// implemented for everything that API accepts and is not one to implement anywhere else.
1218pub trait ToNames: Sized {
1219
1220    /// The names, in the order they were given, absent to target **all** elements, borrowed
1221    /// where they are: what a request copies into its own parameters, once, rather than
1222    /// first into a buffer of their own, see [`Spectrum::from_strs`].
1223    fn iter_names(&self) -> Option<impl Iterator<Item = &str>>;
1224
1225    /// The names as one buffer, see [`Self::iter_names`].
1226    #[inline]
1227    fn to_names(self) -> Option<Spectrum<str>> {
1228        self.iter_names().map(names_buf)
1229    }
1230
1231}
1232
1233/// The names as one buffer, copied out of where they are.
1234fn names_buf<'a>(names: impl Iterator<Item = &'a str>) -> Spectrum<str> {
1235    Spectrum::from_strs(names).expect("names of more than 4 GiB")
1236}
1237
1238/// Same as [`ToNames`] but this marker ensures that no None will be returned by to_names.
1239///
1240/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
1241/// implemented for everything that API accepts and is not one to implement anywhere else.
1242pub trait ToNamesNotAll: ToNames {
1243
1244    /// Utility function to ensure no `None` could be returned for the names.
1245    #[inline]
1246    fn iter_names_not_all(&self) -> impl Iterator<Item = &str> {
1247        self.iter_names().expect("should not return the 'all' marker")
1248    }
1249
1250    /// Utility function to ensure no `None` could be returned for the names.
1251    #[inline]
1252    fn to_names_not_all(self) -> Spectrum<str> {
1253        self.to_names().expect("should not return the 'all' marker")
1254    }
1255
1256}
1257
1258/// Something along with the name it is of, the name sharing the octets it was decoded out
1259/// of like any other string.
1260///
1261/// This is what a request answering for several names gives one of per name: the result
1262/// of that one name, which is not the one of the request. The request failing fails it as
1263/// a whole, while an entry failing is the device failing for that one name, which the other
1264/// entries are unaffected by.
1265#[derive(Debug, Clone, PartialEq)]
1266pub struct Named<T> {
1267    /// The name.
1268    pub name: Scalar<str>,
1269    /// What is of that name.
1270    pub value: T,
1271}
1272
1273impl<T> Named<T> {
1274
1275    #[inline]
1276    pub fn new(name: Scalar<str>, value: T) -> Self {
1277        Self { name, value }
1278    }
1279
1280    /// The same name along with what that function makes of its value.
1281    #[inline]
1282    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Named<U> {
1283        Named { name: self.name, value: f(self.value) }
1284    }
1285
1286}
1287
1288/// What a request answering for several names gives one of per name, see [`Named`].
1289pub type NamedResult<T> = Named<Result<T>>;
1290
1291/// Something along with the time it was recorded at, which is what a history gives one of
1292/// per entry, see [`DevProxy::command_history`]: the result of what was recorded then,
1293/// which is not the one of the request asking for the history.
1294#[derive(Debug, Clone, PartialEq)]
1295pub struct Recorded<T> {
1296    /// When it was recorded.
1297    pub time: SystemTime,
1298    /// What was recorded.
1299    pub value: T,
1300}
1301
1302impl<T> Recorded<T> {
1303
1304    #[inline]
1305    pub fn new(time: SystemTime, value: T) -> Self {
1306        Self { time, value }
1307    }
1308
1309    /// The same time along with what that function makes of its value.
1310    #[inline]
1311    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Recorded<U> {
1312        Recorded { time: self.time, value: f(self.value) }
1313    }
1314
1315}
1316
1317/// What a history gives one of per entry, see [`Recorded`].
1318pub type RecordedResult<T> = Recorded<Result<T>>;
1319
1320/// A marker struct for specifying names for various functions that are using [`ToNames`]
1321/// to specify to select **all** names available on the device.
1322pub struct All;
1323impl ToNames for All {
1324    #[inline]
1325    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1326        None::<std::iter::Empty<&str>>
1327    }
1328}
1329
1330impl ToNames for &str {
1331    #[inline]
1332    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1333        Some(std::iter::once(*self))
1334    }
1335}
1336impl ToNamesNotAll for &str { }
1337
1338impl ToNames for &[&str] {
1339    #[inline]
1340    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1341        Some(self.iter().copied())
1342    }
1343}
1344impl ToNamesNotAll for &[&str] { }
1345
1346impl<const N: usize> ToNames for [&str; N] {
1347    #[inline]
1348    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1349        Some(self.iter().copied())
1350    }
1351}
1352impl<const N: usize> ToNamesNotAll for [&str; N] { }
1353
1354impl ToNames for Spectrum<str> {
1355    #[inline]
1356    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1357        Some(self.iter())
1358    }
1359    #[inline]
1360    fn to_names(self) -> Option<Spectrum<str>> {
1361        Some(self)
1362    }
1363}
1364impl ToNamesNotAll for Spectrum<str> { }
1365
1366impl ToNames for &Spectrum<str> {
1367    #[inline]
1368    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1369        Some(self.iter())
1370    }
1371    #[inline]
1372    fn to_names(self) -> Option<Spectrum<str>> {
1373        Some(self.clone())
1374    }
1375}
1376impl ToNamesNotAll for &Spectrum<str> { }
1377
1378macro_rules! impl_to_names_tuple {
1379    ( @name: $comp:ident ) => { &str };
1380    ( @one: $comp:ident ) => { 1 };
1381    ( $( $comp:ident )* ) => {
1382        #[allow(unused_variables)]
1383        impl ToNames for ( $(impl_to_names_tuple!(@name: $comp),)* ) {
1384            #[inline]
1385            fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1386                let ( $($comp,)* ) = *self;
1387                Some(<[&str; 0 $( + impl_to_names_tuple!(@one: $comp) )*]>::into_iter([ $($comp,)* ]))
1388            }
1389        }
1390        impl ToNamesNotAll for ( $(impl_to_names_tuple!(@name: $comp),)* ) { }
1391    };
1392}
1393
1394impl_to_names_tuple!();
1395impl_to_names_tuple!(a);
1396impl_to_names_tuple!(a b);
1397impl_to_names_tuple!(a b c);
1398impl_to_names_tuple!(a b c d);
1399impl_to_names_tuple!(a b c d e);
1400impl_to_names_tuple!(a b c d e f);
1401
1402/// Generic specification type used for command and attribute info, as there is no 
1403/// special processing to do on the resulted value! This is an extension to the ToNames,
1404/// because this spec's output only depends on the actual type of the names we give.
1405///
1406/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
1407/// implemented for everything that API accepts and is not one to implement anywhere else.
1408pub trait ToNamesSpec<T>: ToNames {
1409
1410    /// The final type of the output.
1411    type Output;
1412
1413    /// Build the output out of the entries the device answered with, one per name.
1414    ///
1415    /// The shape of the output is what the names that were given say it is, which is what
1416    /// this is implemented on.
1417    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output>;
1418
1419}
1420
1421/// One name, one entry: that entry *is* the output, and its failure is the output's own
1422/// since there is no slot of its own to carry it in.
1423impl<T> ToNamesSpec<T> for &str {
1424
1425    type Output = T;
1426
1427    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1428        match entries.into_iter().next() {
1429            Some(entry) => entry.value,
1430            None => Err(unexpected_len_err(1)),
1431        }
1432    }
1433
1434}
1435
1436impl<T> ToNamesSpec<T> for &[&str] {
1437
1438    type Output = Vec<Result<T>>;
1439
1440    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1441        Ok(entries.into_iter().map(|entry| entry.value).collect())
1442    }
1443
1444}
1445
1446impl<T> ToNamesSpec<T> for Spectrum<str> {
1447
1448    type Output = Vec<Result<T>>;
1449
1450    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1451        <&[&str] as ToNamesSpec<T>>::collect(entries)
1452    }
1453
1454}
1455
1456impl<T> ToNamesSpec<T> for &Spectrum<str> {
1457
1458    type Output = Vec<Result<T>>;
1459
1460    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1461        <&[&str] as ToNamesSpec<T>>::collect(entries)
1462    }
1463
1464}
1465
1466impl<T, const N: usize> ToNamesSpec<T> for [&str; N] {
1467
1468    type Output = [Result<T>; N];
1469
1470    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1471        take_array(entries.into_iter().map(|entry| entry.value))
1472    }
1473
1474}
1475
1476impl<T> ToNamesSpec<T> for All {
1477
1478    // Nothing was named here, so every entry carries its own name and this is the one
1479    // form whose output has room for it.
1480    type Output = Vec<NamedResult<T>>;
1481
1482    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1483        Ok(entries)
1484    }
1485
1486}
1487
1488/// The entries of an output of a fixed shape, one per slot and in order: a device that
1489/// answers with another number of them broke what the protocol says of it, which a last
1490/// entry being there behind the slots tells as well as one of them missing.
1491pub(crate) fn take_array<T, const N: usize>(mut entries: impl Iterator<Item = T>) -> Result<[T; N]> {
1492    let mut slots = [const { None }; N];
1493    for slot in &mut slots {
1494        *slot = Some(entries.next().ok_or_else(|| unexpected_len_err(N))?);
1495    }
1496    if entries.next().is_some() {
1497        return Err(unexpected_len_err(N));
1498    }
1499    Ok(slots.map(|slot| slot.expect("every slot is filled")))
1500}
1501
1502macro_rules! impl_to_names_spec_tuple {
1503    ( @one: $index:tt ) => { 1 };
1504    ( @name: $index:tt ) => { &str };
1505    ( @t: $index:tt ) => { T };
1506    ( @next: $index:tt, $entries:ident ) => { $entries.next().ok_or_else(|| unexpected_len_err(LEN))?.value };
1507    ( $( $index:tt )* ) => {
1508
1509        impl<T> ToNamesSpec<T> for ( $( impl_to_names_spec_tuple!(@name: $index), )* ) {
1510
1511            type Output = ( $( Result< impl_to_names_spec_tuple!(@t: $index) >,)* );
1512
1513            fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1514                const LEN: usize = 0 $( + impl_to_names_spec_tuple!(@one: $index) )*;
1515                // The entries are in the order the names were given, which is the order the
1516                // components of a tuple are built in.
1517                let mut entries = entries.into_iter();
1518                let output = ( $( impl_to_names_spec_tuple!(@next: $index, entries), )* );
1519                if entries.next().is_some() {
1520                    return Err(unexpected_len_err(LEN));
1521                }
1522                Ok(output)
1523            }
1524
1525        }
1526
1527    };
1528}
1529
1530impl_to_names_spec_tuple!(0);
1531impl_to_names_spec_tuple!(0 1);
1532impl_to_names_spec_tuple!(0 1 2);
1533impl_to_names_spec_tuple!(0 1 2 3);
1534impl_to_names_spec_tuple!(0 1 2 3 4);
1535impl_to_names_spec_tuple!(0 1 2 3 4 5);
1536
1537// =========================== //
1538//       Read attributes       //
1539// =========================== //
1540
1541/// A trait for attribute specifiers, which allows a great flexibility in how 
1542/// [`DevProxy::read_attr`] works.
1543///
1544/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
1545/// implemented for everything that API accepts and is not one to implement anywhere else.
1546pub trait ReadAttrSpec<Names> {
1547
1548    /// The final type of the output.
1549    type Output;
1550
1551    /// Build the output out of the readings, one per attribute.
1552    ///
1553    /// The shape of the output is what the names that were given say it is, and what every
1554    /// reading is converted into is what this is implemented on.
1555    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output>;
1556
1557}
1558
1559impl<T: FromValue> ReadAttrSpec<&str> for T {
1560
1561    type Output = AttrRead<T>;
1562
1563    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1564        match reads.into_iter().next() {
1565            Some(entry) => entry.value?.parse(),
1566            None => Err(unexpected_len_err(1)),
1567        }
1568    }
1569
1570}
1571
1572impl<T: FromValue> ReadAttrSpec<&'_ [&str]> for T {
1573
1574    type Output = Vec<Result<AttrRead<T>>>;
1575
1576    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1577        Ok(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)).collect())
1578    }
1579
1580}
1581
1582impl<T: FromValue> ReadAttrSpec<Spectrum<str>> for T {
1583
1584    type Output = Vec<Result<AttrRead<T>>>;
1585
1586    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1587        <T as ReadAttrSpec<&[&str]>>::collect(reads)
1588    }
1589
1590}
1591
1592impl<T: FromValue> ReadAttrSpec<&Spectrum<str>> for T {
1593
1594    type Output = Vec<Result<AttrRead<T>>>;
1595
1596    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1597        <T as ReadAttrSpec<&[&str]>>::collect(reads)
1598    }
1599
1600}
1601
1602impl<T: FromValue, const N: usize> ReadAttrSpec<[&str; N]> for T {
1603
1604    type Output = [Result<AttrRead<T>>; N];
1605
1606    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1607        take_array(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)))
1608    }
1609
1610}
1611
1612impl<T: FromValue> ReadAttrSpec<All> for T {
1613
1614    // Nothing was named here, so every reading carries the name of its attribute.
1615    type Output = Vec<NamedResult<AttrRead<T>>>;
1616
1617    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1618        Ok(reads.into_iter().map(|entry| entry.map(|read| read.and_then(AttrRead::parse))).collect())
1619    }
1620
1621}
1622
1623macro_rules! impl_read_attr_spec_tuple {
1624    ( @one: $comp:ident ) => { 1 };
1625    ( @name: $comp:ident ) => { &'name str };
1626    ( $( $comp:ident )* ) => {
1627
1628        impl<'name, $($comp,)* > ReadAttrSpec<( $( impl_read_attr_spec_tuple!(@name: $comp), )* )> for ( $($comp,)* )
1629        where
1630            $( $comp: FromValue, )*
1631        {
1632
1633            type Output = ( $( Result<AttrRead<$comp>>,)* );
1634
1635            fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1636                const LEN: usize = 0 $( + impl_read_attr_spec_tuple!(@one: $comp) )*;
1637                // The readings are in the order the names were given, which is the order
1638                // the components of a tuple are built in, and every one of them is
1639                // converted into the type of its own component.
1640                let mut reads = reads.into_iter();
1641                let output = ( $( reads.next().ok_or_else(|| unexpected_len_err(LEN))?.value.and_then(AttrRead::parse::<$comp>), )* );
1642                if reads.next().is_some() {
1643                    return Err(unexpected_len_err(LEN));
1644                }
1645                Ok(output)
1646            }
1647
1648        }
1649
1650    };
1651}
1652
1653impl_read_attr_spec_tuple!(A);
1654impl_read_attr_spec_tuple!(A B);
1655impl_read_attr_spec_tuple!(A B C);
1656impl_read_attr_spec_tuple!(A B C D);
1657impl_read_attr_spec_tuple!(A B C D E);
1658impl_read_attr_spec_tuple!(A B C D E F);
1659
1660// =========================== //
1661//      Write attributes       //
1662// =========================== //
1663
1664/// The values to write to one or many attributes, see [`DevProxy::write_attr`].
1665///
1666/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
1667/// implemented for everything that API accepts and is not one to implement anywhere else.
1668pub trait WriteAttrSpec<'a>: Sized {
1669
1670    /// Make this write specification into the values it represents, each with the name of
1671    /// its attribute.
1672    fn into_values(self) -> Vec<Named<Value>>;
1673
1674    /// The final type of the output.
1675    type Output;
1676
1677    /// Build the output out of the outcomes of the write, which come in the order the
1678    /// values were given: a write answers one outcome per value and names nothing back,
1679    /// see [`DevProxy::write_attr_async`].
1680    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output>;
1681
1682    /// Names for the read specification type.
1683    type ReadNames;
1684    /// The read specification type, for when using write_read_attr.
1685    type ReadSpec;
1686
1687}
1688
1689/// For a single attribute.
1690impl<'a, T: IntoValue> WriteAttrSpec<'a> for (&'a str, T) {
1691
1692    #[inline]
1693    fn into_values(self) -> Vec<Named<Value>> {
1694        vec![Named::new(Scalar::from(self.0), self.1.into_value())]
1695    }
1696
1697    type Output = ();
1698
1699    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1700        match writes.into_iter().next() {
1701            Some(res) => res,
1702            None => Err(unexpected_len_err(1)),
1703        }
1704    }
1705
1706    type ReadNames = &'a str;
1707    type ReadSpec = T;
1708
1709}
1710
1711/// For an array of attributes.
1712impl<'a, T: IntoValue, const N: usize> WriteAttrSpec<'a> for [(&'a str, T); N] {
1713
1714    #[inline]
1715    fn into_values(self) -> Vec<Named<Value>> {
1716        named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
1717    }
1718
1719    type Output = [Result<()>; N];
1720
1721    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1722        take_array(writes.into_iter())
1723    }
1724
1725    type ReadNames = [&'a str; N];
1726    type ReadSpec = T;
1727
1728}
1729
1730/// For a vector of attributes.
1731impl<'a, T: IntoValue> WriteAttrSpec<'a> for Vec<(&'a str, T)> {
1732
1733    #[inline]
1734    fn into_values(self) -> Vec<Named<Value>> {
1735        named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
1736    }
1737
1738    type Output = Vec<Result<()>>;
1739
1740    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1741        Ok(writes)
1742    }
1743
1744    type ReadNames = &'a [&'a str];
1745    type ReadSpec = T;
1746
1747}
1748
1749/// The values of a write, every name a scalar of its own sharing a single copy of all the
1750/// names, see [`Spectrum::get_scalar`].
1751fn named_values(values: Vec<(&str, Value)>) -> Vec<Named<Value>> {
1752    let names = names_buf(values.iter().map(|&(name, _)| name));
1753    values.into_iter()
1754        .enumerate()
1755        .map(|(index, (_, value))| Named::new(names.get_scalar([index]).expect("as many names as values"), value))
1756        .collect()
1757}
1758
1759macro_rules! impl_write_attr_spec_tuple {
1760    ( @name: $comp:ident ) => { &'a str };
1761    ( @one: $comp:ident ) => { 1 };
1762    ( @unit: $comp:ident ) => { Result<()> };
1763    ( @next: $comp:ident, $writes:ident ) => { $writes.next().ok_or_else(|| unexpected_len_err(LEN))? };
1764    ( $( $comp:ident )* ) => {
1765
1766        #[allow(non_snake_case)]
1767        impl<'a, $($comp: IntoValue, )*> WriteAttrSpec<'a> for ( $( (&'a str, $comp), )* ) {
1768
1769            #[inline]
1770            fn into_values(self) -> Vec<Named<Value>> {
1771                let ( $($comp,)* ) = self;
1772                named_values(vec![$(($comp.0, $comp.1.into_value()), )* ])
1773            }
1774
1775            type Output = ( $( impl_write_attr_spec_tuple!(@unit: $comp), )* );
1776
1777            fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1778                const LEN: usize = 0 $( + impl_write_attr_spec_tuple!(@one: $comp) )*;
1779                // The outcomes come in the order the values were given, which is the order
1780                // the components of a tuple are built in.
1781                let mut writes = writes.into_iter();
1782                let output = ( $( impl_write_attr_spec_tuple!(@next: $comp, writes), )* );
1783                if writes.next().is_some() {
1784                    return Err(unexpected_len_err(LEN));
1785                }
1786                Ok(output)
1787            }
1788
1789            type ReadNames = ( $(impl_write_attr_spec_tuple!(@name: $comp), )* );
1790            type ReadSpec = ( $($comp, )* );
1791
1792        }
1793
1794    };
1795}
1796
1797impl_write_attr_spec_tuple!(A);
1798impl_write_attr_spec_tuple!(A B);
1799impl_write_attr_spec_tuple!(A B C);
1800impl_write_attr_spec_tuple!(A B C D);
1801impl_write_attr_spec_tuple!(A B C D E);
1802impl_write_attr_spec_tuple!(A B C D E F);
1803
1804// =========================== //
1805//          Commands           //
1806// =========================== //
1807
1808/// Static information about a command.
1809#[derive(Debug, Clone, PartialEq, Eq)]
1810pub struct CommandInfo {
1811    /// The display level of the command.
1812    pub display_level: DisplayLevel,
1813    /// Input data kind.
1814    pub in_kind: CommandDataKind,
1815    /// Output data kind.
1816    pub out_kind: CommandDataKind,
1817    /// Description for the input data.
1818    pub in_doc: Option<Cow<'static, str>>,
1819    /// Description for the output data.
1820    pub out_doc: Option<Cow<'static, str>>,
1821}
1822
1823// =========================== //
1824//         Attributes          //
1825// =========================== //
1826
1827/// Represent the reading for a single attribute.
1828#[derive(Debug, Clone, PartialEq, Eq)]
1829pub struct AttrRead<T> {
1830    /// The read value of the attribute, if any. Absent if quality is Invalid or if the
1831    /// attribute is not readable.
1832    pub value: Option<T>,
1833    /// The write value of the attribute, if any. Absent if quality is Invalid or if the
1834    /// attribute is not writable.
1835    pub w_value: Option<T>,
1836    /// The quality of the value.
1837    pub quality: AttrQuality,
1838    /// The timestamp of the value.
1839    pub time: SystemTime,
1840}
1841
1842impl<T> AttrRead<T> {
1843
1844    pub fn new(value: T) -> Self {
1845        Self {
1846            value: Some(value),
1847            w_value: None,
1848            quality: AttrQuality::Valid,
1849            time: SystemTime::now(),
1850        }
1851    }
1852
1853    pub fn with_quality(mut self, quality: AttrQuality) -> Self {
1854        self.quality = quality;
1855        self
1856    }
1857
1858    pub fn with_time(mut self, time: SystemTime) -> Self {
1859        self.time = time;
1860        self
1861    }
1862
1863}
1864
1865impl AttrRead<Value> {
1866
1867    /// This reading, its values converted into the type they are expected to be.
1868    fn parse<T: FromValue>(self) -> Result<AttrRead<T>> {
1869        Ok(AttrRead {
1870            value: match self.value {
1871                Some(value) => Some(T::from_value(value)
1872                    .map_err(|err| context_err(err, "Failed to parse read value"))?),
1873                None => None,
1874            },
1875            w_value: match self.w_value {
1876                Some(value) => Some(T::from_value(value)
1877                    .map_err(|err| context_err(err, "Failed to parse write value"))?),
1878                None => None,
1879            },
1880            quality: self.quality,
1881            time: self.time,
1882        })
1883    }
1884
1885}
1886
1887/// An attribute announced as ready to be read, which is what a data ready event carries:
1888/// nothing of the reading itself, which is the subscriber's to make, but the data kind of
1889/// the attribute and a counter the device gives with every announcement, telling which of
1890/// its readings is the one made ready.
1891#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1892pub struct AttrReady {
1893    /// The data kind of the attribute.
1894    pub data_kind: AttrDataKind,
1895    /// The counter the device gave with this announcement, see
1896    /// [`crate::server::Dev::push_data_ready_event`].
1897    pub counter: i32,
1898}
1899
1900/// Quality factor of the attribute.
1901#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1902pub enum AttrQuality {
1903    Valid,
1904    Invalid,
1905    Alarm,
1906    Changing,
1907    Warning,
1908}
1909
1910/// An extension to AttrConfig with static information about the attribute.
1911#[derive(Debug, Clone, PartialEq)]
1912pub struct AttrInfo {
1913    /// The configuration of the attribute, which is the whole of what a client is allowed
1914    /// to change of it, see [`AttrConfig`]. Everything else here is what the attribute is,
1915    /// which the device declares and nothing outside of it ever changes.
1916    pub config: AttrConfig,
1917    /// The attribute mode.
1918    pub mode: AttrMode,
1919    /// The data kind to be expected.
1920    pub data_kind: AttrDataKind,
1921    /// The data format to be expected, along with the maximum dimensions the attribute is
1922    /// declared with for the formats that have any.
1923    pub data_format: AttrDataFormat,
1924    /// The display level of the attribute.
1925    pub display_level: DisplayLevel,
1926    /// The memorized mode, absent for an attribute that remembers nothing.
1927    pub memorized: Option<AttrMemorized>,
1928}
1929
1930/// Attribute read/write mode.
1931#[derive(Debug, Clone, PartialEq, Eq)]
1932pub enum AttrMode {
1933    /// Read-only attribute
1934    Read,
1935    /// The attribute is technically read-only on the server side, but is associated to
1936    /// another write-only attribute that acts as the set point for that read-only value.
1937    ReadWithWrite {
1938        /// The name of the associated write attribute.
1939        write_attr_name: Cow<'static, str>,
1940    },
1941    /// Write-only attribute.
1942    Write,
1943    /// Read/write attribute.
1944    ReadWrite,
1945}
1946
1947/// Represent the current configuration of an attribute.
1948///
1949/// The default configuration is the one of an attribute that declares nothing: no
1950/// description, no unit, no bound of any kind, no event configured, which is what a
1951/// definition starts from and then fills in.
1952///
1953/// This is exactly what Tango keeps as the *attribute properties* of a device in the
1954/// database, and therefore exactly what a client is allowed to change, which is what
1955/// [`crate::backend::BackendDev::set_attr_config`] carries. Everything else a client reads
1956/// about an attribute is what the attribute *is* rather than how it is configured, and is
1957/// declared by the device, see [`AttrInfo`].
1958#[derive(Debug, Clone, PartialEq, Default)]
1959pub struct AttrConfig {
1960    /// The optional description of the attribute, which a client may change like any other
1961    /// property, `description` being one of the attribute properties Tango keeps.
1962    pub doc: Option<Cow<'static, str>>,
1963    /// The optional label of the attribute.
1964    pub label: Option<Cow<'static, str>>,
1965    /// The optional real unit of the attribute: how the raw attribute value is
1966    /// represented.
1967    pub unit: Option<Cow<'static, str>>,
1968    /// The conversion factor for going from the raw attribute value to the international
1969    /// unit system.
1970    pub standard_unit: Option<f64>,
1971    /// The optional unit for displaying to the user.
1972    pub display_unit: Option<Cow<'static, str>>,
1973    /// The display format of this value. That format is specified using printf formatting
1974    /// symbols using `%`.
1975    pub display_format: Option<Cow<'static, str>>,
1976    /// The allowed range for this attribute's value.
1977    pub range: AttrRangeConfig,
1978    /// The alarm range for this attribute's value.
1979    pub alarm: AttrRangeConfig,
1980    /// The warning range for this attribute's value.
1981    pub warning: AttrRangeConfig,
1982    /// The optional delta alarm.
1983    pub delta: Option<AttrDeltaConfig>,
1984    /// Configuration of the change event.
1985    pub change_event: AttrChangeEventConfig,
1986    /// Configuration of the periodic event.
1987    pub periodic_event: AttrPeriodicEventConfig,
1988    /// Configuration of the archive event.
1989    pub archive_event: AttrArchiveEventConfig,
1990    /// What the values of the attribute mean, which only a `DevEnum` has any of: they name
1991    /// its values, an enumerated value being the index of its own label in this list.
1992    ///
1993    /// This is a property like any other and a client may change it, which is why it is
1994    /// here and not beside [`AttrInfo::data_kind`]: what a value of the attribute *is*
1995    /// stays a `DevEnum` whatever the labels are renamed to.
1996    pub enum_labels: Vec<Cow<'static, str>>,
1997    /// Possible extensions.
1998    pub extensions: Vec<Cow<'static, str>>,
1999}
2000
2001/// A limit of an attribute: one of the bounds of one of its ranges, or the delta of its
2002/// read-different-from-set alarm.
2003///
2004/// Tango carries these as strings, the type to read them as being the one of the attribute
2005/// they belong to, and they are numbers by the time they get here: a limit is something to
2006/// compare a value against, which is not something a string does.
2007///
2008/// They are not all the same number, though. A limit is a value of the attribute itself,
2009/// so an attribute of the widest integer kinds has limits a double cannot hold without
2010/// rounding them, and the three ways a Tango value is a number are kept apart rather than
2011/// all being flattened into one:
2012///
2013/// - a floating attribute, `DevFloat` or `DevDouble`, has [`Self::Double`] limits;
2014/// - a signed one, up to `DevLong64`, has [`Self::Long`] limits;
2015/// - an unsigned one, up to `DevULong64`, has [`Self::ULong`] limits.
2016///
2017/// A device that writes a limit its own type cannot hold — half a unit on an integer
2018/// attribute — still has a limit, and it is read as a [`Self::Double`] rather than lost.
2019///
2020/// Two limits compare as the quantities they stand for, whatever variant each is, so that
2021/// `Long(2)` and `Double(2.0)` are equal: two integers are compared exactly, the widest
2022/// ones included, and anything else as doubles, a not-a-number being equal to nothing and
2023/// ordered with nothing.
2024#[derive(Debug, Clone, Copy)]
2025pub enum AttrNumber {
2026    Double(f64),
2027    Long(i64),
2028    ULong(u64),
2029}
2030
2031impl AttrNumber {
2032
2033    /// This limit as a double, which is what comparing it against a value of any other
2034    /// kind goes through, the widest integers losing their last digits the way they do in
2035    /// any other conversion to a double.
2036    pub fn as_f64(self) -> f64 {
2037        match self {
2038            Self::Double(value) => value,
2039            Self::Long(value) => value as f64,
2040            Self::ULong(value) => value as f64,
2041        }
2042    }
2043
2044}
2045
2046impl PartialEq for AttrNumber {
2047    #[inline]
2048    fn eq(&self, other: &Self) -> bool {
2049        self.partial_cmp(other) == Some(Ordering::Equal)
2050    }
2051}
2052
2053impl PartialOrd for AttrNumber {
2054    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
2055        match (*self, *other) {
2056            (Self::Long(a), Self::Long(b)) => Some(a.cmp(&b)),
2057            (Self::ULong(a), Self::ULong(b)) => Some(a.cmp(&b)),
2058            (Self::Long(a), Self::ULong(b)) => Some(i128::from(a).cmp(&i128::from(b))),
2059            (Self::ULong(a), Self::Long(b)) => Some(i128::from(a).cmp(&i128::from(b))),
2060            (a, b) => a.as_f64().partial_cmp(&b.as_f64()),
2061        }
2062    }
2063}
2064
2065impl fmt::Display for AttrNumber {
2066    /// Written the way a device writes it, which is what it goes back on the wire as.
2067    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2068        match self {
2069            Self::Double(value) => Display::fmt(value, f),
2070            Self::Long(value) => Display::fmt(value, f),
2071            Self::ULong(value) => Display::fmt(value, f),
2072        }
2073    }
2074}
2075
2076/// Every number that fits in one of the three, so that a limit can be given as the value
2077/// it is rather than as the variant holding it.
2078macro_rules! impl_attr_number_from {
2079    ($variant:ident: $($ty:ty),* $(,)?) => { $(
2080        impl From<$ty> for AttrNumber {
2081            #[inline]
2082            fn from(value: $ty) -> Self {
2083                Self::$variant(value.into())
2084            }
2085        }
2086    )* };
2087}
2088
2089impl_attr_number_from!(Double: f32, f64);
2090impl_attr_number_from!(Long: i8, i16, i32, i64);
2091impl_attr_number_from!(ULong: u8, u16, u32, u64);
2092
2093/// The two bounds of one of the ranges of an attribute, either of which is absent for a
2094/// range that is open on that side, which is also how an attribute says it has no such
2095/// range at all.
2096#[derive(Debug, Clone, Copy, PartialEq, Default)]
2097pub struct AttrRangeConfig {
2098    pub min: Option<AttrNumber>,
2099    pub max: Option<AttrNumber>,
2100}
2101
2102impl AttrRangeConfig {
2103    /// Whether that range has any bound, a range with none being no range at all.
2104    #[inline]
2105    pub fn is_set(&self) -> bool {
2106        self.min.is_some() || self.max.is_some()
2107    }
2108}
2109
2110/// The alarm on the read value drifting away from the written one.
2111#[derive(Debug, Clone, Copy, PartialEq)]
2112pub struct AttrDeltaConfig {
2113    /// How far the two may be apart before the attribute is in alarm, in the type of the
2114    /// attribute, being a value of it like any other limit.
2115    pub value: AttrNumber,
2116    /// How long they may stay that far apart before it counts, absent for immediately.
2117    pub time: Option<Duration>,
2118}
2119
2120/// The criteria of the change event: how much the value has to move for the server to send
2121/// one, absolutely or relatively, either of which may be left out.
2122#[derive(Debug, Clone, Copy, PartialEq, Default)]
2123pub struct AttrChangeEventConfig {
2124    /// The relative change, as a percentage of the previous value.
2125    pub rel_change: Option<f64>,
2126    /// The absolute change, in the unit of the attribute.
2127    pub abs_change: Option<f64>,
2128}
2129
2130impl AttrChangeEventConfig {
2131    /// Whether the change event has any criterion, an event with none being configured.
2132    #[inline]
2133    pub fn is_set(&self) -> bool {
2134        self.rel_change.is_some() || self.abs_change.is_some()
2135    }
2136}
2137
2138#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
2139pub struct AttrPeriodicEventConfig {
2140    pub period: Option<Duration>,
2141}
2142
2143impl AttrPeriodicEventConfig {
2144    /// Whether the periodic event has a period of its own.
2145    #[inline]
2146    pub fn is_set(&self) -> bool {
2147        self.period.is_some()
2148    }
2149}
2150
2151/// The criteria of the archive event: the same two changes as the change event, and the
2152/// longest an archiver may go without an event when the value does not move.
2153#[derive(Debug, Clone, Copy, PartialEq, Default)]
2154pub struct AttrArchiveEventConfig {
2155    pub rel_change: Option<f64>,
2156    pub abs_change: Option<f64>,
2157    pub period: Option<Duration>,
2158}
2159
2160impl AttrArchiveEventConfig {
2161    /// Whether the archive event has any criterion, a change or a period.
2162    #[inline]
2163    pub fn is_set(&self) -> bool {
2164        self.rel_change.is_some() || self.abs_change.is_some() || self.period.is_some()
2165    }
2166}
2167
2168/// What a memorized attribute does, when its device is initialized, with the value it
2169/// remembers.
2170///
2171/// Both modes store the value of every write in the database, they only differ in what the
2172/// next initialization of the device does with the value it finds there.
2173#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2174pub enum AttrMemorized {
2175    /// The remembered value is restored as the write value of the attribute and nothing
2176    /// else: a client reading the attribute back finds the set point it left, but the write
2177    /// function is never called, so the device does not act on its hardware only because
2178    /// it was restarted.
2179    RestoreSetPoint,
2180    /// The remembered value is written again, through the write function of the attribute,
2181    /// which puts the device back in the state it was left in.
2182    RestoreAndWrite,
2183}
2184
2185// =========================== //
2186//           Device            //
2187// =========================== //
2188
2189/// Static information about a device: what it is and where it runs.
2190///
2191/// None of this is something a client computes anything from, it is what it shows about a
2192/// device, and all of it is decided by the server rather than by the device itself.
2193#[derive(Debug, Clone, PartialEq, Eq, Default)]
2194pub struct DevInfo {
2195    /// The name of the class the device is an instance of.
2196    pub class: String,
2197    /// The `exe/inst` name of the server running the device, which is what it is
2198    /// registered in the database under.
2199    pub server_id: String,
2200    /// The name of the host the server runs on.
2201    pub server_host: String,
2202    /// The IDL version the server says it implements, which is a statement it makes about
2203    /// itself and decides nothing.
2204    pub server_version: i32,
2205    /// Where the documentation of the class is, when it has any.
2206    pub doc_url: Option<Cow<'static, str>>,
2207    /// What the device is, which is free text a class gives itself, absent for a device
2208    /// whose server is too old to have any.
2209    pub dev_type: Option<Cow<'static, str>>,
2210    /// The versions of whatever the server is built out of, as the pairs of a name and a
2211    /// version that the server chooses to publish. Empty for a server too old to have any,
2212    /// this being the one thing IDL v6 added.
2213    pub version_info: Vec<(Cow<'static, str>, Cow<'static, str>)>,
2214}
2215
2216// =========================== //
2217//        Various enums        //
2218// =========================== //
2219
2220/// All possible device states.
2221#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2222#[repr(u8)]
2223pub enum DevState {
2224    On,
2225    Off,
2226    Close,
2227    Open,
2228    Insert,
2229    Extract,
2230    Moving,
2231    Standby,
2232    Fault,
2233    Init,
2234    Running,
2235    Alarm,
2236    Disable,
2237    Unknown,
2238}
2239
2240/// Source when querying an attribute.
2241#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2242pub enum DevSource {
2243    /// Do not use the polling cache.
2244    Dev,
2245    /// Only use the polling cache.
2246    Cache,
2247    /// First use the polling cache, and fall back without it.
2248    CacheDev,
2249}
2250
2251#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
2252pub enum DisplayLevel {
2253    /// The level everything is at unless it is said otherwise: what a control room shows.
2254    #[default]
2255    Operator,
2256    Expert,
2257}
2258
2259// =========================== //
2260//           Events            //
2261// =========================== //
2262
2263/// The events of a subscription in place, as the backend gives them, of the kind that was
2264/// subscribed to.
2265enum ChannelEvents {
2266    Value(BoxStream<'static, Result<AttrRead<Value>>>),
2267    Ready(BoxStream<'static, Result<AttrReady>>),
2268    Config(BoxStream<'static, Result<AttrInfo>>),
2269}
2270
2271impl ChannelEvents {
2272
2273    /// Poll the readings of an attribute, converted into the type they are expected to be.
2274    fn poll_value<T: FromValue>(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrRead<T>>>> {
2275        let Self::Value(stream) = self else { unreachable!("a value channel subscribes to values") };
2276        stream.poll_next_unpin(cx).map(|item| item.map(|read| read.and_then(AttrRead::parse)))
2277    }
2278
2279    /// Poll an attribute being ready to be read.
2280    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrReady>>> {
2281        let Self::Ready(stream) = self else { unreachable!("a ready channel subscribes to readiness") };
2282        stream.poll_next_unpin(cx)
2283    }
2284
2285    /// Poll the configuration of an attribute.
2286    fn poll_config(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrInfo>>> {
2287        let Self::Config(stream) = self else { unreachable!("a configuration channel subscribes to configurations") };
2288        stream.poll_next_unpin(cx)
2289    }
2290
2291}
2292
2293/// What a [`DevChannel`] subscribes to.
2294#[derive(Debug, Clone, Copy)]
2295enum ChannelKind {
2296    Value(AttrValueEvent),
2297    Ready,
2298    Config,
2299}
2300
2301/// The events of a subscription to a device, which unsubscribes when it is dropped: it is
2302/// a [`Stream`] for an asynchronous caller, and [`Self::wait`] blocks for a synchronous one.
2303///
2304/// It gives the latest event and not every one of them, a subscriber that does not keep up
2305/// skipping to the latest, and never ends. An item that is an error is one event that
2306/// failed, such as the reading the device was to push, or the subscription failing to be
2307/// made or being lost.
2308///
2309/// # Subscribing
2310///
2311/// The subscription is made lazily, the first time the channel is polled, a failure being
2312/// given as an item like any other. A subscriber that wants to know that it is in place
2313/// before it waits for events ensures it, see [`Self::ensure_subscribed_async`].
2314///
2315/// The subscription of a backend lives as long as the connection it was made over, see
2316/// [`BackendDev`], while the device may be restarted, elsewhere even. The channel holds
2317/// the proxy of the device for that, which finds it again the way it does for any
2318/// request: once the subscription is lost, which the backend tells with the very errors a
2319/// request fails with when its connection is lost, the proxy forgets its connection the
2320/// way it does for a request, and the channel makes the subscription again through it.
2321/// The admin device, which subscribing goes through in Tango, is found again along with
2322/// it: its server is the one of the device, which may have moved with it.
2323///
2324/// Every attempt is made on the poll after the one that gave the failure of the previous
2325/// one, so that a subscriber retries as fast as it reads. The events then come back on
2326/// their own, with no initial value either: read the attribute once the channel gave an
2327/// error to know where it stands after it.
2328pub struct DevChannel<T> {
2329    /// The device, which is moved into an attempt at subscribing for as long as it runs,
2330    /// and back once it is over.
2331    dev: Option<DevProxy>,
2332    /// The name of the attribute subscribed to.
2333    name: Arc<str>,
2334    kind: ChannelKind,
2335    /// Takes an event out of the events of the kind subscribed to, see [`ChannelEvents`].
2336    poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
2337    /// The timeout of the device, which bounds the blocking calls of the channel.
2338    timeout: Duration,
2339    state: ChannelState,
2340}
2341
2342enum ChannelState {
2343    /// Giving the events of a subscription in place.
2344    Subscribed(ChannelEvents),
2345    /// Making the subscription, the device being given back with the outcome. The attempt
2346    /// is only made once it is polled.
2347    Subscribing(BoxFuture<'static, (DevProxy, Result<ChannelEvents>)>),
2348}
2349
2350impl ChannelState {
2351
2352    /// An attempt at making the subscription, which goes through the admin device of the
2353    /// device, found again on every attempt, see the [`BackendDev`] subscriptions.
2354    fn subscribing(dev: DevProxy, name: Arc<str>, kind: ChannelKind) -> Self {
2355        Self::Subscribing(Box::pin(async move {
2356            let res = async {
2357                let backend = dev.ensure_backend().await?;
2358                let admin = dev.admin_dev_async().await?;
2359                let admin = admin.ensure_backend().await?;
2360                let res = match kind {
2361                    ChannelKind::Value(event) => backend.subscribe_attr_value(admin, Scalar::from(&*name), event).await.map(ChannelEvents::Value),
2362                    ChannelKind::Ready => backend.subscribe_attr_ready(admin, Scalar::from(&*name)).await.map(ChannelEvents::Ready),
2363                    ChannelKind::Config => backend.subscribe_attr_config(admin, Scalar::from(&*name)).await.map(ChannelEvents::Config),
2364                };
2365                dev.check_backend(res)
2366            }.await;
2367            (dev, res)
2368        }))
2369    }
2370
2371}
2372
2373impl<T> DevChannel<T> {
2374
2375    /// A channel to an attribute of the device, subscribed to once it is polled, its events
2376    /// being taken out of those of the kind with the given function.
2377    fn new(
2378        dev: DevProxy,
2379        name: &str,
2380        kind: ChannelKind,
2381        poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
2382    ) -> Self {
2383        let timeout = dev.timeout();
2384        let name = Arc::<str>::from(name);
2385        let state = ChannelState::subscribing(dev, Arc::clone(&name), kind);
2386        Self { dev: None, name, kind, poll_events, timeout, state }
2387    }
2388
2389    /// Start an attempt at making the subscription again.
2390    fn resubscribe(&mut self) {
2391        let dev = self.dev.take().expect("the device is moved into one attempt at a time");
2392        self.state = ChannelState::subscribing(dev, Arc::clone(&self.name), self.kind);
2393    }
2394
2395    /// Drive the attempt at subscribing to its outcome, if one is running, the next one
2396    /// being made on the next poll if it failed.
2397    fn poll_subscribed(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
2398
2399        let ChannelState::Subscribing(attempt) = &mut self.state else {
2400            return Poll::Ready(Ok(()));
2401        };
2402
2403        let (dev, res) = ready!(attempt.as_mut().poll(cx));
2404        let dev = &*self.dev.insert(dev);
2405
2406        Poll::Ready(match res {
2407            Ok(events) => {
2408                debug!(dev = %dev.inner.target, attr = %self.name, event = ?self.kind, "Subscribed to the events");
2409                self.state = ChannelState::Subscribed(events);
2410                Ok(())
2411            }
2412            Err(err) => {
2413                debug!(dev = %dev.inner.target, attr = %self.name, event = ?self.kind, %err, "Failed to subscribe to the events, retrying on the next poll");
2414                self.resubscribe();
2415                Err(err)
2416            }
2417        })
2418
2419    }
2420
2421    /// Make the subscription if it is not in place, giving the failure of the attempt if
2422    /// it failed: the failure is then not given as an item, and the next attempt is made
2423    /// on the next poll.
2424    pub async fn ensure_subscribed_async(&mut self) -> Result<()> {
2425        poll_fn(|cx| self.poll_subscribed(cx)).await
2426    }
2427
2428    /// Make the subscription if it is not in place, see [`Self::ensure_subscribed_async`],
2429    /// at most for the timeout of the device.
2430    pub fn ensure_subscribed(&mut self) -> Result<()> {
2431        let timeout = self.timeout;
2432        block_on_timeout(self.ensure_subscribed_async(), timeout)?
2433    }
2434
2435    /// Wait for the next event, at most for that long, failing with
2436    /// [`ErrorKind::TimedOut`] when none came.
2437    pub fn wait(&mut self, timeout: Duration) -> Result<T> {
2438        match block_on_timeout(self.next(), timeout)? {
2439            Some(item) => item,
2440            None => unreachable!("a device channel never ends"),
2441        }
2442    }
2443
2444}
2445
2446impl<T> Stream for DevChannel<T> {
2447
2448    type Item = Result<T>;
2449
2450    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
2451
2452        let this = self.get_mut();
2453
2454        loop {
2455
2456            if let Err(err) = ready!(this.poll_subscribed(cx)) {
2457                return Poll::Ready(Some(Err(err)));
2458            }
2459
2460            let ChannelState::Subscribed(events) = &mut this.state else {
2461                unreachable!("subscribed once the attempt succeeded");
2462            };
2463
2464            let dev = this.dev.as_ref().expect("the device is back once subscribed");
2465
2466            match ready!((this.poll_events)(events, cx)) {
2467                Some(Err(err)) if err.kind() == ErrorKind::DevNotConnected => {
2468                    debug!(dev = %dev.inner.target, attr = %this.name, event = ?this.kind, %err, "Lost the subscription to the events, subscribing again on the next poll");
2469                    dev.reset_backend();
2470                    this.resubscribe();
2471                    return Poll::Ready(Some(Err(err)));
2472                }
2473                Some(item) => return Poll::Ready(Some(item)),
2474                // A backend ending its stream without saying why.
2475                None => {
2476                    debug!(dev = %dev.inner.target, attr = %this.name, event = ?this.kind, "The subscription to the events ended, subscribing again");
2477                    this.resubscribe();
2478                }
2479            }
2480
2481        }
2482
2483    }
2484
2485}
2486
2487impl<T> Debug for DevChannel<T> {
2488    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2489        f.debug_struct("DevChannel")
2490            .field("dev", &self.dev)
2491            .field("name", &self.name)
2492            .finish_non_exhaustive()
2493    }
2494}
2495
2496/// Kind of event for a device attribute.
2497#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2498pub enum AttrValueEvent {
2499    Change,
2500    Periodic,
2501    Archive,
2502    User,
2503    Alarm,
2504}
2505
2506/// The failure of a device answering with a number of entries that is not the one it was
2507/// asked for, which a form of a fixed shape has no output to build out of: the device broke
2508/// what the protocol says of it, so the request fails as a whole rather than an entry of it.
2509#[cold]
2510pub(crate) fn unexpected_len_err(expected: usize) -> Error {
2511    Error::new(ErrorKind::Internal)
2512        .with_description(format!("Device answered a different number of entries where {expected} were requested"))
2513}
2514
2515/// The failure of what was being done, described as such, of the kind and severity of the
2516/// failure it comes from, which is its cause: a kind other callers key on, such as the
2517/// reconnection of [`DevProxy`] does, is kept.
2518#[cold]
2519#[track_caller]
2520fn context_err(err: Error, description: impl Into<Cow<'static, str>>) -> Error {
2521    Error::new(err.kind())
2522        .with_description(description)
2523        .with_severity(err.severity())
2524        .with_parent(err)
2525}
2526
2527/// What the failure of a command is turned into: the reasons cppTango answers with are
2528/// mapped onto the kinds of this crate, and an argument type that did not match carries
2529/// the kind that was expected, read back out of the description, see
2530/// [`expected_cpp_tango_type`].
2531fn command_err(err: Error) -> Error {
2532    match err.kind() {
2533        ErrorKind::Reason(reason::API_CommandNotFound) => {
2534            err.with_kind(ErrorKind::CommandNotFound)
2535        }
2536        ErrorKind::Reason(reason::API_IncompatibleCmdArgumentType) => {
2537
2538            let expected_kind = expected_cpp_tango_type(err.description())
2539                .and_then(command_data_kind_from_cpp_tango);
2540
2541            let err = err.with_kind(ErrorKind::CommandArgumentKindInvalid);
2542            match expected_kind {
2543                Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
2544                None => err,
2545            }
2546
2547        }
2548        _ => err,
2549    }
2550}
2551
2552/// What the failure of one attribute write is turned into, see [`command_err`].
2553fn write_attr_err(err: Error) -> Error {
2554    match err.kind() {
2555        ErrorKind::Reason(reason::API_AttrNotFound) => {
2556            err.with_kind(ErrorKind::AttrNotFound)
2557        }
2558        ErrorKind::Reason(reason::API_AttrNotWritable) => {
2559            err.with_kind(ErrorKind::AttrNotWritable)
2560        }
2561        ErrorKind::Reason(reason::API_IncompatibleAttrDataType) => {
2562
2563            let expected_kind = expected_cpp_tango_type(err.description())
2564                .and_then(attr_data_kind_from_cpp_tango);
2565
2566            let err = err.with_kind(ErrorKind::AttrKindInvalid);
2567            match expected_kind {
2568                Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
2569                None => err,
2570            }
2571
2572        }
2573        _ => err,
2574    }
2575}
2576
2577/// Extract the type cppTango says it was expecting out of an error description.
2578///
2579/// The relevant descriptions all name the expected type first, prefixed by the `Tango::`
2580/// namespace, and the type actually found (when they mention it) only after:
2581/// - `Incompatible command argument type, expected type is : Tango::DevBoolean`
2582///   (`Command::throw_bad_type`)
2583/// - `Incompatible attribute type: expected Tango::DevVarDoubleArray (even for single
2584///   value), found Tango::DevVarStringArray` (`WAttribute::check_written_value`)
2585///
2586/// So the first `Tango::`-prefixed word is the expected one, and it's trimmed of any
2587/// trailing punctuation such as the comma or the null byte cppTango's `std::ends` leaves
2588/// behind on some messages.
2589fn expected_cpp_tango_type(description: &str) -> Option<&str> {
2590    description.split_ascii_whitespace()
2591        .find_map(|word| word.strip_prefix("Tango::"))
2592        .map(|ty| ty.trim_matches(|c: char| !c.is_ascii_alphanumeric()))
2593}
2594
2595/// Try to parse the given cpp tango data type into a standard [`CommandDataKind`].
2596///
2597/// The names are the literals given to `Command::throw_bad_type` in cppTango, which match
2598/// the `CmdArgTypeName` table.
2599fn command_data_kind_from_cpp_tango(cpp_type: &str) -> Option<CommandDataKind> {
2600    Some(match cpp_type {
2601        "DevVoid" => CommandDataKind::Void,
2602        "DevBoolean" => CommandDataKind::Bool,
2603        "DevUChar" => CommandDataKind::Octet,
2604        "DevShort" => CommandDataKind::Short,
2605        "DevUShort" => CommandDataKind::UShort,
2606        "DevLong" => CommandDataKind::Long,
2607        "DevULong" => CommandDataKind::ULong,
2608        "DevLong64" => CommandDataKind::Long64,
2609        "DevULong64" => CommandDataKind::ULong64,
2610        "DevFloat" => CommandDataKind::Float,
2611        "DevDouble" => CommandDataKind::Double,
2612        "DevState" => CommandDataKind::State,
2613        "DevString" |
2614        "ConstDevString" => CommandDataKind::String,
2615        "DevVarBooleanArray" => CommandDataKind::BoolArray,
2616        "DevVarCharArray" => CommandDataKind::OctetArray,
2617        "DevVarShortArray" => CommandDataKind::ShortArray,
2618        "DevVarUShortArray" => CommandDataKind::UShortArray,
2619        "DevVarLongArray" => CommandDataKind::LongArray,
2620        "DevVarULongArray" => CommandDataKind::ULongArray,
2621        "DevVarLong64Array" => CommandDataKind::Long64Array,
2622        "DevVarULong64Array" => CommandDataKind::ULong64Array,
2623        "DevVarFloatArray" => CommandDataKind::FloatArray,
2624        "DevVarDoubleArray" => CommandDataKind::DoubleArray,
2625        "DevVarStateArray" => CommandDataKind::StateArray,
2626        "DevVarStringArray" => CommandDataKind::StringArray,
2627        "DevVarLongStringArray" => CommandDataKind::LongStringArray,
2628        "DevVarDoubleStringArray" => CommandDataKind::DoubleStringArray,
2629        "DevEncoded" => CommandDataKind::Encoded,
2630        _ => return None,
2631    })
2632}
2633
2634/// Try to parse the given cpp tango data type into a standard [`AttrDataKind`].
2635///
2636/// An attribute value is always carried as a sequence on the wire, so cppTango always
2637/// names the array type here, even for a scalar attribute; the scalar spellings are
2638/// accepted anyway because [`AttrDataKind`] carries no layout of its own and other
2639/// cppTango versions word it differently.
2640///
2641/// Note that an enumerated attribute is reported as `DevVarShortArray` by cppTango, so it
2642/// is parsed as [`AttrDataKind::Short`]: the labels needed to build
2643/// [`AttrDataKind::Enum`] are simply not in the message.
2644fn attr_data_kind_from_cpp_tango(cpp_type: &str) -> Option<AttrDataKind> {
2645    Some(match cpp_type {
2646        "DevBoolean" | "DevVarBooleanArray" => AttrDataKind::Bool,
2647        "DevUChar" | "DevVarCharArray" => AttrDataKind::Octet,
2648        "DevShort" | "DevVarShortArray" => AttrDataKind::Short,
2649        "DevUShort" | "DevVarUShortArray" => AttrDataKind::UShort,
2650        "DevLong" | "DevVarLongArray" => AttrDataKind::Long,
2651        "DevULong" | "DevVarULongArray" => AttrDataKind::ULong,
2652        "DevLong64" | "DevVarLong64Array" => AttrDataKind::Long64,
2653        "DevULong64" | "DevVarULong64Array" => AttrDataKind::ULong64,
2654        "DevFloat" | "DevVarFloatArray" => AttrDataKind::Float,
2655        "DevDouble" | "DevVarDoubleArray" => AttrDataKind::Double,
2656        "DevState" | "DevVarStateArray" => AttrDataKind::State,
2657        "DevString" | "ConstDevString" | "DevVarStringArray" => AttrDataKind::String,
2658        "DevEncoded" | "DevVarEncodedArray" => AttrDataKind::Encoded,
2659        _ => return None,
2660    })
2661}
2662
2663/// Display the host and port pairs of [`env_databases`] the way they are written in
2664/// `TANGO_HOST`, so that an error naming them can be compared to the variable itself.
2665struct DisplayHostAndPorts<'a>(&'a [(Arc<str>, u16)]);
2666
2667impl Display for DisplayHostAndPorts<'_> {
2668    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2669        for (i, (host, port)) in self.0.iter().enumerate() {
2670            if i != 0 {
2671                f.write_str(", ")?;
2672            }
2673            write!(f, "{host}:{port}")?;
2674        }
2675        Ok(())
2676    }
2677}
2678
2679/// Get the database host/port tuples from the `TANGO_HOST` environment variable.
2680fn env_databases() -> &'static [(Arc<str>, u16)] {
2681    static CACHE: OnceLock<Box<[(Arc<str>, u16)]>> = OnceLock::new();
2682    CACHE.get_or_init(|| {
2683
2684        let mut ret = Vec::new();
2685
2686        if let Ok(var) = env::var("TANGO_HOST") {
2687            for part in var.split(',') {
2688                if let Some((host, port_raw)) = part.rsplit_once(':')
2689                && let Ok(port) = u16::from_str(port_raw) {
2690                    ret.push((host.into(), port));
2691                }
2692            }
2693        }
2694        
2695        ret.into_boxed_slice()
2696
2697    })
2698}
2699
2700
2701#[cfg(test)]
2702mod tests {
2703
2704    use crate::value::Scalar;
2705
2706    use super::*;
2707
2708    /// The property requests of a proxy, awaited within a future boxed as `Send`: this only
2709    /// has to compile.
2710    #[allow(dead_code)]
2711    fn props_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
2712        Box::pin(async move {
2713            let _speed = proxy.get_prop_async::<f64, _>("speed").await?;
2714            let _all = proxy.get_prop_async::<Spectrum<str>, _>(All).await?;
2715            proxy.set_prop_async(("speed", 2.5)).await?;
2716            proxy.delete_prop_async(["speed", "accel"]).await
2717        })
2718    }
2719
2720    /// The requests of a proxy, awaited within a future boxed as `Send`, which is what an
2721    /// asynchronous handler forwarding to another device is: this only has to compile.
2722    #[allow(dead_code)]
2723    fn requests_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
2724        Box::pin(async move {
2725            let _: f64 = proxy.command_async("Add", 1.0).await?;
2726            let _: String = proxy.command_async("Echo", "text").await?;
2727            let _ = proxy.read_attr_async::<f64, _>("speed").await?;
2728            let _ = proxy.read_attr_async::<f64, _>(["speed", "accel"]).await?;
2729            let _ = proxy.read_attr_async::<Value, _>(All).await?;
2730            proxy.write_attr_async(("speed", 2.5)).await?;
2731            let _ = proxy.write_attr_async([("speed", 2.5), ("accel", 1.0)]).await?;
2732            let _ = proxy.write_read_attr_async(("speed", 2.5)).await?;
2733            let _ = proxy.command_info_async("Add").await?;
2734            let _ = proxy.attr_info_async(["speed"]).await?;
2735            let names = proxy.list_prop_async("*").await?;
2736            let _ = proxy.read_attr_async::<f64, _>(&names).await?;
2737            let _ = proxy.attr_info_async(names).await?;
2738            Ok(())
2739        })
2740    }
2741
2742    /// A property request is `Send` whatever the names it is given are, since its future
2743    /// holds nothing of them but the copy it makes of them: this only has to compile.
2744    #[allow(dead_code)]
2745    fn props_of_any_names<'a, Names>(proxy: &'a DevProxy, names: Names) -> BoxFuture<'a, Result<<f64 as GetPropSpec<Names>>::Output>>
2746    where
2747        Names: ToNames + 'a,
2748        f64: GetPropSpec<Names>,
2749    {
2750        Box::pin(proxy.get_prop_async::<f64, Names>(names))
2751    }
2752
2753    #[test]
2754    fn expected_kind() {
2755
2756        // Verbatim from cppTango 'Command::throw_bad_type', trailing null byte included
2757        // because the message is built with 'std::ends'.
2758        let cmd = "Incompatible command argument type, expected type is : Tango::DevVarDoubleStringArray\0";
2759        assert_eq!(expected_cpp_tango_type(cmd), Some("DevVarDoubleStringArray"));
2760        assert_eq!(
2761            expected_cpp_tango_type(cmd).and_then(command_data_kind_from_cpp_tango),
2762            Some(CommandDataKind::DoubleStringArray),
2763        );
2764
2765        // Verbatim from cppTango 'WAttribute::check_written_value', which names the found
2766        // type after the expected one: only the expected one should be picked up.
2767        let attr = "Incompatible attribute type: expected Tango::DevVarDoubleArray (even for single value), found Tango::DevVarStringArray\0";
2768        assert_eq!(expected_cpp_tango_type(attr), Some("DevVarDoubleArray"));
2769        assert_eq!(
2770            expected_cpp_tango_type(attr).and_then(attr_data_kind_from_cpp_tango),
2771            Some(AttrDataKind::Double),
2772        );
2773
2774        assert_eq!(attr_data_kind_from_cpp_tango("DevVarEncodedArray"), Some(AttrDataKind::Encoded));
2775        assert_eq!(command_data_kind_from_cpp_tango("DevEncoded"), Some(CommandDataKind::Encoded));
2776
2777        // Unrelated descriptions and unsupported types should not yield any kind.
2778        assert_eq!(expected_cpp_tango_type("Attribute my_attr is not writable"), None);
2779        assert_eq!(command_data_kind_from_cpp_tango("DevVarEncodedArray"), None);
2780
2781    }
2782
2783    /// Readings decided ahead of time, standing in for the ones a backend would have
2784    /// decoded out of what a device answered.
2785    fn reads<const N: usize>(entries: [(&str, Value); N]) -> Vec<NamedResult<AttrRead<Value>>> {
2786        entries.into_iter()
2787            .map(|(name, value)| Named::new(Scalar::from(name), Ok(AttrRead::new(value))))
2788            .collect()
2789    }
2790
2791    /// One name asks for one reading, which is the output itself: there is no slot around
2792    /// it, so what is given back is that reading and nothing else.
2793    #[test]
2794    fn single_reading() {
2795        let reads = reads([("double_scalar", Value::from(2.5))]);
2796        let read = <f64 as ReadAttrSpec<&str>>::collect(reads).unwrap();
2797        assert_eq!(read.value, Some(2.5));
2798    }
2799
2800    /// The readings come back in the order the names were given, and a form of a fixed
2801    /// shape has one slot per name.
2802    #[test]
2803    fn readings_in_order() {
2804        let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
2805        let [first, second] = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap();
2806        assert_eq!(first.unwrap().value, Some(1.0));
2807        assert_eq!(second.unwrap().value, Some(2.0));
2808    }
2809
2810    /// A tuple of names gives every reading the type of its own component, so a device
2811    /// answering with a double and a string is read into both at once.
2812    #[test]
2813    fn readings_of_their_own_types() {
2814        let reads = reads([("double_scalar", Value::from(2.5)), ("string_scalar", Value::from("hello"))]);
2815        let (double, string) = <(f64, String) as ReadAttrSpec<(&str, &str)>>::collect(reads).unwrap();
2816        assert_eq!(double.unwrap().value, Some(2.5));
2817        assert_eq!(string.unwrap().value.as_deref(), Some("hello"));
2818    }
2819
2820    /// A device answering with fewer readings than there are slots to put them in fails the
2821    /// whole request: the output of a fixed shape cannot be built out of what came back.
2822    #[test]
2823    fn missing_reading() {
2824        let reads = reads([("first", Value::from(1.0))]);
2825        let err = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap_err();
2826        assert_eq!(err.kind(), ErrorKind::Internal);
2827    }
2828
2829    /// Reading every attribute of a device names every reading back, since nothing named
2830    /// them on the way out.
2831    #[test]
2832    fn every_reading_named() {
2833        let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
2834        let all = <Value as ReadAttrSpec<All>>::collect(reads).unwrap();
2835        let names = all.iter().map(|read| read.name.value()).collect::<Vec<_>>();
2836        assert_eq!(names, ["first", "second"]);
2837    }
2838
2839    /// The outcomes of a write come back in the order the values were given, one per value
2840    /// and named by none of them.
2841    #[test]
2842    fn write_outcomes() {
2843        let writes = vec![Ok(()), Err(Error::new(ErrorKind::AttrNotWritable))];
2844        let (first, second) = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap();
2845        assert!(first.is_ok());
2846        assert_eq!(second.unwrap_err().kind(), ErrorKind::AttrNotWritable);
2847    }
2848
2849    /// A device answering for a number of values that is not the one that was written
2850    /// fails the whole request, the same way it does for a read.
2851    #[test]
2852    fn unexpected_outcome() {
2853        let writes = vec![Ok(())];
2854        let err = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap_err();
2855        assert_eq!(err.kind(), ErrorKind::Internal);
2856    }
2857
2858}