rstango 0.4.0

//! The client side: [`DevProxy`], the handle to a device, and what it reads and writes.

use std::sync::{Arc, OnceLock, RwLock};
use std::time::{Duration, SystemTime};
use std::fmt::{self, Debug, Display};
use std::task::{Context, Poll};
use std::hash::{Hash, Hasher};
use std::future::Future;
use std::cmp::Ordering;
use std::str::FromStr;
use std::borrow::Cow;
use std::{env, mem};
use std::pin::Pin;

use futures::stream::{BoxStream, Stream, StreamExt};
use futures::ready;
use tracing::{debug, instrument, warn};
use futures::future::{poll_fn, BoxFuture};

use crate::value::{AttrDataFormat, AttrDataKind, CommandDataKind, FromValue, IntoValue, Scalar, Spectrum, Value};
use crate::database::{prop_params, set_prop_params, DatabaseProxy, GetPropSpec, PropCommands, SetPropSpec};
use crate::backend::{default_backend, Backend, BackendDev};
use crate::error::{Error, ErrorKind, Result, reason};
use crate::name::{ArcTrl, IntoTrl, TrlObject};
use crate::future::block_on_timeout;
use crate::local;


/// The timeout of a new proxy, the same as cppTango's, see [`DevProxy::set_timeout`].
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);

/// The statically known device name for the database device in its server.
pub(crate) const DATABASE_DEV_NAME: &str = "database";

/// A device proxy is an object that allows querying a remote device.
/// 
/// Two device proxies are equal, and hash the same, when they use the same backend (compared
/// by pointer) and have the same internal target, regardless of their `trl`, their connection
/// or their configuration.
#[derive(Debug)]
pub struct DevProxy {
    /// The inner structure, boxed to reduce exposed size.
    inner: Box<ProxyInner>,
    /// The state of this proxy, behind a read-write lock.
    state: RwLock<ProxyState>,
}

#[derive(Debug, Clone)]
struct ProxyInner {
    // The backend, shared with a reference-counted pointer.
    backend: Arc<dyn Backend>,
    /// The source configured for this proxy.
    source: DevSource,
    /// The timeout configured for this proxy.
    timeout: Duration,
    /// The real target for connection of this proxy.
    target: ProxyTarget,
}

#[derive(Debug, Clone)]
struct ProxyState {
    /// The actual backend in use, none if the last call errored.
    backend: Option<Arc<dyn BackendDev>>,
    /// In case the target is directly a database, this contains the host and port of the
    /// database connected to, this is used to construct the TRL.
    database_host_and_port: Option<(Arc<str>, u16)>,
}

/// Target for a device proxy, this is used so that the DevProxy can also be responsible
/// for connecting to a database as well as regular devices. The device name might be an
/// alias!
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) enum ProxyTarget {
    /// Database device from environment.
    DatabaseFromEnv,
    /// Database device from host and port.
    DatabaseAt {
        host: Arc<str>,
        port: u16,
    },
    /// Regular device from a database.
    RegularFromDatabase {
        database: DatabaseProxy,
        name: Arc<str>,
    },
    /// A regular device accessible directly from a given service on a host, no database
    /// is involved!
    RegularAt {
        host: Arc<str>,
        port: u16,
        name: Arc<str>,
    },
}

impl DevProxy {

    /// Same as [`Self::try_new`] but panics if the TRL is invalid for a device proxy.
    pub fn new<'a>(trl: impl IntoTrl<'a>) -> Self {
        Self::new_with_backend(default_backend(), trl)
    }

    /// Create a new device proxy located by the given Tango Resource Locator, the given
    /// TRL should not target an attribute or property, or else this function will return
    /// None.
    pub fn try_new<'a>(trl: impl IntoTrl<'a>) -> Option<Self> {
        Self::try_new_with_backend(default_backend(), trl)
    }

    /// Create a new device proxy from a database and the device's name.
    pub fn new_from_database(database: &DatabaseProxy, name: &str) -> Self {
        Self::new_from_database_with_backend(default_backend(), database, name)
    }

    /// Same as [`Self::try_new_with_backend`] but panics if the TRL is invalid for a device
    /// proxy.
    pub fn new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Self {
        Self::try_new_with_backend(backend, trl).expect("The given TRL does not target a device")
    }

    /// Create a new device proxy located by the given Tango Resource Locator, the given
    /// TRL should not target an attribute or property, or else this function will return
    /// None.
    pub fn try_new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Option<Self> {

        let trl = trl.into_trl().ok()?;

        // We can't target a property!
        if trl.property().is_some() {
            return None;
        }

        let name = match trl.object() {
            TrlObject::Alias(name) => name,
            TrlObject::Dev(name) => name,
            TrlObject::Attr(_, _) => return None,
        };

        let target = match (trl.db(), trl.host_and_port()) {
            (true, Some((host, port))) => ProxyTarget::RegularFromDatabase { 
                database: DatabaseProxy::new_with_backend(Arc::clone(&backend), host, port),
                name: Arc::from(name),
            },
            (true, None) => ProxyTarget::RegularFromDatabase { 
                database: DatabaseProxy::new_from_env_with_backend(Arc::clone(&backend)), 
                name: Arc::from(name),
            },
            (false, Some((host, port))) => ProxyTarget::RegularAt { 
                host: Arc::from(host), 
                port, 
                name: Arc::from(name),
            },
            (false, None) => return None,
        };

        Some(Self::new_from_target_with_backend(backend, target))
        
    }

    /// Create a new device proxy from a database and the device's name, with backend.
    pub fn new_from_database_with_backend(backend: Arc<dyn Backend>, database: &DatabaseProxy, name: &str) -> Self {
        Self::new_from_target_with_backend(backend, ProxyTarget::RegularFromDatabase { 
            database: database.clone(), 
            name: Arc::from(name),
        })
    }

    /// Internal function to create a device proxy for the given target, with backend.
    pub(crate) fn new_from_target_with_backend(backend: Arc<dyn Backend>, target: ProxyTarget) -> Self {
        Self {
            inner: Box::new(ProxyInner {
                backend,
                target,
                source: DevSource::CacheDev,
                timeout: DEFAULT_TIMEOUT,
            }),
            state: RwLock::new(ProxyState {
                backend: None,
                database_host_and_port: None,
            }),
        }
    }

    pub fn backend(&self) -> &Arc<dyn Backend> {
        &self.inner.backend
    }

    /// Retrieve the timeout duration currently used for all requests.
    pub fn timeout(&self) -> Duration {
        self.inner.timeout
    }

    /// Set the timeout duration and return the previous one. See [`Self::timeout`].
    pub fn set_timeout(&mut self, timeout: Duration) -> Duration {
        mem::replace(&mut self.inner.timeout, timeout)
    }

    /// Retrieve the source of data currently used for attribute readings and commands.
    pub fn source(&self) -> DevSource {
        self.inner.source
    }

    /// Set the source of data and return the previous one. See [`Self::source`].
    pub fn set_source(&mut self, source: DevSource) -> DevSource {
        mem::replace(&mut self.inner.source, source)
    }

    /// Internal function to initialize a brand new connection to this device proxy.
    /// 
    /// Boxed, because connecting a device goes through the database, itself a device
    /// connected by this very function.
    fn connect_backend(&self) -> BoxFuture<'_, Result<(Arc<dyn BackendDev>, Option<(Arc<str>, u16)>)>> {
        Box::pin(async move {

            match self.inner.target {
                ProxyTarget::DatabaseFromEnv => {

                    debug!("Connecting to the database of TANGO_HOST");

                    let databases = env_databases();
                    if databases.is_empty() {
                        return Err(match env::var("TANGO_HOST") {
                            Ok(var) => Error::new(ErrorKind::TangoHostNotSet)
                                .with_description("TANGO_HOST is not a comma separated list of host:port")
                                .with_append_description(format_args!("Got: {var:?}")),
                            Err(_) => Error::new(ErrorKind::TangoHostNotSet)
                                .with_description("TANGO_HOST is not defined in the environment"),
                        });
                    }

                    let mut last_err = None;
                    for (host, port) in databases {

                        if let Some(dev_backend) = local::find_at(host, *port, DATABASE_DEV_NAME) {
                            debug!(%host, port, "Connected to the database, served by this process");
                            return Ok((dev_backend, Some((host.clone(), *port))));
                        }

                        match self.inner.backend.connect_at(host, *port, DATABASE_DEV_NAME).await {
                            Ok(dev_backend) => {
                                debug!(%host, port, "Connected to the database");
                                return Ok((dev_backend, Some((host.clone(), *port))));
                            }
                            Err(err) => {
                                warn!(%host, port, %err, "Failed to connect to the database, trying the next one");
                                last_err = Some(err);
                            }
                        }
                    }

                    // Every database in TANGO_HOST has been tried, the last failure being the
                    // one kept as the cause: naming them all is what tells the reader that the
                    // other ones were tried too.
                    let mut err = Error::new(ErrorKind::DatabaseNotConnected)
                        .with_description("Failed to connect to any database")
                        .with_append_description(format_args!("Tried, from TANGO_HOST: {}", DisplayHostAndPorts(databases)));
                    if let Some(last_err) = last_err {
                        err.set_parent(last_err);
                    }
                    Err(err)

                }
                ProxyTarget::DatabaseAt { ref host, port } => {
                
                    debug!(%host, port, "Connecting to the database");

                    if let Some(dev_backend) = local::find_at(host, port, DATABASE_DEV_NAME) {
                        debug!(%host, port, "Connected to the database, served by this process");
                        return Ok((dev_backend, Some((host.clone(), port))));
                    }

                    match self.inner.backend.connect_at(host, port, DATABASE_DEV_NAME).await {
                        Ok(dev_backend) => {
                            debug!(%host, port, "Connected to the database");
                            Ok((dev_backend, Some((host.clone(), port))))
                        }
                        Err(err) => Err(Error::new(ErrorKind::DatabaseNotConnected)
                            .with_description("Failed to connect to the database")
                            .with_append_description(format_args!("Tried: {host}:{port}"))
                            .with_parent(err)),
                    }

                }
                ProxyTarget::RegularFromDatabase { ref database, ref name } => {

                    debug!(dev = %name, "Connecting to the device, through the database");
                    
                    let info = match database.import_dev_async(name).await {
                        Ok(info) if !info.exported =>
                            return Err(Error::new(ErrorKind::DevNotExported)
                                .with_description(format!("Device {name:?} is defined in the database but its server {:?} does not export it",
                                    info.server))),
                        Ok(info) => info,
                        Err(err) if let ErrorKind::DevNotConnected = err.kind() =>
                            return Err(Error::new(ErrorKind::DatabaseNotConnected)
                                .with_description(format!("Failed to import device {name:?} from the database"))
                                .with_parent(err)),
                        Err(err) => return Err(err),
                    };

                    // A device this very process serves is reached directly, whatever the
                    // backend, the reference telling it apart from any other of that name.
                    if let Some(dev_backend) = local::find(info.itr.value()) {
                        debug!(dev = %name, "Connected to the device, served by this process");
                        return Ok((dev_backend, None));
                    }

                    // The kind is the one the backend found, 'with_backend' keying the
                    // reconnection on it, the failure of the backend being the cause.
                    match self.inner.backend.connect(info.itr.value()).await {
                        Ok(dev_backend) => {
                            debug!(dev = %name, "Connected to the device");
                            Ok((dev_backend, None))
                        }
                        Err(err) => Err(context_err(err, format!(
                            "Failed to connect to device {name:?} of server {:?}, exported on host {:?}",
                            info.server, info.host))),
                    }

                }
                ProxyTarget::RegularAt { ref host, port, ref name } => {
                
                    debug!(%host, port, dev = %name, "Connecting to the device");

                    if let Some(dev_backend) = local::find_at(host, port, name) {
                        debug!(%host, port, dev = %name, "Connected to the device, served by this process");
                        return Ok((dev_backend, None));
                    }

                    match self.inner.backend.connect_at(host, port, name).await {
                        Ok(dev_backend) => {
                            debug!(%host, port, dev = %name, "Connected to the device");
                            Ok((dev_backend, None))
                        }
                        Err(err) => Err(context_err(err, format!(
                            "Failed to connect to device {name:?} at {host}:{port}"))),
                    }

                }
            }

        })
    }

    /// Give back the backend this proxy reaches its device through, connecting it if that
    /// has not happened yet.
    ///
    /// The connection is awaited with no lock of the proxy held: a guard cannot travel
    /// across an await, and keeping one here would hold every other reader of the proxy
    /// for the whole of a round trip. Two callers landing here at once therefore both
    /// connect, and the one that stores its backend first is the one that is kept, the
    /// other connection being dropped on the spot.
    async fn ensure_backend(&self) -> Result<Arc<dyn BackendDev>> {

        if let Some(backend) = self.state.read().unwrap().backend.as_ref() {
            return Ok(Arc::clone(backend));
        }

        let (backend, database_host_and_port) = self.connect_backend().await?;

        let mut state = self.state.write().unwrap();
        match state.backend {
            Some(ref backend) => Ok(Arc::clone(backend)),
            None => {
                state.backend = Some(Arc::clone(&backend));
                state.database_host_and_port = database_host_and_port;
                Ok(backend)
            }
        }

    }

    fn reset_backend(&self) {
        let mut state = self.state.write().unwrap();
        state.backend = None;
        state.database_host_and_port = None;
    }

    fn check_backend<T>(&self, res: Result<T>) -> Result<T> {
        if let Err(err) = &res && err.kind() == ErrorKind::DevNotConnected {
            self.reset_backend();
        }
        res
    }

    pub(crate) fn block_on<F: Future>(&self, fut: F) -> Result<F::Output> {
        block_on_timeout(fut, self.inner.timeout)
    }

    fn database_host_and_port(&self) -> Result<Option<(Arc<str>, u16)>> {
        self.block_on(self.database_host_and_port_async())?
    }

    /// The host and port of the database this proxy reached its device through, connecting
    /// it if that has not happened yet, see [`Self::database_host_and_port`].
    pub(crate) async fn database_host_and_port_async(&self) -> Result<Option<(Arc<str>, u16)>> {
        self.ensure_backend().await?;
        Ok(self.state.read().unwrap().database_host_and_port.clone())
    }

    /// Return the [`DatabaseProxy`] this device has been requested from, or this device as a
    /// database if it is one. A device reached without a database fails with
    /// [`ErrorKind::DatabaseNeeded`].
    pub fn database(&self) -> Result<&DatabaseProxy> {
        match self.inner.target {
            ProxyTarget::DatabaseFromEnv |
            ProxyTarget::DatabaseAt { .. } => 
                Ok(DatabaseProxy::from_dev(self)),
            ProxyTarget::RegularFromDatabase { ref database, .. } => 
                Ok(database),
            ProxyTarget::RegularAt { ref host, port, ref name } => 
                Err(Error::new(ErrorKind::DatabaseNeeded)
                    .with_description(format!("Device {name:?} is reached at {host}:{port} rather than through a database"))),
        }
    }

    /// Resolve this device's fully qualified device name.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn trl(&self) -> Result<ArcTrl> {
        match self.inner.target {
            ProxyTarget::DatabaseFromEnv => {
                let name = self.name()?;
                let (host, port) = self.database_host_and_port()?
                    .expect("This should be defined if the target is a database");
                Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
            }
            ProxyTarget::DatabaseAt { ref host, port } => {
                let name = self.name()?;
                Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
            }
            ProxyTarget::RegularFromDatabase { ref database, ref name } => {
                let (host, port) = database.database_host_and_port()?
                    .expect("This should be defined if the target is a database");
                Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
            }
            ProxyTarget::RegularAt { ref host, port, ref name } => {
                Ok(format!("tango://{host}:{port}/{name}#dbase=no").parse::<ArcTrl>().unwrap())
            }
        }
    }

    /// Connect this proxy to its device if that has not happened yet, so that its clones
    /// share that connection instead of each making its own.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn connect(&self) -> Result<()> {
        self.block_on(self.connect_async())?
    }

    /// Connect this proxy to its device if that has not happened yet, see [`Self::connect`].
    pub async fn connect_async(&self) -> Result<()> {
        self.ensure_backend().await.map(drop)
    }

    /// Resolve this device's name.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn name(&self) -> Result<Arc<str>> {
        self.block_on(self.name_async())?
    }

    /// Resolve this device's name, see [`Self::name`].
    pub async fn name_async(&self) -> Result<Arc<str>> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.name().await)
    }

    /// Resolve this device's admin device name.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn admin_dev_name(&self) -> Result<Arc<str>> {
        self.block_on(self.admin_dev_name_async())?
    }

    /// Resolve this device's admin device name, see [`Self::admin_dev_name`].
    pub async fn admin_dev_name_async(&self) -> Result<Arc<str>> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.admin_dev_name().await)
    }

    /// Resolve this device's admin device and return it!
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn admin_dev(&self) -> Result<DevProxy> {
        Ok(self.admin_dev_of(self.admin_dev_name()?))
    }

    /// Resolve this device's admin device and return it, see [`Self::admin_dev`].
    pub async fn admin_dev_async(&self) -> Result<DevProxy> {
        Ok(self.admin_dev_of(self.admin_dev_name_async().await?))
    }

    /// The proxy to the admin device of this one, from the name of it: asking the device
    /// for that name is the only part of this that reaches it, so it is what the two
    /// methods above do and this is everything they share.
    fn admin_dev_of(&self, name: Arc<str>) -> DevProxy {

        let backend = Arc::clone(&self.inner.backend);

        DevProxy::new_from_target_with_backend(backend, match &self.inner.target {
            ProxyTarget::DatabaseFromEnv |
            ProxyTarget::DatabaseAt { .. } => {
                // Self can be a DatabaseProxy, so we use it!
                ProxyTarget::RegularFromDatabase { 
                    database: DatabaseProxy::with_dev(self.clone()), 
                    name,
                }
            }
            ProxyTarget::RegularFromDatabase { database, .. } => {
                ProxyTarget::RegularFromDatabase { 
                    database: database.clone(), 
                    name,
                }
            }
            ProxyTarget::RegularAt { host, port, .. } => {
                ProxyTarget::RegularAt { 
                    host: Arc::clone(host), 
                    port: *port, 
                    name,
                }
            }
        })

    }

    /// Get what the device says it is: the class it is an instance of, the server running
    /// it and where, and the versions of whatever that server is built out of.
    ///
    /// All of it is the server's own description of itself and none of it is asked of the
    /// database, so it answers for a device served without one.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn info(&self) -> Result<DevInfo> {
        self.block_on(self.info_async())?
    }

    /// Get what the device says it is, see [`Self::info`].
    pub async fn info_async(&self) -> Result<DevInfo> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.info().await)
    }

    /// Ping this device.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn ping(&self) -> Result<()> {
        self.block_on(self.ping_async())?
    }

    /// Ping this device, see [`Self::ping`].
    pub async fn ping_async(&self) -> Result<()> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.ping().await)
    }

    /// Get the device state.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn state(&self) -> Result<DevState> {
        self.block_on(self.state_async())?
    }

    /// Get the device state, see [`Self::state`].
    pub async fn state_async(&self) -> Result<DevState> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.state().await)
    }

    /// Get the device status.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn status(&self) -> Result<Scalar<str>> {
        self.block_on(self.status_async())?
    }

    /// Get the device status, see [`Self::status`].
    pub async fn status_async(&self) -> Result<Scalar<str>> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.status().await)
    }

    /// Initialize the device again, executing its `Init` command.
    pub fn init(&self) -> Result<()> {
        self.command("Init", ())
    }

    /// Initialize the device again, see [`Self::init`].
    pub fn init_async(&self) -> impl Future<Output = Result<()>> + Send {
        self.command_async("Init", ())
    }

    /// Request the device black box.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub fn black_box(&self, max: Option<usize>) -> Result<Spectrum<str>> {
        self.block_on(self.black_box_async(max))?
    }

    /// Request the device black box, see [`Self::black_box`].
    pub async fn black_box_async(&self, max: Option<usize>) -> Result<Spectrum<str>> {
        let backend = self.ensure_backend().await?;
        self.check_backend(backend.black_box(max).await)
    }

    /// Execute a remote command, giving back what it answered with, converted into the type
    /// it is expected to be, see [`FromValue`].
    ///
    /// The value shares the octets the device answered with, so asking for a [`Value`] or
    /// for one of the buffers it holds — a [`crate::value::Spectrum`] of a million points
    /// above all — copies nothing, where the usual owned types are copied out of it.
    #[instrument(skip(self, input), fields(target = %self.inner.target))]
    pub fn command_async<I, O>(&self, name: &str, input: I) -> impl Future<Output = Result<O>> + Send
    where
        I: IntoValue,
        O: FromValue,
    {

        // TODO: Support for API_CommandTimedOut
        //   It happens when we execute a command on a server with device-level
        //   serialization, which prevents it from executing our command because another
        //   action is holding the lock on the device monitor thread (another long
        //   command or attribute).

        let command = Scalar::from(name);
        let input = input.into_value();
        async move {
            let backend = self.ensure_backend().await?;
            let res = backend.command(command, input, self.inner.source).await;
            O::from_value(self.check_backend(res).map_err(command_err)?)
        }

    }

    /// Execute a remote command, giving back what it answered with, see
    /// [`Self::command_async`].
    pub fn command<I, O>(&self, name: &str, input: I) -> Result<O>
    where
        I: IntoValue,
        O: FromValue,
    {
        self.block_on(self.command_async(name, input))?
    }

    /// Read attributes given a specification of attributes. This specification is really
    /// flexible and allows various signatures for the function:
    /// 1. `read_attr<T>(&str) -> Result<AttrRead<T>>`
    /// 2. `read_attr<T>(&[&str]) -> Result<Vec<Result<AttrRead<T>>>>`
    /// 3. `read_attr<T>([&str; N]) -> Result<[Result<AttrRead<T>>; N]>`
    /// 4. `read_attr<(A, B, ...)>((&str, &str, ...)) -> Result<(Result<AttrRead<A>>, Result<AttrRead<B>>, ...)>`
    /// 5. `read_attr<T>(All) -> Result<Vec<NamedResult<AttrRead<T>>>>`
    ///
    /// Forms 1 to 3 and 5 convert every attribute into the same type, whereas form 4 gives
    /// each attribute its own type. Nothing in the names constrains the specification, so
    /// it has to come either from a turbofish, leaving the names inferred as in
    /// `read_attr::<i16, _>("enum_scalar")`, or from the type of the binding it is assigned
    /// to. Reading into a [`Value`] keeps whatever the device answered with, and reading
    /// into a buffer copies nothing, see [`Self::command_async`].
    ///
    /// Each reading comes with the [`Result`] of that one reading, so that a per-attribute
    /// failure is seen without the whole request failing. Note that, unlike the
    /// many-attributes forms, the single-attribute form has no per-attribute slot in its
    /// output and therefore returns that attribute's error directly. The fixed-size forms
    /// 3 and 4 additionally fail with [`ErrorKind::Internal`] if the device does not answer
    /// with exactly as many readings as were requested.
    ///
    /// All requested attributes are read within a single request, which matters beyond
    /// saving round trips: if the device enforces serialization, it is held for the whole
    /// batch and only released at the end, so the readings are consistent with each other
    /// instead of being sampled at instants another client could have written between.
    #[instrument(skip(self, names), fields(target = %self.inner.target))]
    pub fn read_attr_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
    where
        Spec: ReadAttrSpec<Names>,
    {
        let names = names.to_names();
        async move {
            let backend = self.ensure_backend().await?;
            let res = backend.read_attr(names, self.inner.source).await;
            Spec::collect(self.check_backend(res)?)
        }
    }

    /// Read attributes, see [`Self::read_attr_async`].
    pub fn read_attr<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
    where
        Spec: ReadAttrSpec<Names>,
    {
        self.block_on(self.read_attr_async::<Spec, Names>(names))?
    }

    /// Write attributes given a specification of attributes and their values, giving back
    /// the outcome of each individual write, in the order they were given. In Tango a write
    /// only carries a value: quality and timestamp cannot be written. Like for
    /// [`Self::read_attr`], this specification is really flexible and allows various
    /// signatures for the function:
    /// 1. `write_attr((&str, T)) -> Result<()>`
    /// 2. `write_attr(Vec<(&str, T)>) -> Result<Vec<Result<()>>>`
    /// 3. `write_attr([(&str, T); N]) -> Result<[Result<()>; N]>`
    /// 4. `write_attr(((&str, A), (&str, B), ...)) -> Result<(Result<()>, Result<()>, ...)>`
    ///
    /// Each written value only needs to implement [`IntoValue`], so unlike for
    /// [`Self::read_attr`] the value types are inferred from the specification and never
    /// have to be named. Note that, unlike the many-attributes forms, the single-attribute
    /// form has no per-attribute slot in its output and therefore returns that attribute's
    /// error directly. The fixed-size forms 3 and 4 fail with [`ErrorKind::Internal`] if
    /// the device does not answer for exactly as many attributes as were written.
    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
    pub fn write_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send
    where
        Spec: WriteAttrSpec<'a>,
    {

        let values = spec.into_values();

        async move {

            let backend = self.ensure_backend().await?;
            let res = backend.write_attr(values).await;

            // A refused write is given the kind that fits it rather than the reason code
            // the device raised with, the same way the request failing as a whole is.
            let outcomes = self.check_backend(res)?.into_iter()
                .map(|res| res.map_err(write_attr_err))
                .collect();

            Spec::collect(outcomes)

        }

    }

    /// Write attributes, see [`Self::write_attr_async`].
    pub fn write_attr<'a, Spec>(&self, spec: Spec) -> Result<Spec::Output>
    where
        Spec: WriteAttrSpec<'a>,
    {
        self.block_on(self.write_attr_async(spec))?
    }

    /// Write attributes and read them back, given the same flexible specification as
    /// [`Self::write_attr`], but returning the read-back values instead of unit results.
    /// Each value is converted back into the very type that was written:
    /// 1. `write_read_attr((&str, T)) -> Result<AttrRead<T>>`
    /// 2. `write_read_attr(Vec<(&str, T)>) -> Result<Vec<Result<AttrRead<T>>>>`
    /// 3. `write_read_attr([(&str, T); N]) -> Result<[Result<AttrRead<T>>; N]>`
    /// 4. `write_read_attr(((&str, A), (&str, B), ...)) -> Result<(Result<AttrRead<A>>, Result<AttrRead<B>>, ...)>`
    ///
    /// The read is done by the server right after the write, so this saves a round trip
    /// compared to a [`Self::write_attr`] followed by a [`Self::read_attr`], and no other
    /// request can interleave between the two when the device enforces serialization.
    ///
    /// Because the read-back is converted back into the written type, each value type must
    /// also implement [`FromValue`] here, whereas [`Self::write_attr`] only requires
    /// [`IntoValue`].
    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
    pub fn write_read_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>> + Send
    where
        Spec: WriteAttrSpec<'a>,
        Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
    {
        let values = spec.into_values();
        async move {
            let backend = self.ensure_backend().await?;
            let res = backend.write_read_attr(values).await;
            Spec::ReadSpec::collect(self.check_backend(res)?)
        }
    }

    /// Write attributes and read them back, see [`Self::write_read_attr_async`].
    pub fn write_read_attr<'a, Spec>(&self, spec: Spec) -> Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>
    where
        Spec: WriteAttrSpec<'a>,
        Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
    {
        self.block_on(self.write_read_attr_async(spec))?
    }

    /// Get the static information of commands, given the same flexible specification of
    /// names as [`Self::read_attr`] takes:
    /// 1. `command_info(&str) -> Result<CommandInfo>`
    /// 2. `command_info(&[&str]) -> Result<Vec<Result<CommandInfo>>>`
    /// 3. `command_info([&str; N]) -> Result<[Result<CommandInfo>; N]>`
    /// 4. `command_info(All) -> Result<Vec<NamedResult<CommandInfo>>>`
    ///
    /// Every command comes with the [`Result`] of that one command, an unknown command name
    /// and a data type that no [`CommandDataKind`] has a variant for being the two the
    /// device answers with.
    ///
    /// Note that, unlike the attributes, the Tango interface has no operation taking a list
    /// of command names, so asking for several of them costs one request each, where [`All`]
    /// only costs one for the whole listing.
    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
    pub fn command_info_async<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
        let names = spec.to_names();
        async move {
            let backend = self.ensure_backend().await?;
            let res = backend.command_info(names).await;
            Spec::collect(self.check_backend(res)?)
        }
    }

    /// Get the static information of commands, see [`Self::command_info_async`].
    pub fn command_info<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
        self.block_on(self.command_info_async(spec))?
    }

    /// Get the static information of attributes, their runtime-modifiable configuration
    /// included, given the same flexible specification of names as [`Self::read_attr`]
    /// takes:
    /// 1. `attr_info(&str) -> Result<AttrInfo>`
    /// 2. `attr_info(&[&str]) -> Result<Vec<Result<AttrInfo>>>`
    /// 3. `attr_info([&str; N]) -> Result<[Result<AttrInfo>; N]>`
    /// 4. `attr_info(All) -> Result<Vec<NamedResult<AttrInfo>>>`
    ///
    /// Every attribute comes with the [`Result`] of that one attribute, a data type that no
    /// [`AttrDataKind`] has a variant for being what the device answers with. Unlike the
    /// commands, an unknown attribute name may fail the whole request, all of them travelling
    /// in a single one, depending on the backend.
    #[instrument(skip(self, spec), fields(target = %self.inner.target))]
    pub fn attr_info_async<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
        let names = spec.to_names();
        async move {
            let backend = self.ensure_backend().await?;
            let res = backend.attr_info(names).await;
            Spec::collect(self.check_backend(res)?)
        }
    }

    /// Get the static information of attributes, see [`Self::attr_info_async`].
    pub fn attr_info<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
        self.block_on(self.attr_info_async(spec))?
    }

    /// Change the configuration of the given attributes, giving back the outcome of each
    /// individual change, in the order they were given.
    ///
    /// Only the configuration is writable, which is the part of an [`AttrInfo`] that a
    /// client is allowed to set: what an attribute *is* — its mode, its data kind and its
    /// format — is what its definition says and is never written from the outside.
    ///
    /// The whole configuration of an attribute goes out every time, so changing one
    /// property of it is reading the configuration with [`Self::attr_info`] and writing
    /// back what was read with that one property changed, which is what the protocol
    /// carries.
    ///
    /// A field with no value brings it back to what the attribute inherits from its class,
    /// if anything.
    #[instrument(skip(self, values), fields(target = %self.inner.target))]
    pub async fn set_attr_config_async(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
        let configs = values.iter()
            .map(|(name, config)| Named::new(Scalar::from(*name), config.clone()))
            .collect();
        let backend = self.ensure_backend().await?;
        let res = backend.set_attr_config(configs).await;
        self.check_backend(res).map_err(write_attr_err)
    }

    /// Change the configuration of the given attributes, see
    /// [`Self::set_attr_config_async`].
    pub fn set_attr_config(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
        self.block_on(self.set_attr_config_async(values))?
    }

    /// The recorded history of a command, oldest entry first and at most `max` of them:
    /// what the polling of the device recorded, or the device itself, which may manage the
    /// history of a command it does not poll. Every entry is what the command answered
    /// then, converted into the type it is expected to be, a run that failed carrying the
    /// failure it failed with, see [`Recorded`].
    ///
    /// A command nothing was recorded of has no history to give, which is a failure and
    /// not an empty history, see [`ErrorKind::CommandNotPolled`] and
    /// [`ErrorKind::AttrNotPolled`].
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub async fn command_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<T>>> {
        let backend = self.ensure_backend().await?;
        let res = backend.command_history(Scalar::from(name), max).await;
        Ok(self.check_backend(res).map_err(command_err)?.into_iter()
            .map(|entry| entry.map(|value| value.and_then(T::from_value)))
            .collect())
    }

    /// The recorded history of a command, see [`Self::command_history_async`].
    pub fn command_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<T>>> {
        self.block_on(self.command_history_async(name, max))?
    }

    /// The recorded history of an attribute, see [`Self::command_history_async`], a reading
    /// carrying the time it was recorded at as its own.
    #[instrument(skip(self), fields(target = %self.inner.target))]
    pub async fn attr_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<AttrRead<T>>>> {
        let backend = self.ensure_backend().await?;
        let res = backend.attr_history(Scalar::from(name), max).await;
        Ok(self.check_backend(res)?.into_iter()
            .map(|entry| entry.map(|read| read.and_then(AttrRead::parse)))
            .collect())
    }

    /// The recorded history of an attribute, see [`Self::attr_history_async`].
    pub fn attr_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<AttrRead<T>>>> {
        self.block_on(self.attr_history_async(name, max))?
    }

    // The polling of a device is its server's, so everything below asks the admin device
    // of its server, which writes what it changes to the database along, so that the
    // device is polled the same once restarted.

    /// Poll that attribute of this device at that period, or at another one if it is
    /// polled already, see the `AddObjPolling` and `UpdObjPollingPeriod` commands of the
    /// admin device.
    pub async fn poll_attr_async(&self, name: &str, period: Duration) -> Result<()> {
        self.poll_async("attribute", name, period).await
    }

    /// Poll that attribute of this device, see [`Self::poll_attr_async`].
    pub fn poll_attr(&self, name: &str, period: Duration) -> Result<()> {
        self.block_on(self.poll_attr_async(name, period))?
    }

    /// Poll that command of this device, see [`Self::poll_attr_async`]: only a command
    /// taking no argument can be polled.
    pub async fn poll_command_async(&self, name: &str, period: Duration) -> Result<()> {
        self.poll_async("command", name, period).await
    }

    /// Poll that command of this device, see [`Self::poll_command_async`].
    pub fn poll_command(&self, name: &str, period: Duration) -> Result<()> {
        self.block_on(self.poll_command_async(name, period))?
    }

    /// Stop polling that attribute of this device, failing if it is not polled, see the
    /// `RemObjPolling` command of the admin device.
    pub async fn stop_poll_attr_async(&self, name: &str) -> Result<()> {
        self.stop_poll_async("attribute", name).await
    }

    /// Stop polling that attribute of this device, see [`Self::stop_poll_attr_async`].
    pub fn stop_poll_attr(&self, name: &str) -> Result<()> {
        self.block_on(self.stop_poll_attr_async(name))?
    }

    /// Stop polling that command of this device, see [`Self::stop_poll_attr_async`].
    pub async fn stop_poll_command_async(&self, name: &str) -> Result<()> {
        self.stop_poll_async("command", name).await
    }

    /// Stop polling that command of this device, see [`Self::stop_poll_command_async`].
    pub fn stop_poll_command(&self, name: &str) -> Result<()> {
        self.block_on(self.stop_poll_command_async(name))?
    }

    /// The period that attribute of this device is polled at, absent if it is not polled,
    /// out of its polling status, see [`Self::polling_status_async`].
    pub async fn attr_poll_period_async(&self, name: &str) -> Result<Option<Duration>> {
        Ok(status_period(&self.polling_status_async().await?, "attribute", name))
    }

    /// The period that attribute of this device is polled at, see
    /// [`Self::attr_poll_period_async`].
    pub fn attr_poll_period(&self, name: &str) -> Result<Option<Duration>> {
        self.block_on(self.attr_poll_period_async(name))?
    }

    /// The period that command of this device is polled at, see
    /// [`Self::attr_poll_period_async`].
    pub async fn command_poll_period_async(&self, name: &str) -> Result<Option<Duration>> {
        Ok(status_period(&self.polling_status_async().await?, "command", name))
    }

    /// The period that command of this device is polled at, see
    /// [`Self::command_poll_period_async`].
    pub fn command_poll_period(&self, name: &str) -> Result<Option<Duration>> {
        self.block_on(self.command_poll_period_async(name))?
    }

    /// The polling status of this device, one text per object polled, as its admin device
    /// gives it with its `DevPollStatus` command, none for a device that is not polled.
    pub async fn polling_status_async(&self) -> Result<Vec<String>> {
        let name = self.name_async().await?;
        self.admin_dev_async().await?.command_async("DevPollStatus", &*name).await
    }

    /// The polling status of this device, see [`Self::polling_status_async`].
    pub fn polling_status(&self) -> Result<Vec<String>> {
        self.block_on(self.polling_status_async())?
    }

    /// Poll that object of this device, of that type as the admin device names it, at that
    /// period, whether it is polled already or not, see [`Self::poll_attr_async`].
    async fn poll_async(&self, kind: &str, name: &str, period: Duration) -> Result<()> {

        let dev_name = self.name_async().await?;
        let admin = self.admin_dev_async().await?;
        let period = i32::try_from(period.as_millis()).map_err(|_| Error::new(ErrorKind::InvalidData)
            .with_description(format!("Polling period {period:?} is too long")))?;

        let input = Value::LongString(
            Spectrum::from_slice(&[period], [1]).expect("one period"), 
            Spectrum::from(&[&*dev_name, kind, name][..]));

        // Only an object that is not polled can be added, whose period is otherwise
        // updated, as cppTango does.
        match admin.command_async::<_, ()>("AddObjPolling", input.clone()).await {
            Err(err) if err.reason() == reason::API_AlreadyPolled => {
                admin.command_async("UpdObjPollingPeriod", input).await
            }
            res => res,
        }

    }

    /// Stop polling that object of this device, see [`Self::stop_poll_attr_async`].
    async fn stop_poll_async(&self, kind: &str, name: &str) -> Result<()> {
        let dev_name = self.name_async().await?;
        let admin = self.admin_dev_async().await?;
        admin.command_async("RemObjPolling", Spectrum::<str>::from(&[&*dev_name, kind, name][..])).await
    }

    // The properties of a device are the database's and not the device's, so everything
    // below asks the device for its name and then the database for what it holds of it,
    // see the very same methods on [`DatabaseProxy`].

    /// The names of the properties of this device that match the given filter.
    pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
        let dev_name = self.name_async().await?;
        self.database()?.list_dev_prop_async(&dev_name, filter).await
    }

    /// The names of the properties of this device, see [`Self::list_prop_async`].
    pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
        self.block_on(self.list_prop_async(filter))?
    }

    /// The properties of this device, parsed into the shape the given names ask for. Like
    /// for [`Self::read_attr`], this specification is really flexible and allows various
    /// signatures for the function:
    /// 1. `get_prop<T>(&str) -> Result<T>`
    /// 2. `get_prop<T>(&[&str]) -> Result<Vec<Result<T>>>`
    /// 3. `get_prop<T>([&str; N]) -> Result<[Result<T>; N]>`
    /// 4. `get_prop<(A, B, ...)>((&str, &str, ...)) -> Result<(Result<A>, Result<B>, ...)>`
    /// 5. `get_prop<Spectrum<str>>(All) -> Result<Vec<Named<Spectrum<str>>>>`
    ///
    /// See [`DatabaseProxy::get_dev_prop_async`] for what the values may be parsed into
    /// and what a property the database holds nothing for gives.
    pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
    where
        Spec: GetPropSpec<Names>,
    {
        // The name of the device, which goes first, is only known once it is asked for, so
        // the names are taken out as a buffer of their own first.
        let names = names.to_names();
        async move {
            let dev_name = self.name_async().await?;
            let params = names.map(|names| prop_params(&dev_name, names.iter()));
            let mut reply = self.database()?.get_raw_prop_async(PropCommands::DEV, &dev_name, params).await?;
            Spec::collect(&mut reply)
        }
    }

    /// The properties of this device, see [`Self::get_prop_async`].
    pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
    where
        Spec: GetPropSpec<Names>,
    {
        self.block_on(self.get_prop_async::<Spec, Names>(names))?
    }

    /// Write the given properties of this device, given as pairs of a name and a value,
    /// which allows various signatures for the function:
    /// 1. `set_prop((&str, T)) -> Result<()>`
    /// 2. `set_prop([(&str, T); N]) -> Result<()>`
    /// 3. `set_prop(Vec<(&str, T)>) -> Result<()>`
    /// 4. `set_prop(((&str, A), (&str, B), ...)) -> Result<()>`
    ///
    /// See [`DatabaseProxy::set_dev_prop_async`] for what the values may be.
    pub fn set_prop_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<()>> + Send
    where
        Spec: SetPropSpec<'a>,
    {
        let params = set_prop_params(spec);
        async move {
            let dev_name = self.name_async().await?;
            self.database()?.prop_command_async("DbPutDeviceProperty", prop_params(&dev_name, params.iter().map(|param| &**param))).await
        }
    }

    /// Write the given properties of this device, see [`Self::set_prop_async`].
    pub fn set_prop<'a, Spec>(&self, spec: Spec) -> Result<()>
    where
        Spec: SetPropSpec<'a>,
    {
        self.block_on(self.set_prop_async(spec))?
    }

    /// Delete the given properties of this device, which allows various signatures for the
    /// function:
    /// 1. `delete_prop(&str) -> Result<()>`
    /// 2. `delete_prop(&[&str]) -> Result<()>`
    /// 3. `delete_prop([&str; N]) -> Result<()>`
    /// 4. `delete_prop((&str, &str, ...)) -> Result<()>`
    ///
    /// See [`DatabaseProxy::delete_dev_prop_async`].
    pub fn delete_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
        let names = names.to_names_not_all();
        async move {
            let dev_name = self.name_async().await?;
            self.database()?.prop_command_async("DbDeleteDeviceProperty", prop_params(&dev_name, names.iter())).await
        }
    }

    /// Delete the given properties of this device, see [`Self::delete_prop_async`].
    pub fn delete_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
        self.block_on(self.delete_prop_async(names))?
    }

    /// Subscribe to the readings of an attribute, of one of the kinds a device pushes them
    /// under, the channel giving the latest of them, see [`DevChannel`].
    ///
    /// Nothing is subscribed until the channel is polled, or its subscription ensured, see
    /// [`DevChannel::ensure_subscribed_async`]. There is no initial value: the first event
    /// is the next one the device pushes, so read the attribute once subscribed to know
    /// where it stands.
    pub fn subscribe_attr_value<T: FromValue>(&self, name: &str, event: AttrValueEvent) -> DevChannel<AttrRead<T>> {
        DevChannel::new(self.clone(), name, ChannelKind::Value(event), ChannelEvents::poll_value)
    }

    /// Subscribe to an attribute being announced as ready to be read, which carries nothing
    /// of the reading: making it is the subscriber's to do, see [`AttrReady`]. See
    /// [`Self::subscribe_attr_value`].
    pub fn subscribe_attr_ready(&self, name: &str) -> DevChannel<AttrReady> {
        DevChannel::new(self.clone(), name, ChannelKind::Ready, ChannelEvents::poll_ready)
    }

    /// Subscribe to the configuration of an attribute changing, every event carrying the
    /// configuration as it stands after the change. See [`Self::subscribe_attr_value`].
    pub fn subscribe_attr_config(&self, name: &str) -> DevChannel<AttrInfo> {
        DevChannel::new(self.clone(), name, ChannelKind::Config, ChannelEvents::poll_config)
    }

}

/// The clone shares the connection of this proxy if it can read it right away, and connects
/// on its own otherwise.
/// The period an object is polled at, out of the polling status of its device, as the
/// `DevPollStatus` command of its admin device gives it: the text of each object polled
/// starts with `Polled <kind> name = <name>` and tells its period in a later line, which is
/// what cppTango gives and its clients parse.
pub(crate) fn status_period(status: &[impl AsRef<str>], kind: &str, name: &str) -> Option<Duration> {
    let prefix = format!("Polled {kind} name = ");
    status.iter().find_map(|entry| {
        let mut lines = entry.as_ref().lines();
        let polled = lines.next()?.strip_prefix(&prefix)?;
        if !polled.trim().eq_ignore_ascii_case(name) {
            return None;
        }
        lines.find_map(|line| line.strip_prefix("Polling period (mS) = "))
            .and_then(|ms| ms.trim().parse().ok())
            .map(Duration::from_millis)
    })
}

impl Clone for DevProxy {
    fn clone(&self) -> Self {
        Self {
            inner: Box::new(ProxyInner {
                backend: self.inner.backend.clone(),
                target: self.inner.target.clone(),
                source: self.inner.source,
                timeout: self.inner.timeout,
            }),
            state: self.state.try_read().ok()
                .map(|guard| RwLock::new((*guard).clone()))
                .unwrap_or_else(|| RwLock::new(ProxyState {
                    backend: None,
                    database_host_and_port: None,
                })),
        }
    }
}

impl PartialEq for DevProxy {
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner.backend, &other.inner.backend) && self.inner.target == other.inner.target
    }
}

impl Eq for DevProxy { }

impl Hash for DevProxy {
    #[inline]
    fn hash<H: Hasher>(&self, state: &mut H) {
        Arc::as_ptr(&self.inner.backend).hash(state);
        self.inner.target.hash(state);
    }
}

/// The enum is internal so we just impl Display on it, in order to properly display it
/// in the tracing spans!
impl fmt::Display for ProxyTarget {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ProxyTarget::DatabaseFromEnv => 
                write!(f, "db"),
            ProxyTarget::DatabaseAt { host, port } => 
                write!(f, "db@{host}:{port}"),
            ProxyTarget::RegularFromDatabase { name, .. } => 
                write!(f, "{name}"),
            ProxyTarget::RegularAt { host, port, name } => 
                write!(f, "{name}@{host}:{port}"),
        }
    }
}

// =========================== //
//          To names           //
// =========================== //

/// Supporting trait for being generic over the different types that can be interpreted 
/// as names, either single or many names.
///
/// The names are given to a backend as one buffer of strings, which a served device shares
/// in what it answers with, and which a buffer of names already given passes through
/// untouched: names read back from a device, or listed by the database, are given again
/// without being copied.
///
/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
/// implemented for everything that API accepts and is not one to implement anywhere else.
pub trait ToNames: Sized {

    /// The names, in the order they were given, absent to target **all** elements, borrowed
    /// where they are: what a request copies into its own parameters, once, rather than
    /// first into a buffer of their own, see [`Spectrum::from_strs`].
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>>;

    /// The names as one buffer, see [`Self::iter_names`].
    #[inline]
    fn to_names(self) -> Option<Spectrum<str>> {
        self.iter_names().map(names_buf)
    }

}

/// The names as one buffer, copied out of where they are.
fn names_buf<'a>(names: impl Iterator<Item = &'a str>) -> Spectrum<str> {
    Spectrum::from_strs(names).expect("names of more than 4 GiB")
}

/// Same as [`ToNames`] but this marker ensures that no None will be returned by to_names.
///
/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
/// implemented for everything that API accepts and is not one to implement anywhere else.
pub trait ToNamesNotAll: ToNames {

    /// Utility function to ensure no `None` could be returned for the names.
    #[inline]
    fn iter_names_not_all(&self) -> impl Iterator<Item = &str> {
        self.iter_names().expect("should not return the 'all' marker")
    }

    /// Utility function to ensure no `None` could be returned for the names.
    #[inline]
    fn to_names_not_all(self) -> Spectrum<str> {
        self.to_names().expect("should not return the 'all' marker")
    }

}

/// Something along with the name it is of, the name sharing the octets it was decoded out
/// of like any other string.
///
/// This is what a request answering for several names gives one of per name: the result
/// of that one name, which is not the one of the request. The request failing fails it as
/// a whole, while an entry failing is the device failing for that one name, which the other
/// entries are unaffected by.
#[derive(Debug, Clone, PartialEq)]
pub struct Named<T> {
    /// The name.
    pub name: Scalar<str>,
    /// What is of that name.
    pub value: T,
}

impl<T> Named<T> {

    #[inline]
    pub fn new(name: Scalar<str>, value: T) -> Self {
        Self { name, value }
    }

    /// The same name along with what that function makes of its value.
    #[inline]
    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Named<U> {
        Named { name: self.name, value: f(self.value) }
    }

}

/// What a request answering for several names gives one of per name, see [`Named`].
pub type NamedResult<T> = Named<Result<T>>;

/// Something along with the time it was recorded at, which is what a history gives one of
/// per entry, see [`DevProxy::command_history`]: the result of what was recorded then,
/// which is not the one of the request asking for the history.
#[derive(Debug, Clone, PartialEq)]
pub struct Recorded<T> {
    /// When it was recorded.
    pub time: SystemTime,
    /// What was recorded.
    pub value: T,
}

impl<T> Recorded<T> {

    #[inline]
    pub fn new(time: SystemTime, value: T) -> Self {
        Self { time, value }
    }

    /// The same time along with what that function makes of its value.
    #[inline]
    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Recorded<U> {
        Recorded { time: self.time, value: f(self.value) }
    }

}

/// What a history gives one of per entry, see [`Recorded`].
pub type RecordedResult<T> = Recorded<Result<T>>;

/// A marker struct for specifying names for various functions that are using [`ToNames`]
/// to specify to select **all** names available on the device.
pub struct All;
impl ToNames for All {
    #[inline]
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
        None::<std::iter::Empty<&str>>
    }
}

impl ToNames for &str {
    #[inline]
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
        Some(std::iter::once(*self))
    }
}
impl ToNamesNotAll for &str { }

impl ToNames for &[&str] {
    #[inline]
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
        Some(self.iter().copied())
    }
}
impl ToNamesNotAll for &[&str] { }

impl<const N: usize> ToNames for [&str; N] {
    #[inline]
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
        Some(self.iter().copied())
    }
}
impl<const N: usize> ToNamesNotAll for [&str; N] { }

impl ToNames for Spectrum<str> {
    #[inline]
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
        Some(self.iter())
    }
    #[inline]
    fn to_names(self) -> Option<Spectrum<str>> {
        Some(self)
    }
}
impl ToNamesNotAll for Spectrum<str> { }

impl ToNames for &Spectrum<str> {
    #[inline]
    fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
        Some(self.iter())
    }
    #[inline]
    fn to_names(self) -> Option<Spectrum<str>> {
        Some(self.clone())
    }
}
impl ToNamesNotAll for &Spectrum<str> { }

macro_rules! impl_to_names_tuple {
    ( @name: $comp:ident ) => { &str };
    ( @one: $comp:ident ) => { 1 };
    ( $( $comp:ident )* ) => {
        #[allow(unused_variables)]
        impl ToNames for ( $(impl_to_names_tuple!(@name: $comp),)* ) {
            #[inline]
            fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
                let ( $($comp,)* ) = *self;
                Some(<[&str; 0 $( + impl_to_names_tuple!(@one: $comp) )*]>::into_iter([ $($comp,)* ]))
            }
        }
        impl ToNamesNotAll for ( $(impl_to_names_tuple!(@name: $comp),)* ) { }
    };
}

impl_to_names_tuple!();
impl_to_names_tuple!(a);
impl_to_names_tuple!(a b);
impl_to_names_tuple!(a b c);
impl_to_names_tuple!(a b c d);
impl_to_names_tuple!(a b c d e);
impl_to_names_tuple!(a b c d e f);

/// Generic specification type used for command and attribute info, as there is no 
/// special processing to do on the resulted value! This is an extension to the ToNames,
/// because this spec's output only depends on the actual type of the names we give.
///
/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
/// implemented for everything that API accepts and is not one to implement anywhere else.
pub trait ToNamesSpec<T>: ToNames {

    /// The final type of the output.
    type Output;

    /// Build the output out of the entries the device answered with, one per name.
    ///
    /// The shape of the output is what the names that were given say it is, which is what
    /// this is implemented on.
    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output>;

}

/// One name, one entry: that entry *is* the output, and its failure is the output's own
/// since there is no slot of its own to carry it in.
impl<T> ToNamesSpec<T> for &str {

    type Output = T;

    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
        match entries.into_iter().next() {
            Some(entry) => entry.value,
            None => Err(unexpected_len_err(1)),
        }
    }

}

impl<T> ToNamesSpec<T> for &[&str] {

    type Output = Vec<Result<T>>;

    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
        Ok(entries.into_iter().map(|entry| entry.value).collect())
    }

}

impl<T> ToNamesSpec<T> for Spectrum<str> {

    type Output = Vec<Result<T>>;

    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
        <&[&str] as ToNamesSpec<T>>::collect(entries)
    }

}

impl<T> ToNamesSpec<T> for &Spectrum<str> {

    type Output = Vec<Result<T>>;

    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
        <&[&str] as ToNamesSpec<T>>::collect(entries)
    }

}

impl<T, const N: usize> ToNamesSpec<T> for [&str; N] {

    type Output = [Result<T>; N];

    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
        take_array(entries.into_iter().map(|entry| entry.value))
    }

}

impl<T> ToNamesSpec<T> for All {

    // Nothing was named here, so every entry carries its own name and this is the one
    // form whose output has room for it.
    type Output = Vec<NamedResult<T>>;

    fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
        Ok(entries)
    }

}

/// The entries of an output of a fixed shape, one per slot and in order: a device that
/// answers with another number of them broke what the protocol says of it, which a last
/// entry being there behind the slots tells as well as one of them missing.
pub(crate) fn take_array<T, const N: usize>(mut entries: impl Iterator<Item = T>) -> Result<[T; N]> {
    let mut slots = [const { None }; N];
    for slot in &mut slots {
        *slot = Some(entries.next().ok_or_else(|| unexpected_len_err(N))?);
    }
    if entries.next().is_some() {
        return Err(unexpected_len_err(N));
    }
    Ok(slots.map(|slot| slot.expect("every slot is filled")))
}

macro_rules! impl_to_names_spec_tuple {
    ( @one: $index:tt ) => { 1 };
    ( @name: $index:tt ) => { &str };
    ( @t: $index:tt ) => { T };
    ( @next: $index:tt, $entries:ident ) => { $entries.next().ok_or_else(|| unexpected_len_err(LEN))?.value };
    ( $( $index:tt )* ) => {

        impl<T> ToNamesSpec<T> for ( $( impl_to_names_spec_tuple!(@name: $index), )* ) {

            type Output = ( $( Result< impl_to_names_spec_tuple!(@t: $index) >,)* );

            fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
                const LEN: usize = 0 $( + impl_to_names_spec_tuple!(@one: $index) )*;
                // The entries are in the order the names were given, which is the order the
                // components of a tuple are built in.
                let mut entries = entries.into_iter();
                let output = ( $( impl_to_names_spec_tuple!(@next: $index, entries), )* );
                if entries.next().is_some() {
                    return Err(unexpected_len_err(LEN));
                }
                Ok(output)
            }

        }

    };
}

impl_to_names_spec_tuple!(0);
impl_to_names_spec_tuple!(0 1);
impl_to_names_spec_tuple!(0 1 2);
impl_to_names_spec_tuple!(0 1 2 3);
impl_to_names_spec_tuple!(0 1 2 3 4);
impl_to_names_spec_tuple!(0 1 2 3 4 5);

// =========================== //
//       Read attributes       //
// =========================== //

/// A trait for attribute specifiers, which allows a great flexibility in how 
/// [`DevProxy::read_attr`] works.
///
/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
/// implemented for everything that API accepts and is not one to implement anywhere else.
pub trait ReadAttrSpec<Names> {

    /// The final type of the output.
    type Output;

    /// Build the output out of the readings, one per attribute.
    ///
    /// The shape of the output is what the names that were given say it is, and what every
    /// reading is converted into is what this is implemented on.
    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output>;

}

impl<T: FromValue> ReadAttrSpec<&str> for T {

    type Output = AttrRead<T>;

    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
        match reads.into_iter().next() {
            Some(entry) => entry.value?.parse(),
            None => Err(unexpected_len_err(1)),
        }
    }

}

impl<T: FromValue> ReadAttrSpec<&'_ [&str]> for T {

    type Output = Vec<Result<AttrRead<T>>>;

    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
        Ok(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)).collect())
    }

}

impl<T: FromValue> ReadAttrSpec<Spectrum<str>> for T {

    type Output = Vec<Result<AttrRead<T>>>;

    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
        <T as ReadAttrSpec<&[&str]>>::collect(reads)
    }

}

impl<T: FromValue> ReadAttrSpec<&Spectrum<str>> for T {

    type Output = Vec<Result<AttrRead<T>>>;

    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
        <T as ReadAttrSpec<&[&str]>>::collect(reads)
    }

}

impl<T: FromValue, const N: usize> ReadAttrSpec<[&str; N]> for T {

    type Output = [Result<AttrRead<T>>; N];

    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
        take_array(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)))
    }

}

impl<T: FromValue> ReadAttrSpec<All> for T {

    // Nothing was named here, so every reading carries the name of its attribute.
    type Output = Vec<NamedResult<AttrRead<T>>>;

    fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
        Ok(reads.into_iter().map(|entry| entry.map(|read| read.and_then(AttrRead::parse))).collect())
    }

}

macro_rules! impl_read_attr_spec_tuple {
    ( @one: $comp:ident ) => { 1 };
    ( @name: $comp:ident ) => { &'name str };
    ( $( $comp:ident )* ) => {

        impl<'name, $($comp,)* > ReadAttrSpec<( $( impl_read_attr_spec_tuple!(@name: $comp), )* )> for ( $($comp,)* )
        where
            $( $comp: FromValue, )*
        {

            type Output = ( $( Result<AttrRead<$comp>>,)* );

            fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
                const LEN: usize = 0 $( + impl_read_attr_spec_tuple!(@one: $comp) )*;
                // The readings are in the order the names were given, which is the order
                // the components of a tuple are built in, and every one of them is
                // converted into the type of its own component.
                let mut reads = reads.into_iter();
                let output = ( $( reads.next().ok_or_else(|| unexpected_len_err(LEN))?.value.and_then(AttrRead::parse::<$comp>), )* );
                if reads.next().is_some() {
                    return Err(unexpected_len_err(LEN));
                }
                Ok(output)
            }

        }

    };
}

impl_read_attr_spec_tuple!(A);
impl_read_attr_spec_tuple!(A B);
impl_read_attr_spec_tuple!(A B C);
impl_read_attr_spec_tuple!(A B C D);
impl_read_attr_spec_tuple!(A B C D E);
impl_read_attr_spec_tuple!(A B C D E F);

// =========================== //
//      Write attributes       //
// =========================== //

/// The values to write to one or many attributes, see [`DevProxy::write_attr`].
///
/// **Supporting trait**: it is only there to make this crate's API easier to use, it is
/// implemented for everything that API accepts and is not one to implement anywhere else.
pub trait WriteAttrSpec<'a>: Sized {

    /// Make this write specification into the values it represents, each with the name of
    /// its attribute.
    fn into_values(self) -> Vec<Named<Value>>;

    /// The final type of the output.
    type Output;

    /// Build the output out of the outcomes of the write, which come in the order the
    /// values were given: a write answers one outcome per value and names nothing back,
    /// see [`DevProxy::write_attr_async`].
    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output>;

    /// Names for the read specification type.
    type ReadNames;
    /// The read specification type, for when using write_read_attr.
    type ReadSpec;

}

/// For a single attribute.
impl<'a, T: IntoValue> WriteAttrSpec<'a> for (&'a str, T) {

    #[inline]
    fn into_values(self) -> Vec<Named<Value>> {
        vec![Named::new(Scalar::from(self.0), self.1.into_value())]
    }

    type Output = ();

    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
        match writes.into_iter().next() {
            Some(res) => res,
            None => Err(unexpected_len_err(1)),
        }
    }

    type ReadNames = &'a str;
    type ReadSpec = T;

}

/// For an array of attributes.
impl<'a, T: IntoValue, const N: usize> WriteAttrSpec<'a> for [(&'a str, T); N] {

    #[inline]
    fn into_values(self) -> Vec<Named<Value>> {
        named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
    }

    type Output = [Result<()>; N];

    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
        take_array(writes.into_iter())
    }

    type ReadNames = [&'a str; N];
    type ReadSpec = T;

}

/// For a vector of attributes.
impl<'a, T: IntoValue> WriteAttrSpec<'a> for Vec<(&'a str, T)> {

    #[inline]
    fn into_values(self) -> Vec<Named<Value>> {
        named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
    }

    type Output = Vec<Result<()>>;

    fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
        Ok(writes)
    }

    type ReadNames = &'a [&'a str];
    type ReadSpec = T;

}

/// The values of a write, every name a scalar of its own sharing a single copy of all the
/// names, see [`Spectrum::get_scalar`].
fn named_values(values: Vec<(&str, Value)>) -> Vec<Named<Value>> {
    let names = names_buf(values.iter().map(|&(name, _)| name));
    values.into_iter()
        .enumerate()
        .map(|(index, (_, value))| Named::new(names.get_scalar([index]).expect("as many names as values"), value))
        .collect()
}

macro_rules! impl_write_attr_spec_tuple {
    ( @name: $comp:ident ) => { &'a str };
    ( @one: $comp:ident ) => { 1 };
    ( @unit: $comp:ident ) => { Result<()> };
    ( @next: $comp:ident, $writes:ident ) => { $writes.next().ok_or_else(|| unexpected_len_err(LEN))? };
    ( $( $comp:ident )* ) => {

        #[allow(non_snake_case)]
        impl<'a, $($comp: IntoValue, )*> WriteAttrSpec<'a> for ( $( (&'a str, $comp), )* ) {

            #[inline]
            fn into_values(self) -> Vec<Named<Value>> {
                let ( $($comp,)* ) = self;
                named_values(vec![$(($comp.0, $comp.1.into_value()), )* ])
            }

            type Output = ( $( impl_write_attr_spec_tuple!(@unit: $comp), )* );

            fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
                const LEN: usize = 0 $( + impl_write_attr_spec_tuple!(@one: $comp) )*;
                // The outcomes come in the order the values were given, which is the order
                // the components of a tuple are built in.
                let mut writes = writes.into_iter();
                let output = ( $( impl_write_attr_spec_tuple!(@next: $comp, writes), )* );
                if writes.next().is_some() {
                    return Err(unexpected_len_err(LEN));
                }
                Ok(output)
            }

            type ReadNames = ( $(impl_write_attr_spec_tuple!(@name: $comp), )* );
            type ReadSpec = ( $($comp, )* );

        }

    };
}

impl_write_attr_spec_tuple!(A);
impl_write_attr_spec_tuple!(A B);
impl_write_attr_spec_tuple!(A B C);
impl_write_attr_spec_tuple!(A B C D);
impl_write_attr_spec_tuple!(A B C D E);
impl_write_attr_spec_tuple!(A B C D E F);

// =========================== //
//          Commands           //
// =========================== //

/// Static information about a command.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommandInfo {
    /// The display level of the command.
    pub display_level: DisplayLevel,
    /// Input data kind.
    pub in_kind: CommandDataKind,
    /// Output data kind.
    pub out_kind: CommandDataKind,
    /// Description for the input data.
    pub in_doc: Option<Cow<'static, str>>,
    /// Description for the output data.
    pub out_doc: Option<Cow<'static, str>>,
}

// =========================== //
//         Attributes          //
// =========================== //

/// Represent the reading for a single attribute.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttrRead<T> {
    /// The read value of the attribute, if any. Absent if quality is Invalid or if the
    /// attribute is not readable.
    pub value: Option<T>,
    /// The write value of the attribute, if any. Absent if quality is Invalid or if the
    /// attribute is not writable.
    pub w_value: Option<T>,
    /// The quality of the value.
    pub quality: AttrQuality,
    /// The timestamp of the value.
    pub time: SystemTime,
}

impl<T> AttrRead<T> {

    pub fn new(value: T) -> Self {
        Self {
            value: Some(value),
            w_value: None,
            quality: AttrQuality::Valid,
            time: SystemTime::now(),
        }
    }

    pub fn with_quality(mut self, quality: AttrQuality) -> Self {
        self.quality = quality;
        self
    }

    pub fn with_time(mut self, time: SystemTime) -> Self {
        self.time = time;
        self
    }

}

impl AttrRead<Value> {

    /// This reading, its values converted into the type they are expected to be.
    fn parse<T: FromValue>(self) -> Result<AttrRead<T>> {
        Ok(AttrRead {
            value: match self.value {
                Some(value) => Some(T::from_value(value)
                    .map_err(|err| context_err(err, "Failed to parse read value"))?),
                None => None,
            },
            w_value: match self.w_value {
                Some(value) => Some(T::from_value(value)
                    .map_err(|err| context_err(err, "Failed to parse write value"))?),
                None => None,
            },
            quality: self.quality,
            time: self.time,
        })
    }

}

/// An attribute announced as ready to be read, which is what a data ready event carries:
/// nothing of the reading itself, which is the subscriber's to make, but the data kind of
/// the attribute and a counter the device gives with every announcement, telling which of
/// its readings is the one made ready.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AttrReady {
    /// The data kind of the attribute.
    pub data_kind: AttrDataKind,
    /// The counter the device gave with this announcement, see
    /// [`crate::server::Dev::push_data_ready_event`].
    pub counter: i32,
}

/// Quality factor of the attribute.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AttrQuality {
    Valid,
    Invalid,
    Alarm,
    Changing,
    Warning,
}

/// An extension to AttrConfig with static information about the attribute.
#[derive(Debug, Clone, PartialEq)]
pub struct AttrInfo {
    /// The configuration of the attribute, which is the whole of what a client is allowed
    /// to change of it, see [`AttrConfig`]. Everything else here is what the attribute is,
    /// which the device declares and nothing outside of it ever changes.
    pub config: AttrConfig,
    /// The attribute mode.
    pub mode: AttrMode,
    /// The data kind to be expected.
    pub data_kind: AttrDataKind,
    /// The data format to be expected, along with the maximum dimensions the attribute is
    /// declared with for the formats that have any.
    pub data_format: AttrDataFormat,
    /// The display level of the attribute.
    pub display_level: DisplayLevel,
    /// The memorized mode, absent for an attribute that remembers nothing.
    pub memorized: Option<AttrMemorized>,
}

/// Attribute read/write mode.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttrMode {
    /// Read-only attribute
    Read,
    /// The attribute is technically read-only on the server side, but is associated to
    /// another write-only attribute that acts as the set point for that read-only value.
    ReadWithWrite {
        /// The name of the associated write attribute.
        write_attr_name: Cow<'static, str>,
    },
    /// Write-only attribute.
    Write,
    /// Read/write attribute.
    ReadWrite,
}

/// Represent the current configuration of an attribute.
///
/// The default configuration is the one of an attribute that declares nothing: no
/// description, no unit, no bound of any kind, no event configured, which is what a
/// definition starts from and then fills in.
///
/// This is exactly what Tango keeps as the *attribute properties* of a device in the
/// database, and therefore exactly what a client is allowed to change, which is what
/// [`crate::backend::BackendDev::set_attr_config`] carries. Everything else a client reads
/// about an attribute is what the attribute *is* rather than how it is configured, and is
/// declared by the device, see [`AttrInfo`].
#[derive(Debug, Clone, PartialEq, Default)]
pub struct AttrConfig {
    /// The optional description of the attribute, which a client may change like any other
    /// property, `description` being one of the attribute properties Tango keeps.
    pub doc: Option<Cow<'static, str>>,
    /// The optional label of the attribute.
    pub label: Option<Cow<'static, str>>,
    /// The optional real unit of the attribute: how the raw attribute value is
    /// represented.
    pub unit: Option<Cow<'static, str>>,
    /// The conversion factor for going from the raw attribute value to the international
    /// unit system.
    pub standard_unit: Option<f64>,
    /// The optional unit for displaying to the user.
    pub display_unit: Option<Cow<'static, str>>,
    /// The display format of this value. That format is specified using printf formatting
    /// symbols using `%`.
    pub display_format: Option<Cow<'static, str>>,
    /// The allowed range for this attribute's value.
    pub range: AttrRangeConfig,
    /// The alarm range for this attribute's value.
    pub alarm: AttrRangeConfig,
    /// The warning range for this attribute's value.
    pub warning: AttrRangeConfig,
    /// The optional delta alarm.
    pub delta: Option<AttrDeltaConfig>,
    /// Configuration of the change event.
    pub change_event: AttrChangeEventConfig,
    /// Configuration of the periodic event.
    pub periodic_event: AttrPeriodicEventConfig,
    /// Configuration of the archive event.
    pub archive_event: AttrArchiveEventConfig,
    /// What the values of the attribute mean, which only a `DevEnum` has any of: they name
    /// its values, an enumerated value being the index of its own label in this list.
    ///
    /// This is a property like any other and a client may change it, which is why it is
    /// here and not beside [`AttrInfo::data_kind`]: what a value of the attribute *is*
    /// stays a `DevEnum` whatever the labels are renamed to.
    pub enum_labels: Vec<Cow<'static, str>>,
    /// Possible extensions.
    pub extensions: Vec<Cow<'static, str>>,
}

/// A limit of an attribute: one of the bounds of one of its ranges, or the delta of its
/// read-different-from-set alarm.
///
/// Tango carries these as strings, the type to read them as being the one of the attribute
/// they belong to, and they are numbers by the time they get here: a limit is something to
/// compare a value against, which is not something a string does.
///
/// They are not all the same number, though. A limit is a value of the attribute itself,
/// so an attribute of the widest integer kinds has limits a double cannot hold without
/// rounding them, and the three ways a Tango value is a number are kept apart rather than
/// all being flattened into one:
///
/// - a floating attribute, `DevFloat` or `DevDouble`, has [`Self::Double`] limits;
/// - a signed one, up to `DevLong64`, has [`Self::Long`] limits;
/// - an unsigned one, up to `DevULong64`, has [`Self::ULong`] limits.
///
/// A device that writes a limit its own type cannot hold — half a unit on an integer
/// attribute — still has a limit, and it is read as a [`Self::Double`] rather than lost.
///
/// Two limits compare as the quantities they stand for, whatever variant each is, so that
/// `Long(2)` and `Double(2.0)` are equal: two integers are compared exactly, the widest
/// ones included, and anything else as doubles, a not-a-number being equal to nothing and
/// ordered with nothing.
#[derive(Debug, Clone, Copy)]
pub enum AttrNumber {
    Double(f64),
    Long(i64),
    ULong(u64),
}

impl AttrNumber {

    /// This limit as a double, which is what comparing it against a value of any other
    /// kind goes through, the widest integers losing their last digits the way they do in
    /// any other conversion to a double.
    pub fn as_f64(self) -> f64 {
        match self {
            Self::Double(value) => value,
            Self::Long(value) => value as f64,
            Self::ULong(value) => value as f64,
        }
    }

}

impl PartialEq for AttrNumber {
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        self.partial_cmp(other) == Some(Ordering::Equal)
    }
}

impl PartialOrd for AttrNumber {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        match (*self, *other) {
            (Self::Long(a), Self::Long(b)) => Some(a.cmp(&b)),
            (Self::ULong(a), Self::ULong(b)) => Some(a.cmp(&b)),
            (Self::Long(a), Self::ULong(b)) => Some(i128::from(a).cmp(&i128::from(b))),
            (Self::ULong(a), Self::Long(b)) => Some(i128::from(a).cmp(&i128::from(b))),
            (a, b) => a.as_f64().partial_cmp(&b.as_f64()),
        }
    }
}

impl fmt::Display for AttrNumber {
    /// Written the way a device writes it, which is what it goes back on the wire as.
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Double(value) => Display::fmt(value, f),
            Self::Long(value) => Display::fmt(value, f),
            Self::ULong(value) => Display::fmt(value, f),
        }
    }
}

/// Every number that fits in one of the three, so that a limit can be given as the value
/// it is rather than as the variant holding it.
macro_rules! impl_attr_number_from {
    ($variant:ident: $($ty:ty),* $(,)?) => { $(
        impl From<$ty> for AttrNumber {
            #[inline]
            fn from(value: $ty) -> Self {
                Self::$variant(value.into())
            }
        }
    )* };
}

impl_attr_number_from!(Double: f32, f64);
impl_attr_number_from!(Long: i8, i16, i32, i64);
impl_attr_number_from!(ULong: u8, u16, u32, u64);

/// The two bounds of one of the ranges of an attribute, either of which is absent for a
/// range that is open on that side, which is also how an attribute says it has no such
/// range at all.
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct AttrRangeConfig {
    pub min: Option<AttrNumber>,
    pub max: Option<AttrNumber>,
}

impl AttrRangeConfig {
    /// Whether that range has any bound, a range with none being no range at all.
    #[inline]
    pub fn is_set(&self) -> bool {
        self.min.is_some() || self.max.is_some()
    }
}

/// The alarm on the read value drifting away from the written one.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct AttrDeltaConfig {
    /// How far the two may be apart before the attribute is in alarm, in the type of the
    /// attribute, being a value of it like any other limit.
    pub value: AttrNumber,
    /// How long they may stay that far apart before it counts, absent for immediately.
    pub time: Option<Duration>,
}

/// The criteria of the change event: how much the value has to move for the server to send
/// one, absolutely or relatively, either of which may be left out.
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct AttrChangeEventConfig {
    /// The relative change, as a percentage of the previous value.
    pub rel_change: Option<f64>,
    /// The absolute change, in the unit of the attribute.
    pub abs_change: Option<f64>,
}

impl AttrChangeEventConfig {
    /// Whether the change event has any criterion, an event with none being configured.
    #[inline]
    pub fn is_set(&self) -> bool {
        self.rel_change.is_some() || self.abs_change.is_some()
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct AttrPeriodicEventConfig {
    pub period: Option<Duration>,
}

impl AttrPeriodicEventConfig {
    /// Whether the periodic event has a period of its own.
    #[inline]
    pub fn is_set(&self) -> bool {
        self.period.is_some()
    }
}

/// The criteria of the archive event: the same two changes as the change event, and the
/// longest an archiver may go without an event when the value does not move.
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct AttrArchiveEventConfig {
    pub rel_change: Option<f64>,
    pub abs_change: Option<f64>,
    pub period: Option<Duration>,
}

impl AttrArchiveEventConfig {
    /// Whether the archive event has any criterion, a change or a period.
    #[inline]
    pub fn is_set(&self) -> bool {
        self.rel_change.is_some() || self.abs_change.is_some() || self.period.is_some()
    }
}

/// What a memorized attribute does, when its device is initialized, with the value it
/// remembers.
///
/// Both modes store the value of every write in the database, they only differ in what the
/// next initialization of the device does with the value it finds there.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AttrMemorized {
    /// The remembered value is restored as the write value of the attribute and nothing
    /// else: a client reading the attribute back finds the set point it left, but the write
    /// function is never called, so the device does not act on its hardware only because
    /// it was restarted.
    RestoreSetPoint,
    /// The remembered value is written again, through the write function of the attribute,
    /// which puts the device back in the state it was left in.
    RestoreAndWrite,
}

// =========================== //
//           Device            //
// =========================== //

/// Static information about a device: what it is and where it runs.
///
/// None of this is something a client computes anything from, it is what it shows about a
/// device, and all of it is decided by the server rather than by the device itself.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct DevInfo {
    /// The name of the class the device is an instance of.
    pub class: String,
    /// The `exe/inst` name of the server running the device, which is what it is
    /// registered in the database under.
    pub server_id: String,
    /// The name of the host the server runs on.
    pub server_host: String,
    /// The IDL version the server says it implements, which is a statement it makes about
    /// itself and decides nothing.
    pub server_version: i32,
    /// Where the documentation of the class is, when it has any.
    pub doc_url: Option<Cow<'static, str>>,
    /// What the device is, which is free text a class gives itself, absent for a device
    /// whose server is too old to have any.
    pub dev_type: Option<Cow<'static, str>>,
    /// The versions of whatever the server is built out of, as the pairs of a name and a
    /// version that the server chooses to publish. Empty for a server too old to have any,
    /// this being the one thing IDL v6 added.
    pub version_info: Vec<(Cow<'static, str>, Cow<'static, str>)>,
}

// =========================== //
//        Various enums        //
// =========================== //

/// All possible device states.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[repr(u8)]
pub enum DevState {
    On,
    Off,
    Close,
    Open,
    Insert,
    Extract,
    Moving,
    Standby,
    Fault,
    Init,
    Running,
    Alarm,
    Disable,
    Unknown,
}

/// Source when querying an attribute.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum DevSource {
    /// Do not use the polling cache.
    Dev,
    /// Only use the polling cache.
    Cache,
    /// First use the polling cache, and fall back without it.
    CacheDev,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
pub enum DisplayLevel {
    /// The level everything is at unless it is said otherwise: what a control room shows.
    #[default]
    Operator,
    Expert,
}

// =========================== //
//           Events            //
// =========================== //

/// The events of a subscription in place, as the backend gives them, of the kind that was
/// subscribed to.
enum ChannelEvents {
    Value(BoxStream<'static, Result<AttrRead<Value>>>),
    Ready(BoxStream<'static, Result<AttrReady>>),
    Config(BoxStream<'static, Result<AttrInfo>>),
}

impl ChannelEvents {

    /// Poll the readings of an attribute, converted into the type they are expected to be.
    fn poll_value<T: FromValue>(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrRead<T>>>> {
        let Self::Value(stream) = self else { unreachable!("a value channel subscribes to values") };
        stream.poll_next_unpin(cx).map(|item| item.map(|read| read.and_then(AttrRead::parse)))
    }

    /// Poll an attribute being ready to be read.
    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrReady>>> {
        let Self::Ready(stream) = self else { unreachable!("a ready channel subscribes to readiness") };
        stream.poll_next_unpin(cx)
    }

    /// Poll the configuration of an attribute.
    fn poll_config(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrInfo>>> {
        let Self::Config(stream) = self else { unreachable!("a configuration channel subscribes to configurations") };
        stream.poll_next_unpin(cx)
    }

}

/// What a [`DevChannel`] subscribes to.
#[derive(Debug, Clone, Copy)]
enum ChannelKind {
    Value(AttrValueEvent),
    Ready,
    Config,
}

/// The events of a subscription to a device, which unsubscribes when it is dropped: it is
/// a [`Stream`] for an asynchronous caller, and [`Self::wait`] blocks for a synchronous one.
///
/// It gives the latest event and not every one of them, a subscriber that does not keep up
/// skipping to the latest, and never ends. An item that is an error is one event that
/// failed, such as the reading the device was to push, or the subscription failing to be
/// made or being lost.
///
/// # Subscribing
///
/// The subscription is made lazily, the first time the channel is polled, a failure being
/// given as an item like any other. A subscriber that wants to know that it is in place
/// before it waits for events ensures it, see [`Self::ensure_subscribed_async`].
///
/// The subscription of a backend lives as long as the connection it was made over, see
/// [`BackendDev`], while the device may be restarted, elsewhere even. The channel holds
/// the proxy of the device for that, which finds it again the way it does for any
/// request: once the subscription is lost, which the backend tells with the very errors a
/// request fails with when its connection is lost, the proxy forgets its connection the
/// way it does for a request, and the channel makes the subscription again through it.
/// The admin device, which subscribing goes through in Tango, is found again along with
/// it: its server is the one of the device, which may have moved with it.
///
/// Every attempt is made on the poll after the one that gave the failure of the previous
/// one, so that a subscriber retries as fast as it reads. The events then come back on
/// their own, with no initial value either: read the attribute once the channel gave an
/// error to know where it stands after it.
pub struct DevChannel<T> {
    /// The device, which is moved into an attempt at subscribing for as long as it runs,
    /// and back once it is over.
    dev: Option<DevProxy>,
    /// The name of the attribute subscribed to.
    name: Arc<str>,
    kind: ChannelKind,
    /// Takes an event out of the events of the kind subscribed to, see [`ChannelEvents`].
    poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
    /// The timeout of the device, which bounds the blocking calls of the channel.
    timeout: Duration,
    state: ChannelState,
}

enum ChannelState {
    /// Giving the events of a subscription in place.
    Subscribed(ChannelEvents),
    /// Making the subscription, the device being given back with the outcome. The attempt
    /// is only made once it is polled.
    Subscribing(BoxFuture<'static, (DevProxy, Result<ChannelEvents>)>),
}

impl ChannelState {

    /// An attempt at making the subscription, which goes through the admin device of the
    /// device, found again on every attempt, see the [`BackendDev`] subscriptions.
    fn subscribing(dev: DevProxy, name: Arc<str>, kind: ChannelKind) -> Self {
        Self::Subscribing(Box::pin(async move {
            let res = async {
                let backend = dev.ensure_backend().await?;
                let admin = dev.admin_dev_async().await?;
                let admin = admin.ensure_backend().await?;
                let res = match kind {
                    ChannelKind::Value(event) => backend.subscribe_attr_value(admin, Scalar::from(&*name), event).await.map(ChannelEvents::Value),
                    ChannelKind::Ready => backend.subscribe_attr_ready(admin, Scalar::from(&*name)).await.map(ChannelEvents::Ready),
                    ChannelKind::Config => backend.subscribe_attr_config(admin, Scalar::from(&*name)).await.map(ChannelEvents::Config),
                };
                dev.check_backend(res)
            }.await;
            (dev, res)
        }))
    }

}

impl<T> DevChannel<T> {

    /// A channel to an attribute of the device, subscribed to once it is polled, its events
    /// being taken out of those of the kind with the given function.
    fn new(
        dev: DevProxy,
        name: &str,
        kind: ChannelKind,
        poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
    ) -> Self {
        let timeout = dev.timeout();
        let name = Arc::<str>::from(name);
        let state = ChannelState::subscribing(dev, Arc::clone(&name), kind);
        Self { dev: None, name, kind, poll_events, timeout, state }
    }

    /// Start an attempt at making the subscription again.
    fn resubscribe(&mut self) {
        let dev = self.dev.take().expect("the device is moved into one attempt at a time");
        self.state = ChannelState::subscribing(dev, Arc::clone(&self.name), self.kind);
    }

    /// Drive the attempt at subscribing to its outcome, if one is running, the next one
    /// being made on the next poll if it failed.
    fn poll_subscribed(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {

        let ChannelState::Subscribing(attempt) = &mut self.state else {
            return Poll::Ready(Ok(()));
        };

        let (dev, res) = ready!(attempt.as_mut().poll(cx));
        let dev = &*self.dev.insert(dev);

        Poll::Ready(match res {
            Ok(events) => {
                debug!(dev = %dev.inner.target, attr = %self.name, event = ?self.kind, "Subscribed to the events");
                self.state = ChannelState::Subscribed(events);
                Ok(())
            }
            Err(err) => {
                debug!(dev = %dev.inner.target, attr = %self.name, event = ?self.kind, %err, "Failed to subscribe to the events, retrying on the next poll");
                self.resubscribe();
                Err(err)
            }
        })

    }

    /// Make the subscription if it is not in place, giving the failure of the attempt if
    /// it failed: the failure is then not given as an item, and the next attempt is made
    /// on the next poll.
    pub async fn ensure_subscribed_async(&mut self) -> Result<()> {
        poll_fn(|cx| self.poll_subscribed(cx)).await
    }

    /// Make the subscription if it is not in place, see [`Self::ensure_subscribed_async`],
    /// at most for the timeout of the device.
    pub fn ensure_subscribed(&mut self) -> Result<()> {
        let timeout = self.timeout;
        block_on_timeout(self.ensure_subscribed_async(), timeout)?
    }

    /// Wait for the next event, at most for that long, failing with
    /// [`ErrorKind::TimedOut`] when none came.
    pub fn wait(&mut self, timeout: Duration) -> Result<T> {
        match block_on_timeout(self.next(), timeout)? {
            Some(item) => item,
            None => unreachable!("a device channel never ends"),
        }
    }

}

impl<T> Stream for DevChannel<T> {

    type Item = Result<T>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {

        let this = self.get_mut();

        loop {

            if let Err(err) = ready!(this.poll_subscribed(cx)) {
                return Poll::Ready(Some(Err(err)));
            }

            let ChannelState::Subscribed(events) = &mut this.state else {
                unreachable!("subscribed once the attempt succeeded");
            };

            let dev = this.dev.as_ref().expect("the device is back once subscribed");

            match ready!((this.poll_events)(events, cx)) {
                Some(Err(err)) if err.kind() == ErrorKind::DevNotConnected => {
                    debug!(dev = %dev.inner.target, attr = %this.name, event = ?this.kind, %err, "Lost the subscription to the events, subscribing again on the next poll");
                    dev.reset_backend();
                    this.resubscribe();
                    return Poll::Ready(Some(Err(err)));
                }
                Some(item) => return Poll::Ready(Some(item)),
                // A backend ending its stream without saying why.
                None => {
                    debug!(dev = %dev.inner.target, attr = %this.name, event = ?this.kind, "The subscription to the events ended, subscribing again");
                    this.resubscribe();
                }
            }

        }

    }

}

impl<T> Debug for DevChannel<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DevChannel")
            .field("dev", &self.dev)
            .field("name", &self.name)
            .finish_non_exhaustive()
    }
}

/// Kind of event for a device attribute.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AttrValueEvent {
    Change,
    Periodic,
    Archive,
    User,
    Alarm,
}

/// The failure of a device answering with a number of entries that is not the one it was
/// asked for, which a form of a fixed shape has no output to build out of: the device broke
/// what the protocol says of it, so the request fails as a whole rather than an entry of it.
#[cold]
pub(crate) fn unexpected_len_err(expected: usize) -> Error {
    Error::new(ErrorKind::Internal)
        .with_description(format!("Device answered a different number of entries where {expected} were requested"))
}

/// The failure of what was being done, described as such, of the kind and severity of the
/// failure it comes from, which is its cause: a kind other callers key on, such as the
/// reconnection of [`DevProxy`] does, is kept.
#[cold]
#[track_caller]
fn context_err(err: Error, description: impl Into<Cow<'static, str>>) -> Error {
    Error::new(err.kind())
        .with_description(description)
        .with_severity(err.severity())
        .with_parent(err)
}

/// What the failure of a command is turned into: the reasons cppTango answers with are
/// mapped onto the kinds of this crate, and an argument type that did not match carries
/// the kind that was expected, read back out of the description, see
/// [`expected_cpp_tango_type`].
fn command_err(err: Error) -> Error {
    match err.kind() {
        ErrorKind::Reason(reason::API_CommandNotFound) => {
            err.with_kind(ErrorKind::CommandNotFound)
        }
        ErrorKind::Reason(reason::API_IncompatibleCmdArgumentType) => {

            let expected_kind = expected_cpp_tango_type(err.description())
                .and_then(command_data_kind_from_cpp_tango);

            let err = err.with_kind(ErrorKind::CommandArgumentKindInvalid);
            match expected_kind {
                Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
                None => err,
            }

        }
        _ => err,
    }
}

/// What the failure of one attribute write is turned into, see [`command_err`].
fn write_attr_err(err: Error) -> Error {
    match err.kind() {
        ErrorKind::Reason(reason::API_AttrNotFound) => {
            err.with_kind(ErrorKind::AttrNotFound)
        }
        ErrorKind::Reason(reason::API_AttrNotWritable) => {
            err.with_kind(ErrorKind::AttrNotWritable)
        }
        ErrorKind::Reason(reason::API_IncompatibleAttrDataType) => {

            let expected_kind = expected_cpp_tango_type(err.description())
                .and_then(attr_data_kind_from_cpp_tango);

            let err = err.with_kind(ErrorKind::AttrKindInvalid);
            match expected_kind {
                Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
                None => err,
            }

        }
        _ => err,
    }
}

/// Extract the type cppTango says it was expecting out of an error description.
///
/// The relevant descriptions all name the expected type first, prefixed by the `Tango::`
/// namespace, and the type actually found (when they mention it) only after:
/// - `Incompatible command argument type, expected type is : Tango::DevBoolean`
///   (`Command::throw_bad_type`)
/// - `Incompatible attribute type: expected Tango::DevVarDoubleArray (even for single
///   value), found Tango::DevVarStringArray` (`WAttribute::check_written_value`)
///
/// So the first `Tango::`-prefixed word is the expected one, and it's trimmed of any
/// trailing punctuation such as the comma or the null byte cppTango's `std::ends` leaves
/// behind on some messages.
fn expected_cpp_tango_type(description: &str) -> Option<&str> {
    description.split_ascii_whitespace()
        .find_map(|word| word.strip_prefix("Tango::"))
        .map(|ty| ty.trim_matches(|c: char| !c.is_ascii_alphanumeric()))
}

/// Try to parse the given cpp tango data type into a standard [`CommandDataKind`].
///
/// The names are the literals given to `Command::throw_bad_type` in cppTango, which match
/// the `CmdArgTypeName` table.
fn command_data_kind_from_cpp_tango(cpp_type: &str) -> Option<CommandDataKind> {
    Some(match cpp_type {
        "DevVoid" => CommandDataKind::Void,
        "DevBoolean" => CommandDataKind::Bool,
        "DevUChar" => CommandDataKind::Octet,
        "DevShort" => CommandDataKind::Short,
        "DevUShort" => CommandDataKind::UShort,
        "DevLong" => CommandDataKind::Long,
        "DevULong" => CommandDataKind::ULong,
        "DevLong64" => CommandDataKind::Long64,
        "DevULong64" => CommandDataKind::ULong64,
        "DevFloat" => CommandDataKind::Float,
        "DevDouble" => CommandDataKind::Double,
        "DevState" => CommandDataKind::State,
        "DevString" |
        "ConstDevString" => CommandDataKind::String,
        "DevVarBooleanArray" => CommandDataKind::BoolArray,
        "DevVarCharArray" => CommandDataKind::OctetArray,
        "DevVarShortArray" => CommandDataKind::ShortArray,
        "DevVarUShortArray" => CommandDataKind::UShortArray,
        "DevVarLongArray" => CommandDataKind::LongArray,
        "DevVarULongArray" => CommandDataKind::ULongArray,
        "DevVarLong64Array" => CommandDataKind::Long64Array,
        "DevVarULong64Array" => CommandDataKind::ULong64Array,
        "DevVarFloatArray" => CommandDataKind::FloatArray,
        "DevVarDoubleArray" => CommandDataKind::DoubleArray,
        "DevVarStateArray" => CommandDataKind::StateArray,
        "DevVarStringArray" => CommandDataKind::StringArray,
        "DevVarLongStringArray" => CommandDataKind::LongStringArray,
        "DevVarDoubleStringArray" => CommandDataKind::DoubleStringArray,
        "DevEncoded" => CommandDataKind::Encoded,
        _ => return None,
    })
}

/// Try to parse the given cpp tango data type into a standard [`AttrDataKind`].
///
/// An attribute value is always carried as a sequence on the wire, so cppTango always
/// names the array type here, even for a scalar attribute; the scalar spellings are
/// accepted anyway because [`AttrDataKind`] carries no layout of its own and other
/// cppTango versions word it differently.
///
/// Note that an enumerated attribute is reported as `DevVarShortArray` by cppTango, so it
/// is parsed as [`AttrDataKind::Short`]: the labels needed to build
/// [`AttrDataKind::Enum`] are simply not in the message.
fn attr_data_kind_from_cpp_tango(cpp_type: &str) -> Option<AttrDataKind> {
    Some(match cpp_type {
        "DevBoolean" | "DevVarBooleanArray" => AttrDataKind::Bool,
        "DevUChar" | "DevVarCharArray" => AttrDataKind::Octet,
        "DevShort" | "DevVarShortArray" => AttrDataKind::Short,
        "DevUShort" | "DevVarUShortArray" => AttrDataKind::UShort,
        "DevLong" | "DevVarLongArray" => AttrDataKind::Long,
        "DevULong" | "DevVarULongArray" => AttrDataKind::ULong,
        "DevLong64" | "DevVarLong64Array" => AttrDataKind::Long64,
        "DevULong64" | "DevVarULong64Array" => AttrDataKind::ULong64,
        "DevFloat" | "DevVarFloatArray" => AttrDataKind::Float,
        "DevDouble" | "DevVarDoubleArray" => AttrDataKind::Double,
        "DevState" | "DevVarStateArray" => AttrDataKind::State,
        "DevString" | "ConstDevString" | "DevVarStringArray" => AttrDataKind::String,
        "DevEncoded" | "DevVarEncodedArray" => AttrDataKind::Encoded,
        _ => return None,
    })
}

/// Display the host and port pairs of [`env_databases`] the way they are written in
/// `TANGO_HOST`, so that an error naming them can be compared to the variable itself.
struct DisplayHostAndPorts<'a>(&'a [(Arc<str>, u16)]);

impl Display for DisplayHostAndPorts<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        for (i, (host, port)) in self.0.iter().enumerate() {
            if i != 0 {
                f.write_str(", ")?;
            }
            write!(f, "{host}:{port}")?;
        }
        Ok(())
    }
}

/// Get the database host/port tuples from the `TANGO_HOST` environment variable.
fn env_databases() -> &'static [(Arc<str>, u16)] {
    static CACHE: OnceLock<Box<[(Arc<str>, u16)]>> = OnceLock::new();
    CACHE.get_or_init(|| {

        let mut ret = Vec::new();

        if let Ok(var) = env::var("TANGO_HOST") {
            for part in var.split(',') {
                if let Some((host, port_raw)) = part.rsplit_once(':')
                && let Ok(port) = u16::from_str(port_raw) {
                    ret.push((host.into(), port));
                }
            }
        }
        
        ret.into_boxed_slice()

    })
}


#[cfg(test)]
mod tests {

    use crate::value::Scalar;

    use super::*;

    /// The property requests of a proxy, awaited within a future boxed as `Send`: this only
    /// has to compile.
    #[allow(dead_code)]
    fn props_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
        Box::pin(async move {
            let _speed = proxy.get_prop_async::<f64, _>("speed").await?;
            let _all = proxy.get_prop_async::<Spectrum<str>, _>(All).await?;
            proxy.set_prop_async(("speed", 2.5)).await?;
            proxy.delete_prop_async(["speed", "accel"]).await
        })
    }

    /// The requests of a proxy, awaited within a future boxed as `Send`, which is what an
    /// asynchronous handler forwarding to another device is: this only has to compile.
    #[allow(dead_code)]
    fn requests_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
        Box::pin(async move {
            let _: f64 = proxy.command_async("Add", 1.0).await?;
            let _: String = proxy.command_async("Echo", "text").await?;
            let _ = proxy.read_attr_async::<f64, _>("speed").await?;
            let _ = proxy.read_attr_async::<f64, _>(["speed", "accel"]).await?;
            let _ = proxy.read_attr_async::<Value, _>(All).await?;
            proxy.write_attr_async(("speed", 2.5)).await?;
            let _ = proxy.write_attr_async([("speed", 2.5), ("accel", 1.0)]).await?;
            let _ = proxy.write_read_attr_async(("speed", 2.5)).await?;
            let _ = proxy.command_info_async("Add").await?;
            let _ = proxy.attr_info_async(["speed"]).await?;
            let names = proxy.list_prop_async("*").await?;
            let _ = proxy.read_attr_async::<f64, _>(&names).await?;
            let _ = proxy.attr_info_async(names).await?;
            Ok(())
        })
    }

    /// A property request is `Send` whatever the names it is given are, since its future
    /// holds nothing of them but the copy it makes of them: this only has to compile.
    #[allow(dead_code)]
    fn props_of_any_names<'a, Names>(proxy: &'a DevProxy, names: Names) -> BoxFuture<'a, Result<<f64 as GetPropSpec<Names>>::Output>>
    where
        Names: ToNames + 'a,
        f64: GetPropSpec<Names>,
    {
        Box::pin(proxy.get_prop_async::<f64, Names>(names))
    }

    #[test]
    fn expected_kind() {

        // Verbatim from cppTango 'Command::throw_bad_type', trailing null byte included
        // because the message is built with 'std::ends'.
        let cmd = "Incompatible command argument type, expected type is : Tango::DevVarDoubleStringArray\0";
        assert_eq!(expected_cpp_tango_type(cmd), Some("DevVarDoubleStringArray"));
        assert_eq!(
            expected_cpp_tango_type(cmd).and_then(command_data_kind_from_cpp_tango),
            Some(CommandDataKind::DoubleStringArray),
        );

        // Verbatim from cppTango 'WAttribute::check_written_value', which names the found
        // type after the expected one: only the expected one should be picked up.
        let attr = "Incompatible attribute type: expected Tango::DevVarDoubleArray (even for single value), found Tango::DevVarStringArray\0";
        assert_eq!(expected_cpp_tango_type(attr), Some("DevVarDoubleArray"));
        assert_eq!(
            expected_cpp_tango_type(attr).and_then(attr_data_kind_from_cpp_tango),
            Some(AttrDataKind::Double),
        );

        assert_eq!(attr_data_kind_from_cpp_tango("DevVarEncodedArray"), Some(AttrDataKind::Encoded));
        assert_eq!(command_data_kind_from_cpp_tango("DevEncoded"), Some(CommandDataKind::Encoded));

        // Unrelated descriptions and unsupported types should not yield any kind.
        assert_eq!(expected_cpp_tango_type("Attribute my_attr is not writable"), None);
        assert_eq!(command_data_kind_from_cpp_tango("DevVarEncodedArray"), None);

    }

    /// Readings decided ahead of time, standing in for the ones a backend would have
    /// decoded out of what a device answered.
    fn reads<const N: usize>(entries: [(&str, Value); N]) -> Vec<NamedResult<AttrRead<Value>>> {
        entries.into_iter()
            .map(|(name, value)| Named::new(Scalar::from(name), Ok(AttrRead::new(value))))
            .collect()
    }

    /// One name asks for one reading, which is the output itself: there is no slot around
    /// it, so what is given back is that reading and nothing else.
    #[test]
    fn single_reading() {
        let reads = reads([("double_scalar", Value::from(2.5))]);
        let read = <f64 as ReadAttrSpec<&str>>::collect(reads).unwrap();
        assert_eq!(read.value, Some(2.5));
    }

    /// The readings come back in the order the names were given, and a form of a fixed
    /// shape has one slot per name.
    #[test]
    fn readings_in_order() {
        let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
        let [first, second] = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap();
        assert_eq!(first.unwrap().value, Some(1.0));
        assert_eq!(second.unwrap().value, Some(2.0));
    }

    /// A tuple of names gives every reading the type of its own component, so a device
    /// answering with a double and a string is read into both at once.
    #[test]
    fn readings_of_their_own_types() {
        let reads = reads([("double_scalar", Value::from(2.5)), ("string_scalar", Value::from("hello"))]);
        let (double, string) = <(f64, String) as ReadAttrSpec<(&str, &str)>>::collect(reads).unwrap();
        assert_eq!(double.unwrap().value, Some(2.5));
        assert_eq!(string.unwrap().value.as_deref(), Some("hello"));
    }

    /// A device answering with fewer readings than there are slots to put them in fails the
    /// whole request: the output of a fixed shape cannot be built out of what came back.
    #[test]
    fn missing_reading() {
        let reads = reads([("first", Value::from(1.0))]);
        let err = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Internal);
    }

    /// Reading every attribute of a device names every reading back, since nothing named
    /// them on the way out.
    #[test]
    fn every_reading_named() {
        let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
        let all = <Value as ReadAttrSpec<All>>::collect(reads).unwrap();
        let names = all.iter().map(|read| read.name.value()).collect::<Vec<_>>();
        assert_eq!(names, ["first", "second"]);
    }

    /// The outcomes of a write come back in the order the values were given, one per value
    /// and named by none of them.
    #[test]
    fn write_outcomes() {
        let writes = vec![Ok(()), Err(Error::new(ErrorKind::AttrNotWritable))];
        let (first, second) = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap();
        assert!(first.is_ok());
        assert_eq!(second.unwrap_err().kind(), ErrorKind::AttrNotWritable);
    }

    /// A device answering for a number of values that is not the one that was written
    /// fails the whole request, the same way it does for a read.
    #[test]
    fn unexpected_outcome() {
        let writes = vec![Ok(())];
        let err = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Internal);
    }

}