tango/corba/

mod.rs

//! The CORBA backend, the protocol every Tango implementation speaks, with the ZMQ event system
//! as part of it.

mod orb;
mod tango;
mod zmq;

use std::ops::ControlFlow;
use std::sync::{Arc, OnceLock};
use std::time::SystemTime;
use std::future::Future;
use std::borrow::Cow;
use std::pin::Pin;
use std::task;

use futures::future::{ready, BoxFuture};
use futures::stream::BoxStream;
use tracing::{debug, trace, warn};
use uuid::Uuid;

use crate::dev::{AttrConfig, AttrInfo, AttrQuality, AttrRead, AttrReady, AttrValueEvent, CommandInfo, DevInfo, DevSource, DevState, Named, NamedResult, RecordedResult};
use crate::backend::{Backend, BackendDev, BackendServer, BackendServerConfig};
use crate::error::{reason, Error, ErrorKind, Result};
use crate::value::{Scalar, Spectrum, Value};
use crate::future::resolve;

use orb::client::{Client, ClientObject, ClientRequest, ClientReply};
use orb::server::{Server, ServerObject, ServerRequest, ServerReply};
use orb::cdr::{CdrWrite, CdrRead, CdrReader, CdrWriter};
use orb::giop::ReplyStatus;
use zmq::{Consumer, Supplier};
use tango::{
    tango_repo_id, 
    ClientIdent,
    NamedError,
    write_command_data, read_command_data,
    read_attr, write_attr,
    read_command_info, read_attr_info,
    read_dev_info, write_dev_info,
    write_command_info, write_attr_info,
    read_command_history, write_command_history,
    read_attr_history, write_attr_history,
};


/// Special name for reading all attributes.
const ALL_ATTRIBUTES: &str = "All attributes";
/// Special name for reading all attributes on IDL v3+.
const ALL_ATTRIBUTES_3: &str = "All attributes_3";

/// Static client identifier.
static CLIENT_ID: ClientIdent = ClientIdent::Java { 
    main_class: Cow::Borrowed("rstango"), 
    uuid: Uuid::nil(),
};

// =========================== //
//        Common backend       //
// =========================== //

/// A pure rust CORBA backend.
#[derive(Debug)]
pub struct CorbaBackend {
    /// The ORB client.
    client: OnceLock<Client>,
    /// The event consumer, shared with every device connected through this backend, and
    /// only started by the first subscription.
    consumer: Arc<OnceLock<Consumer>>,
}

impl CorbaBackend {

    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            client: OnceLock::new(),
            consumer: Arc::new(OnceLock::new()),
        })
    }

    fn ensure_client(&self) -> Result<&Client> {

        if let Some(client) = self.client.get() {
            return Ok(client);
        }

        let client = Client::new()?;
        Ok(self.client.get_or_init(|| client))

    }

    /// Find the most recent device IDL version the object implements and build the device
    /// backend speaking it, failing with [`ErrorKind::DevNotConnected`] if it implements none.
    async fn check_device(&self, object: ClientObject) -> Result<Arc<dyn BackendDev>> {

        // Iterate over the versions to find the largest one supported. We iterate in
        // reverse order because nowadays it's much more likely to encounter a recent
        // version than an old one!
        let mut versions_it = DevVersion::ALL.into_iter().rev();
        let version = loop {

            let Some(version) = versions_it.next() else {
                return Err(Error::new(ErrorKind::DevNotConnected)
                    .with_description(format!("Object \"{}\" implements no supported version of the device interface",
                        object.object().escape_ascii())));
            };

            let reply = object.request(b"_is_a", &version.repo_id()).await.map_err(corba_err)?;
            let (mut reader, exception_id) = reply.read().map_err(corba_err)?;

            if let Some(exception_id) = exception_id.as_deref() {
                return Err(unexpected_user_exception_err(exception_id));
            }

            if reader.read_bool()? {
                break version;
            }

        };
        
        debug!(?version, "Connected to the device over CORBA");

        Ok(Arc::new(CorbaDevBackend {
            object,
            version,
            consumer: Arc::clone(&self.consumer),
        }))

    }

}

impl Backend for CorbaBackend {

    fn connect(&self, itr: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>> {
        if let Some(_corbaloc) = itr.strip_prefix("corbaloc:iiop:") {
            Box::pin(ready(Err(Error::not_implemented()
                .with_description("Scheme corbaloc:iiop: is not implemented"))))
        } else if let Some(ior) = itr.strip_prefix("IOR:") {
            let ior = ior.to_string();
            Box::pin(async move {
                let client = self.ensure_client()?;
                let object = client.connect_object_ior(&ior, &DevVersion::REPO_ID).await.map_err(corba_err)?;
                self.check_device(object).await
            })
        } else {
            Box::pin(ready(Err(Error::new(ErrorKind::NotHandled))))
        }
    }

    fn connect_at(&self, host: &str, port: u16, name: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>> {
        let host = host.to_string();
        let object: Arc<[u8]> = Arc::from(name.to_ascii_lowercase().into_bytes());
        Box::pin(async move {
            let client = self.ensure_client()?;
            let object = client.connect_object_iiop(&host, port, None, None, object).await.map_err(corba_err)?;
            self.check_device(object).await
        })
    }

    fn serve<'a>(&'a self, config: &'a BackendServerConfig) -> BoxFuture<'a, Result<Arc<dyn BackendServer>>> {
        Box::pin(async move {

            // The port the server is bound to is what the references it hands out then
            // point at, see [`CorbaServerBackend::build_ior`].
            let host = config.host.as_deref().unwrap_or("0.0.0.0");
            let ports = config.port.unwrap_or((0, 0));

            let addrs = resolve(host).await?;

            // A name is bound at the first of its addresses that can be, as the standard
            // library does.
            let mut last_err = None;
            for addr in addrs {
                match Server::bind(addr, ports) {
                    Ok(server) => {

                        debug!(addr = %server.local_addr(), addrs = ?server.addrs(), "Serving devices over CORBA");
                        
                        let supplier = config.admin_name.as_deref()
                            .map(|admin_name| Supplier::new(
                                admin_name,
                                config.tango_host.as_ref().map(|(host, port)| (&**host, *port)),
                                &server.host(),
                                server.local_addr().port(),
                                server.local_addr().ip(),
                                config.zmq_heartbeat_port.unwrap_or((0, 0)),
                                config.zmq_event_port.unwrap_or((0, 0)),
                                server.addrs(),
                            ));

                        return Ok(Arc::new(CorbaServerBackend { server, supplier }) as Arc<dyn BackendServer>);

                    }
                    Err(err) => last_err = Some(err),
                }
            }

            Err(last_err.unwrap_or_else(|| Error::new(ErrorKind::Corba)
                .with_description(format!("Host {host:?} resolves to no address to bind on ports {}..={}", ports.0, ports.1))))

        })
    }

}

// =========================== //
//        Client backend       //
// =========================== //

/// Our internal backend to device proxies.
#[derive(Debug)]
struct CorbaDevBackend {
    object: ClientObject,
    version: DevVersion,
    /// The event consumer of the backend, see [`Self::consumer`].
    consumer: Arc<OnceLock<Consumer>>,
}

impl CorbaDevBackend {

    /// Send a request to this device, over the object's own connection.
    ///
    /// The request goes out as soon as this returns, and not once the future it gives
    /// back is polled, see [`ClientObject::request`]: whatever was borrowed to build
    /// `input` need not outlive this call, only the future it returns has to be awaited.
    fn request(&self, operation: &[u8], input: &dyn CdrWrite) -> CorbaDevRequest {
        CorbaDevRequest { inner: self.object.request(operation, input) }
    }

    /// The event consumer of the backend, started if this is the first subscription. Two
    /// first subscriptions at once may both start one, the one that is not kept stopping
    /// as soon as it is dropped.
    fn consumer(&self) -> Result<&Consumer> {

        if let Some(consumer) = self.consumer.get() {
            return Ok(consumer);
        }

        let consumer = Consumer::start()?;
        Ok(self.consumer.get_or_init(|| consumer))

    }

}

impl BackendDev for CorbaDevBackend {

    fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
        let request = self.request(b"_get_name", &());
        Box::pin(async move {
            let reply = request.await?;
            Ok(Arc::from(reply.read::<String>()?))
        })
    }

    fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
        let request = self.request(b"_get_adm_name", &());
        Box::pin(async move {
            let reply = request.await?;
            Ok(Arc::from(reply.read::<String>()?))
        })
    }

    fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
        let request = self.request(b"_get_description", &());
        Box::pin(async move {
            let reply = request.await?;
            Ok(Arc::from(reply.read::<String>()?))
        })
    }

    fn ping(&self) -> BoxFuture<'_, Result<()>> {
        let request = self.request(b"ping", &());
        Box::pin(async move {
            let reply = request.await?;
            reply.read::<()>()
        })
    }

    fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {

        // IDL variants:
        // DevInfo info()
        // DevInfo_3 info_3()
        // DevInfo_6 info_6()
        //
        // The second member is the version of the structure the info is read in, which is
        // the one of the operation and not the one of the device: v2 answers with the v1
        // structure, and v4 and v5 with the v3 one.
        let (operation, info_version): (&[u8], DevVersion) = match self.version {
            DevVersion::V1 |
            DevVersion::V2 => (b"info", DevVersion::V1),
            DevVersion::V3 |
            DevVersion::V4 |
            DevVersion::V5 => (b"info_3", DevVersion::V3),
            DevVersion::V6 => (b"info_6", DevVersion::V6),
        };

        let request = self.request(operation, &());

        Box::pin(async move {
            let reply = request.await?;
            let mut reader = reply.read_simple()?;
            read_dev_info(&mut reader, info_version)
        })

    }

    fn state(&self) -> BoxFuture<'_, Result<DevState>> {
        let request = self.request(b"_get_state", &());
        Box::pin(async move {
            let reply = request.await?;
            reply.read::<DevState>()
        })
    }

    fn status(&self) -> BoxFuture<'_, Result<Scalar<str>>> {
        let request = self.request(b"_get_status", &());
        Box::pin(async move {
            let reply = request.await?;
            reply.read::<Scalar<str>>()
        })
    }

    fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Spectrum<str>>> {
        let max = max.and_then(|n| i32::try_from(n).ok()).unwrap_or(i32::MAX);
        let request = self.request(b"black_box", &max);
        Box::pin(async move {
            let reply = request.await?;
            reply.read::<Spectrum<str>>()
        })
    }

    fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {

        // IDL variants:
        // any command_inout(in string command, in any argin)
        // any command_inout_2(in string command, in any argin, in DevSource source)
        // any command_inout_4(in string command, in any argin, in DevSource source, in ClntIdent cl_ident)

        struct Input<'a> {
            version: DevVersion,
            name: &'a str,
            input: Value,
            source: DevSource,
        }

        impl CdrWrite for Input<'_> {
            fn write(&self, buf: &mut CdrWriter) -> Result<()> {
                buf.write_string(self.name)?;
                write_command_data(buf, &self.input)?;
                if self.version >= DevVersion::V2 {
                    buf.write(&self.source)?;
                }
                if self.version >= DevVersion::V4 {
                    buf.write(&CLIENT_ID)?;
                }
                Ok(())
            }
        }

        let operation: &[u8] = match self.version {
            DevVersion::V1 => b"command_inout",
            DevVersion::V2 |
            DevVersion::V3 => b"command_inout_2",
            DevVersion::V4 |
            DevVersion::V5 |
            DevVersion::V6 => b"command_inout_4",
        };

        let request = self.request(operation, &Input { version: self.version, name: name.value(), input, source });

        Box::pin(async move {
            let reply = request.await?;
            read_command_data(&mut reply.read_simple()?)
        })

    }

    fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {

        // IDL variants:
        // AttributeValueList read_attributes(in DevVarStringArray names)
        // AttributeValueList read_attributes_2(in DevVarStringArray names, in DevSource source)
        // AttributeValueList_3 read_attributes_3(in DevVarStringArray names, in DevSource source)
        // AttributeValueList_4 read_attributes_4(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)
        // AttributeValueList_5 read_attributes_5(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)

        struct Input<'a> {
            version: DevVersion,
            names: &'a dyn CdrWrite,
            source: DevSource,
        }

        impl CdrWrite for Input<'_> {
            fn write(&self, buf: &mut CdrWriter) -> Result<()> {
                buf.write(&self.names)?;
                if self.version >= DevVersion::V2 {
                    buf.write(&self.source)?;
                }
                if self.version >= DevVersion::V4 {
                    buf.write(&CLIENT_ID)?;
                }
                Ok(())
            }
        }
        
        if let Some(names) = &names
        && let Err(err) = check_no_all_marker(names) {
            return Box::pin(ready(Err(err)));
        }

        let operation: &[u8] = match self.version {
            DevVersion::V1 => b"read_attributes",
            DevVersion::V2 => b"read_attributes_2",
            DevVersion::V3 => b"read_attributes_3",
            DevVersion::V4 => b"read_attributes_4",
            DevVersion::V5 |
            DevVersion::V6 => b"read_attributes_5",
        };

        let version = self.version;
        let names = match &names {
            Some(names) => names as &dyn CdrWrite,
            None => &[ALL_ATTRIBUTES],
        };

        let request = self.request(operation, &Input { version, names, source });

        Box::pin(async move {
            let reply = request.await?;
            read_attr_list(&mut reply.read_simple()?, version)
        })

    }

    fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {

        // IDL variants:
        // void write_attributes(in AttributeValueList values)
        // void write_attributes_3(in AttributeValueList values)
        // void write_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
        //
        // Note that v2 has no variant of its own and reuses the v1 one, and that v3 only
        // adds MultiDevFailed to the raised exceptions, the argument staying the same.

        struct Input<'a> {
            /// The version of the structure the values go out in, which is the one of the
            /// operation and *not* the one of the device, see below.
            version: DevVersion,
            values: &'a [Named<Value>],
        }

        impl CdrWrite for Input<'_> {
            fn write(&self, buf: &mut CdrWriter) -> Result<()> {
                buf.write_sequence_len(self.values.len())?;
                for Named { name, value } in self.values {
                    write_attr(buf, self.version, name.value(), &Ok(AttrRead {
                        value: None,
                        w_value: Some(value.clone()),
                        quality: AttrQuality::Valid,
                        time: SystemTime::UNIX_EPOCH,
                    }))?;
                }
                if self.version >= DevVersion::V4 {
                    buf.write(&CLIENT_ID)?;
                }
                Ok(())
            }
        }

        // The second member is the version of the structure the values are written in,
        // which is the one of the operation and not the one of the device: v3 takes the v1
        // structure like v1 and v2 do, and v5 and v6 have no write operation of their own
        // so they take the v4 one. Sending the structure of the device version instead
        // makes the server fail to unmarshal the request entirely.
        let (operation, input_version): (&[u8], DevVersion) = match self.version {
            DevVersion::V1 |
            DevVersion::V2 => (b"write_attributes", DevVersion::V1),
            DevVersion::V3 => (b"write_attributes_3", DevVersion::V1),
            DevVersion::V4 |
            DevVersion::V5 |
            DevVersion::V6 => (b"write_attributes_4", DevVersion::V4),
        };

        let request = self.request(operation, &Input { version: input_version, values: &values });

        let count = values.len();

        Box::pin(async move {

            let reply = request.await?;
            let (mut reader, exception_id) = reply.read_raw()?;

            let Some(exception_id) = exception_id else {
                return Ok((0..count).map(|_| Ok(())).collect());
            };

            if exception_id.as_ref() != tango_repo_id!("MultiDevFailed") {
                return Err(unexpected_user_exception_err(&exception_id));
            }

            // A `MultiDevFailed` only names the attributes that failed to write, the
            // ones behind them in the list having gone through: unlike a `DevFailed`,
            // this is per-value and not the request as a whole.
            let mut results: Vec<Result<()>> = (0..count).map(|_| Ok(())).collect();
            for failure in reader.read::<Vec<NamedError>>()? {
                if let Some(result) = results.get_mut(failure.index) {
                    *result = Err(failure.err.unwrap_or_else(|| Error::new(ErrorKind::Internal)
                        .with_description("The device reported a failed attribute write holding no error")));
                }
            }

            Ok(results)

        })

    }

    fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {

        // IDL variants:
        // AttributeValueList_4 write_read_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
        // AttributeValueList_5 write_read_attributes_5(in AttributeValueList_4 values, in DevVarStringArray r_names, in ClntIdent cl_ident)
        //
        // Note that this operation only exists from v4 onward, and that v5 is the first
        // one able to read back attributes other than the written ones. Both variants take
        // the v4 structure for what they write, only what they answer with differs.

        struct Input<'a> {
            version: DevVersion,
            values: &'a [Named<Value>],
        }

        impl CdrWrite for Input<'_> {
            fn write(&self, buf: &mut CdrWriter) -> Result<()> {

                // Hardcoded v4 and not the device's version: both operations take the very
                // same AttributeValueList_4, v5 having never added a list of its own for
                // what it writes, only for what it answers with.
                buf.write_sequence_len(self.values.len())?;
                for Named { name, value } in self.values {
                    write_attr(buf, DevVersion::V4, name.value(), &Ok(AttrRead {
                        value: None,
                        w_value: Some(value.clone()),
                        quality: AttrQuality::Valid,
                        time: SystemTime::UNIX_EPOCH,
                    }))?;
                }

                // Only v5 onward has this second list, of the attributes to read back, and
                // we have no way of telling which ones the caller wants apart from the
                // written ones, so it reads those back.
                if self.version >= DevVersion::V5 {
                    buf.write_sequence_len(self.values.len())?;
                    for write in self.values {
                        buf.write_string(write.name.value())?;
                    }
                }

                buf.write(&CLIENT_ID)?;
                Ok(())

            }
        }

        // The second member is the version the *reply* is read with, which is the one of
        // the operation and not the one of the device: v4 answers with its own list, v5
        // with the one it added, and v6 reuses the v5 operation and therefore its list.
        // So a v6 device is read back as a v5 one here, unlike everywhere else.
        let (operation, reply_version): (&[u8], DevVersion) = match self.version {
            DevVersion::V1 |
            DevVersion::V2 |
            DevVersion::V3 => {
                return Box::pin(ready(Err(Error::not_implemented()
                    .with_description("Writing and reading attributes back in a single request needs a device of IDL v4 or later")
                    .with_append_description(format_args!("Got: {:?}", self.version)))));
            }
            DevVersion::V4 => (b"write_read_attributes_4", DevVersion::V4),
            DevVersion::V5 |
            DevVersion::V6 => (b"write_read_attributes_5", DevVersion::V5),
        };

        let request = self.request(operation, &Input { version: self.version, values: &values });

        // Only needed if the write raises `MultiDevFailed`, to name the failure of every
        // value nothing more specific was reported for.
        let names = values.into_iter().map(|write| write.name).collect::<Vec<_>>();

        Box::pin(async move {

            let reply = request.await?;
            let (mut reader, exception_id) = reply.read_raw()?;

            let Some(exception_id) = exception_id else {
                return read_attr_list(&mut reader, reply_version);
            };

            if exception_id.as_ref() != tango_repo_id!("MultiDevFailed") {
                return Err(unexpected_user_exception_err(&exception_id));
            }

            // A `MultiDevFailed` here means none of the values was read back at all, so
            // every one of them is reported as a failure, the ones it does not name
            // getting a generic error of their own.
            let mut failures = reader.read::<Vec<NamedError>>()?;
            let entries = names.into_iter().enumerate().map(|(index, name)| {
                let err = failures.iter_mut()
                    .find(|failure| failure.index == index)
                    .and_then(|failure| failure.err.take())
                    .unwrap_or_else(|| Error::new(ErrorKind::Internal)
                        .with_description("The device wrote this attribute but read none of them back, another one having failed"));
                Named::new(name, Err(err))
            }).collect();

            Ok(entries)

        })

    }

    fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {

        // IDL variants:
        // DevCmdInfoList command_list_query()
        // DevCmdInfo command_query(in string command)
        // DevCmdInfoList_2 command_list_query_2()
        // DevCmdInfo_2 command_query_2(in string command)
        //
        // Note that these take no name list, so querying a subset means either one
        // request per name or filtering the whole list afterward. We do the former, which
        // is what cppTango does too. The v2 variants only add the display level to the
        // returned structures, and no version after them has one of its own.

        // The second member is the version of the structure the infos are read in, which
        // is the one of the operation and not the one of the device, every version from
        // v2 onward answering with the v2 structure.
        let (query_operation, list_operation, info_version): (&[u8], &[u8], DevVersion) = match self.version {
            DevVersion::V1 => (b"command_query", b"command_list_query", DevVersion::V1),
            _ => (b"command_query_2", b"command_list_query_2", DevVersion::V2),
        };

        if let Some(names) = names {

            // One request per name, sent right away and all in flight together, each
            // answering with the info of that very command, so there is no order to
            // check here, see [`ClientObject::request`].
            let requests = (0..names.len())
                .map(|index| names.get_scalar([index]).expect("a name at every index"))
                .map(|name| {
                    let request = self.request(query_operation, &name.value());
                    (name, request)
                })
                .collect::<Vec<_>>();

            Box::pin(async move {

                let mut entries = Vec::with_capacity(requests.len());

                for (name, request) in requests {

                    let res: Result<CommandInfo> = async {
                        let reply = request.await?;
                        let mut reader = reply.read_simple()?;
                        read_command_info(&mut reader, info_version)?.1
                    }.await;

                    // The request of a single command failing, an unknown command name
                    // being the usual reason, only fails that one command: it is
                    // reported for it and the following ones are still awaited, so
                    // that the caller gets exactly one answer for each name it gave.
                    // Only a failure that makes the rest pointless aborts the batch.
                    match res {
                        Err(err) if matches!(err.kind(), ErrorKind::TimedOut | ErrorKind::DevNotConnected) => {
                            return Err(err);
                        }
                        res => entries.push(Named::new(name, res)),
                    }

                }

                Ok(entries)

            })

        } else {

            // List everything!
            let request = self.request(list_operation, &());

            Box::pin(async move {
                let reply = request.await?;
                let mut reader = reply.read_simple()?;
                let len = reader.read_sequence_len()?;
                let mut entries = Vec::with_capacity(len);
                for _ in 0..len {
                    let (name, info) = read_command_info(&mut reader, info_version)?;
                    entries.push(Named::new(Scalar::<str>::from(&*name), info));
                }
                Ok(entries)
            })

        }

    }

    fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {

        // IDL variants:
        // void set_attribute_config(in AttributeConfigList new_conf)
        // void set_attribute_config_3(in AttributeConfigList_3 new_conf)
        // void set_attribute_config_4(in AttributeConfigList_3 new_conf, in ClntIdent cl_ident)
        // void set_attribute_config_5(in AttributeConfigList_5 new_conf, in ClntIdent cl_ident)
        //
        // Note that v2 has no variant of its own and reuses the v1 one, as v4 does with the
        // v3 structure. The structure sent is the whole one a configuration is read in,
        // what the attribute is included: its type above all, which is what its limits are
        // written as. That part is never changed by the device, so it is taken from the
        // attribute itself, asked for first, which is what a client of cppTango does as
        // well, reading a configuration before sending it back changed.

        struct Input<'a> {
            /// The version of the structure the configurations go out in.
            version: DevVersion,
            infos: &'a [(Scalar<str>, AttrInfo)],
            /// Whether the operation takes the identification of the client after them.
            with_client: bool,
        }

        impl CdrWrite for Input<'_> {
            fn write(&self, buf: &mut CdrWriter) -> Result<()> {
                buf.write_sequence_len(self.infos.len())?;
                for (name, info) in self.infos {
                    write_attr_info(buf, self.version, name.value(), info)?;
                }
                if self.with_client {
                    buf.write(&CLIENT_ID)?;
                }
                Ok(())
            }
        }

        // The second member is the version of the structure the configurations are
        // written in, which is the one of the operation and not the one of the device.
        let (operation, config_version, with_client): (&'static [u8], DevVersion, bool) = match self.version {
            DevVersion::V1 |
            DevVersion::V2 => (b"set_attribute_config", DevVersion::V1, false),
            DevVersion::V3 => (b"set_attribute_config_3", DevVersion::V3, false),
            DevVersion::V4 => (b"set_attribute_config_4", DevVersion::V3, true),
            DevVersion::V5 |
            DevVersion::V6 => (b"set_attribute_config_5", DevVersion::V5, true),
        };

        let names = configs.iter().map(|config| config.name.value()).collect::<Vec<_>>();
        let current = self.attr_info(Some(Spectrum::<str>::from(&names[..])));
        let count = configs.len();

        Box::pin(async move {

            // An attribute that cannot be described fails the whole call, which is what
            // the device would do of it anyway, a single DevFailed being all it raises.
            let mut infos = Vec::with_capacity(count);
            for (Named { name, value: config }, current) in configs.into_iter().zip(current.await?) {
                let mut info = current.value?;
                info.config = config;
                infos.push((name, info));
            }

            // The device raises a single DevFailed for the first attribute it failed to
            // change, the ones before having been changed and the ones behind not even
            // tried, which it does not tell apart: only a single attribute is known to be
            // the one it is about, and the call fails as a whole otherwise.
            let reply = self.request(operation, &Input { version: config_version, infos: &infos, with_client }).await?;
            let (mut reader, exception_id) = reply.inner.read().map_err(corba_err)?;
            match exception_id.as_deref() {
                None => Ok((0..count).map(|_| Ok(())).collect()),
                Some(tango_repo_id!("DevFailed")) => {
                    let err = reader.read::<Error>()?;
                    match count {
                        1 => Ok(vec![Err(err)]),
                        _ => Err(err),
                    }
                }
                Some(exception_id) => Err(unexpected_user_exception_err(exception_id)),
            }

        })

    }

    fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {

        // IDL variants:
        // AttributeConfigList get_attribute_config(in DevVarStringArray names)
        // AttributeConfigList_2 get_attribute_config_2(in DevVarStringArray names)
        // AttributeConfigList_3 get_attribute_config_3(in DevVarStringArray names)
        // AttributeConfigList_5 get_attribute_config_5(in DevVarStringArray names)
        //
        // Note that v4 has no variant of its own and reuses the v3 one, and that the
        // 'all attributes' marker to pass for reading them all is version-dependent
        // here, ALL_ATTRIBUTES_3 from v3 onward and ALL_ATTRIBUTES before.

        // The second member is the version of the structure the configurations are read
        // in, which is the one of the operation and not the one of the device, a v4 device
        // answering with the v3 structure and a v6 one with the v5 structure.
        let (operation, config_version): (&[u8], DevVersion) = match self.version {
            DevVersion::V1 => (b"get_attribute_config", DevVersion::V1),
            DevVersion::V2 => (b"get_attribute_config_2", DevVersion::V2),
            DevVersion::V3 |
            DevVersion::V4 => (b"get_attribute_config_3", DevVersion::V3),
            DevVersion::V5 |
            DevVersion::V6 => (b"get_attribute_config_5", DevVersion::V5),
        };

        if let Some(names) = &names
        && let Err(err) = check_no_all_marker(names) {
            return Box::pin(ready(Err(err)));
        }

        // Unlike the commands, a single request carries every name, so an unknown one
        // fails the whole request, which is the device raising and not something we could
        // report for that one attribute only.
        let request = match &names {
            Some(names) => self.request(operation, names),
            // The marker itself is what changed in v3 and not the operation taking it, so
            // it follows the version of the structure like the operation does.
            None if config_version >= DevVersion::V3 => self.request(operation, &[ALL_ATTRIBUTES_3]),
            None => self.request(operation, &[ALL_ATTRIBUTES]),
        };

        // Kept so that the order can still be checked once the reply comes back.
        let expected = names;

        Box::pin(async move {

            let reply = request.await?;
            let mut reader = reply.read_simple()?;
            let len = reader.read_sequence_len()?;
            let mut entries = Vec::with_capacity(len);

            for index in 0..len {

                let (name, info) = read_attr_info(&mut reader, config_version)?;

                // When the attributes have been named, the device answers with them in the
                // order they were asked for, which the caller is allowed to rely on to tell
                // one answer from the next, so it is checked here.
                if let Some(expected) = expected.as_ref().and_then(|names| names.get([index]))
                && !name.eq_ignore_ascii_case(expected) {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description(format!("Attribute {expected:?} is answered with the configuration of another attribute"))
                        .with_append_description(format_args!("Got: {name:?}")));
                }

                entries.push(Named::new(Scalar::<str>::from(&*name), info));

            }

            Ok(entries)

        })

    }

    fn subscribe_attr_value(&self, admin: Arc<dyn BackendDev>, name: Scalar<str>, event: AttrValueEvent) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
        match self.consumer() {
            Ok(consumer) => zmq::subscribe_attr_value(consumer, self.name(), admin, name, event),
            Err(err) => Box::pin(ready(Err(err))),
        }
    }

    fn subscribe_attr_ready(&self, admin: Arc<dyn BackendDev>, name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrReady>>>> {
        match self.consumer() {
            Ok(consumer) => zmq::subscribe_attr_ready(consumer, self.name(), admin, name),
            Err(err) => Box::pin(ready(Err(err))),
        }
    }

    fn subscribe_attr_config(&self, admin: Arc<dyn BackendDev>, name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrInfo>>>> {
        match self.consumer() {
            Ok(consumer) => zmq::subscribe_attr_config(consumer, self.name(), admin, name),
            Err(err) => Box::pin(ready(Err(err))),
        }
    }

    fn command_history(&self, name: Scalar<str>, max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<Value>>>> {

        // IDL variants:
        // DevCmdHistoryList command_inout_history_2(in string command, in long n)
        // DevCmdHistory_4 command_inout_history_4(in string command, in long n)
        //
        // Note that v3 has no variant of its own and reuses the v2 one, and that v1 has no
        // history at all, the polling having come with v2.

        let operation: &[u8] = match self.version {
            DevVersion::V1 => return Box::pin(ready(Err(no_history_err(self.version)))),
            DevVersion::V2 |
            DevVersion::V3 => b"command_inout_history_2",
            DevVersion::V4 |
            DevVersion::V5 |
            DevVersion::V6 => b"command_inout_history_4",
        };

        let version = self.version;
        let request = self.request(operation, &HistoryInput { name: name.value(), max });

        Box::pin(async move {
            let reply = request.await?;
            read_command_history(&mut reply.read_simple()?, version)
        })

    }

    fn attr_history(&self, name: Scalar<str>, max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<AttrRead<Value>>>>> {

        // IDL variants:
        // DevAttrHistoryList read_attribute_history_2(in string name, in long n)
        // DevAttrHistoryList_3 read_attribute_history_3(in string name, in long n)
        // DevAttrHistory_4 read_attribute_history_4(in string name, in long n)
        // DevAttrHistory_5 read_attribute_history_5(in string name, in long n)

        let operation: &[u8] = match self.version {
            DevVersion::V1 => return Box::pin(ready(Err(no_history_err(self.version)))),
            DevVersion::V2 => b"read_attribute_history_2",
            DevVersion::V3 => b"read_attribute_history_3",
            DevVersion::V4 => b"read_attribute_history_4",
            DevVersion::V5 |
            DevVersion::V6 => b"read_attribute_history_5",
        };

        let version = self.version;
        let request = self.request(operation, &HistoryInput { name: name.value(), max });

        Box::pin(async move {
            let reply = request.await?;
            read_attr_history(&mut reply.read_simple()?, version)
        })

    }

}

/// The arguments every history operation takes: the name of the command or the attribute,
/// and how many records at most, which the IDL declares as a long.
struct HistoryInput<'a> {
    name: &'a str,
    max: usize,
}

impl CdrWrite for HistoryInput<'_> {
    fn write(&self, buf: &mut CdrWriter) -> Result<()> {
        buf.write_string(self.name)?;
        buf.write_long(i32::try_from(self.max).unwrap_or(i32::MAX))
    }
}

/// The error asking a device of that version for a history fails with, which only exists
/// from v2 on.
fn no_history_err(version: DevVersion) -> Error {
    Error::new(ErrorKind::Reason(reason::API_NotSupported))
        .with_description(format!("A device of IDL {version:?} has no history to give"))
}

/// A request to a device, resolving to its reply.
struct CorbaDevRequest {
    inner: ClientRequest,
}

impl Future for CorbaDevRequest {
    type Output = Result<CorbaDevReply>;
    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
        Pin::new(&mut self.get_mut().inner).poll(cx)
            .map(|inner| inner
                .map(|inner| CorbaDevReply { inner })
                .map_err(corba_err))
    }
}

/// The failure of a CORBA call to a device, which is taken as the device not being
/// reachable, or no longer: whatever it is, the connection to it is not one to go on with.
/// The CORBA failure becomes the parent of an [`ErrorKind::DevNotConnected`], which is what
/// tells a caller to find the device again, see [`crate::dev::DevProxy`]. What the device
/// itself answered with, a `DevFailed` above all, is left as it is.
fn corba_err(err: Error) -> Error {
    if err.kind() != ErrorKind::Corba {
        return err;
    }
    Error::new(ErrorKind::DevNotConnected).with_parent(err)
}

/// The reply of a device to a request.
#[derive(Debug)]
struct CorbaDevReply {
    inner: ClientReply,
}

impl CorbaDevReply {

    /// Read this reply, decoding the `DevFailed` every operation of the IDL may raise on
    /// its own into an [`Err`] of its own: nothing that reads a reply afterward has to
    /// deal with that one. Whichever other user exception turned up, if any, is left for
    /// the caller to make sense of, see [`Self::read_simple`] for operations that raise
    /// nothing else.
    fn read_raw(&self) -> Result<(CdrReader<'_>, Option<Cow<'_, str>>)> {
        let (mut reader, exception_id) = self.inner.read().map_err(corba_err)?;
        if let Some(tango_repo_id!("DevFailed")) = exception_id.as_deref() {
            Err(match reader.read::<Error>() {
                Ok(err) => err,
                Err(err) => err,
            })
        } else {
            Ok((reader, exception_id))
        }
    }

    /// Same as [`Self::read`], for an operation that raises nothing but `DevFailed`:
    /// whichever other user exception turned up is a failure of its own, nothing here
    /// ever expecting one.
    fn read_simple(&self) -> Result<CdrReader<'_>> {
        let (reader, exception_id) = self.read_raw()?;
        if let Some(exception_id) = exception_id.as_deref() {
            Err(unexpected_user_exception_err(exception_id))
        } else {
            Ok(reader)
        }
    }

    /// Same as [`Self::read_simple`], decoding the answer straight out of the reader.
    fn read<'a, O: CdrRead<'a>>(&'a self) -> Result<O> {
        self.read_simple()?.read()
    }

}

/// Read a list of attribute readings, one entry per attribute, in the structure of that
/// version.
fn read_attr_list(reader: &mut CdrReader<'_>, version: DevVersion) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
    let len = reader.read_sequence_len()?;
    let mut entries = Vec::with_capacity(len);
    for _ in 0..len {
        entries.push(read_attr(reader, version)?);
    }
    Ok(entries)
}

// =========================== //
//       Server backend        //
// =========================== //

/// The IDL version the objects served by this backend speak, which is what a client is
/// told they are and therefore which operations it sends them.
///
/// Six, the latest, whose only operation of its own, `info_6`, is answered: a client of an
/// older release still reaches the objects, which announce every earlier interface as well
/// and answer `_is_a` for each of them, see [`SERVED_REPO_IDS`].
const SERVED_VERSION: DevVersion = DevVersion::V6;

/// The Tango library release this backend tells its clients it is, in the encoding cppTango
/// uses, `major * 100 + minor * 10 + patch`: the first cppTango release whose devices are
/// of IDL v6, as the ones it serves are, see [`SERVED_VERSION`].
const LIB_VERSION: i32 = 1000;

/// The interfaces a served device announces, most derived first, which is every version of
/// the Tango IDL up to the one this backend speaks, each of them deriving from the previous
/// one, see [`ServerObject::repo_ids`].
const SERVED_REPO_IDS: [&str; SERVED_VERSION.number() as usize] = {
    let mut ids = [""; SERVED_VERSION.number() as usize];
    let mut index = 0;
    while index < ids.len() {
        ids[index] = DevVersion::REPO_ID[SERVED_VERSION as usize - index];
        index += 1;
    }
    ids
};

/// Internal structure implementing [`BackendServer`]
#[derive(Debug)]
struct CorbaServerBackend {
    /// The CORBA server itself, which is where the devices are served from and what the
    /// references handed out for them are built by. It serves from the moment it is bound
    /// and needs nothing run for it to, see [`Server::bind`].
    server: Server<CorbaServerObject>,
    /// What the events of the devices are published with, none when the server was not
    /// given the name of its admin device, which names them, see
    /// [`BackendServerConfig::admin_name`].
    supplier: Option<Supplier>,
}

impl CorbaServerBackend {

    /// The supplier of the events, failing if there is none, see the `supplier` field.
    fn supplier(&self) -> Result<&Supplier> {
        self.supplier.as_ref().ok_or_else(|| Error::not_implemented()
            .with_description("This server publishes no event over ZMQ"))
    }

    /// The device of that name, which must be lowercased, if it is served here, see
    /// [`CorbaServerObject::name`].
    fn served(&self, name: &str) -> Option<Arc<dyn BackendDev>> {
        self.server.get(name.as_bytes()).map(|object| Arc::clone(&object.dev))
    }

    /// Whether a device of that name, which must be lowercased, is served here, see
    /// [`Self::served`].
    fn is_served(&self, name: &str) -> bool {
        self.server.is_registered(name.as_bytes())
    }

}

impl BackendServer for CorbaServerBackend {

    fn add(&self, name: &str, dev: Arc<dyn BackendDev>) -> Result<Arc<str>> {

        // Tango addresses a device by its own name, lowercased, which is both what the
        // reference the server gives back carries and what makes a plain
        // 'corbaloc:iiop:host:port/dev/name' reach it. Registering it is all there is to
        // it, the server answering from the moment it is registered.
        let name = Arc::<str>::from(name.to_ascii_lowercase());
        let itr = self.server.register(Arc::new(CorbaServerObject {
            name: Arc::clone(&name),
            dev,
        })).map(Arc::<str>::from);

        if itr.is_ok() {
            debug!(dev = %name, "Serving the device over CORBA");
        }

        itr

    }

    fn remove(&self, name: &str) -> bool {
        let id = name.to_ascii_lowercase().into_bytes();
        self.server.deregister(&id)
    }

    fn host(&self) -> Arc<str> {
        self.server.host()
    }

    fn port(&self) -> u16 {
        self.server.local_addr().port()
    }

    fn version(&self) -> i32 {
        SERVED_VERSION.number() as i32
    }

    fn stop(&self) {
        if let Some(supplier) = &self.supplier {
            supplier.stop();
        }
        self.server.stop();
    }

    fn wait(&self) -> BoxFuture<'_, ()> {
        Box::pin(self.server.wait())
    }

    fn zmq_event_subscription_change(&self, args: Spectrum<str>) -> BoxFuture<'_, Result<(Spectrum<i32>, Spectrum<str>)>> {
        Box::pin(async move {
            let supplier = self.supplier()?;
            supplier.subscription_change(&args, |name| self.served(name), SERVED_VERSION.number()).await
        })
    }

    fn event_confirm_subscription(&self, args: Spectrum<str>) -> BoxFuture<'_, Result<()>> {
        Box::pin(async move {
            let supplier = self.supplier()?;
            supplier.confirm_subscription(&args, |name| self.is_served(name)).await
        })
    }

}

/// One served device, seen as the CORBA object a client sends its requests to: this is the
/// mirror of [`CorbaDevBackend`], turning the operations of the Tango IDL into calls of the
/// device backend instead of the other way around.
#[derive(Debug)]
struct CorbaServerObject {
    /// The name of the device, lowercased, which is also its object key, see
    /// [`ServerObject::id`].
    name: Arc<str>,
    /// The device itself, whatever it is: this backend only knows it through the trait.
    dev: Arc<dyn BackendDev>,
}

impl ServerObject for CorbaServerObject {

    type RequestFuture = BoxFuture<'static, Result<ServerReply>>;

    fn id(&self) -> &[u8] {
        self.name.as_bytes()
    }

    fn repo_ids(&self) -> &[&str] {
        &SERVED_REPO_IDS
    }

    fn request(self: Arc<Self>, request: ServerRequest) -> Self::RequestFuture {
        // Which operation this is for rides in the request and is read back out of the
        // octets it arrived as, so nothing of it is copied here to outlive the call.
        Box::pin(async move { self.handle_request(&request).await })
    }

}

/// What an operation came to, turned into what goes back to the client.
///
/// Whatever it failed with is answered as the `DevFailed` every operation of the IDL
/// raises, whatever its kind: a device fails with any kind it likes, and what it says is
/// what a Tango client reads. Only a reply that could not be written at all is given back
/// to the backend, which turns it into the CORBA system exception that says what it is.
fn answer(request: &ServerRequest, res: Result<ServerReply>) -> Result<ServerReply> {

    let err = match res {
        Ok(reply) => return Ok(reply),
        Err(err) => err,
    };

    let mut reply = request.reply_user_exception(tango_repo_id!("DevFailed"))?;
    reply.write(&err)?;

    Ok(reply)

}

impl CorbaServerObject {

    /// Answer one request of this object, which is what the backend drives, see
    /// [`ServerObject::request`]. A failure is answered as a `DevFailed`, see [`answer`].
    async fn handle_request(&self, request: &ServerRequest) -> Result<ServerReply> {

        let operation = request.operation();
        let res = self.dispatch(operation, request).await;

        if let Err(err) = &res {
            trace!(dev = %self.name, operation = %String::from_utf8_lossy(operation), %err, "Failed to run an operation");
        }

        answer(request, res)

    }

    /// Run one operation of the Tango IDL, reading its arguments out of the octets of the
    /// request and writing what it answers straight into the ones of the reply.
    ///
    /// Failing with [`ErrorKind::NotHandled`] is an operation this object does not have,
    /// which the client is told about like any other failure, see [`answer`].
    async fn dispatch(&self, operation: &[u8], request: &ServerRequest) -> Result<ServerReply> {

        let mut input = request.read();

        // The answer is written straight into the octets that go out, so the reply is
        // started before the operation runs and is what it writes into. One that raises
        // afterwards drops this and starts another, see [`Self::handle_request`].
        let mut reply = request.reply(ReplyStatus::NoException)?;
        let output = &mut reply;

        match operation {
            b"_get_name" => output.write_string(&self.dev.name().await?)?,
            b"_get_adm_name" => output.write_string(&self.dev.admin_dev_name().await?)?,
            b"_get_description" => output.write_string(&self.dev.description().await?)?,
            b"_get_state" => output.write(&self.dev.state().await?)?,
            b"_get_status" => output.write_string(&self.dev.status().await?.value())?,
            b"ping" => self.dev.ping().await?,

            // IDL variants:
            // DevInfo info()
            // DevInfo_3 info_3()
            // DevInfo_6 info_6()
            b"info" | b"info_3" | b"info_6" => {
                let version = match operation {
                    b"info" => DevVersion::V1,
                    b"info_3" => DevVersion::V3,
                    _ => DevVersion::V6,
                };
                write_dev_info(output, version, &self.dev.info().await?)?;
            }
            b"black_box" => {

                // 'n' is a long and not a sequence length, a negative one meaning that the
                // client wants the whole of it.
                let max = input.read_long()?;
                let entries = self.dev.black_box(usize::try_from(max).ok()).await?;
                output.write(&entries)?;

            }
            b"command_inout" | 
            b"command_inout_2" | 
            b"command_inout_4" => {

                // IDL variants:
                // any command_inout(in string command, in any argin)
                // any command_inout_2(in string command, in any argin, in DevSource source)
                // any command_inout_4(in string command, in any argin, in DevSource source, in ClntIdent cl_ident)
                //
                // The identity of the client is read by none of them: this implementation has
                // nowhere to report it and nothing follows it in the request.

                let name = input.read::<Scalar<str>>()?;
                let arg = read_command_data(&mut input)?;

                // v1 has no source of its own, and the device is what a client that cannot
                // ask for anything else gets.
                let source = if operation == b"command_inout" {
                    DevSource::Dev
                } else {
                    input.read()?
                };

                let answer = self.dev.command(name, arg, source).await?;
                write_command_data(output, &answer)?;

            }
            b"read_attributes" |
            b"read_attributes_2" |
            b"read_attributes_3" |
            b"read_attributes_4" |
            b"read_attributes_5" => {

                // IDL variants:
                // AttributeValueList read_attributes(in DevVarStringArray names)
                // AttributeValueList read_attributes_2(in DevVarStringArray names, in DevSource source)
                // AttributeValueList_3 read_attributes_3(in DevVarStringArray names, in DevSource source)
                // AttributeValueList_4 read_attributes_4(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)
                // AttributeValueList_5 read_attributes_5(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)

                // The version of the structure the values go out in, which is the one of
                // the operation and not the one of this object: a client that speaks an
                // older version is answered in the one it asked in.
                let version = match operation {
                    b"read_attributes" => DevVersion::V1,
                    b"read_attributes_2" => DevVersion::V2,
                    b"read_attributes_3" => DevVersion::V3,
                    b"read_attributes_4" => DevVersion::V4,
                    _ => DevVersion::V5,
                };

                let names = input.read::<Spectrum<str>>()?;
                let source = if version >= DevVersion::V2 {
                    input.read()?
                } else {
                    DevSource::Dev
                };

                let readings = self.dev.read_attr(all_attributes(names), source).await?;

                output.write_sequence_len(readings.len())?;
                for Named { name, value: read } in &readings {
                    write_attr(output, version, name.value(), read)?;
                }

            }
            b"write_attributes" | 
            b"write_attributes_3" | 
            b"write_attributes_4" => {

                // IDL variants:
                // void write_attributes(in AttributeValueList values)
                // void write_attributes_3(in AttributeValueList values)
                // void write_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
                //
                // Note that v2 has no variant of its own and reuses the v1 one, and that v3
                // only adds MultiDevFailed to the raised exceptions, the argument staying the
                // same.

                let version = if operation == b"write_attributes_4" {
                    DevVersion::V4
                } else {
                    DevVersion::V1
                };

                let values = read_attr_values(&mut input, version)?;
                let names = values.iter().map(|write| write.name.clone()).collect::<Vec<_>>();

                let results = self.dev.write_attr(values).await?;

                // One outcome per value and in the order they were given, so which
                // attribute a failure is of is where it lies and not anything the backend
                // says: only the index is kept, and the names given are what the name of it
                // is looked up in below, see [`BackendDev::write_attr`].
                let mut failures = results.into_iter()
                    .enumerate()
                    .filter_map(|(index, res)| res.err().map(|err| (index, err)))
                    .collect::<Vec<_>>();

                if !failures.is_empty() {

                    // v1 has no exception for some of the attributes having failed, so the
                    // first failure is what the whole call failed with, and the client is
                    // not told which one of them it was.
                    if version < DevVersion::V4 {
                        return Err(failures.swap_remove(0).1);
                    }

                    // Which starts the reply over as that exception, dropping whatever
                    // of a normal answer had been written into it.
                    *output = request.reply_user_exception(tango_repo_id!("MultiDevFailed"))?;
                    output.write_sequence_len(failures.len())?;
                    for (index, err) in failures {
                        output.write(&NamedError {
                            name: Cow::Borrowed(names.get(index).map(|name| name.value()).unwrap_or("")),
                            index,
                            err: Some(err),
                        })?;
                    }

                }

            }
            b"write_read_attributes_4" | 
            b"write_read_attributes_5" => {

                // IDL variants:
                // AttributeValueList_4 write_read_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
                // AttributeValueList_5 write_read_attributes_5(in AttributeValueList_4 values, in DevVarStringArray r_names, in ClntIdent cl_ident)
                //
                // Both take the very same v4 list of values, only what they answer with
                // differs. The list of the attributes v5 lets a client read back other than
                // the written ones is read and ignored: the backend reads back what it wrote.

                let reply_version = if operation == b"write_read_attributes_5" {
                    DevVersion::V5
                } else {
                    DevVersion::V4
                };

                let values = read_attr_values(&mut input, DevVersion::V4)?;

                // A failure of one of the attributes is carried by its own entry of the
                // answer, which has a place for one, so there is no MultiDevFailed to
                // raise here unlike above.
                let readings = self.dev.write_read_attr(values).await?;

                output.write_sequence_len(readings.len())?;
                for Named { name, value: read } in &readings {
                    write_attr(output, reply_version, name.value(), read)?;
                }

            }
            b"command_query" | 
            b"command_query_2" => {

                // IDL variants:
                // DevCmdInfo command_query(in string command)
                // DevCmdInfo_2 command_query_2(in string command)

                let version = if operation == b"command_query_2" {
                    DevVersion::V2
                } else {
                    DevVersion::V1
                };

                let name = input.read::<Scalar<str>>()?;
                let names = name.clone().reshape([1]).expect("one name");

                // One command was asked about, so the first entry is the answer and
                // anything behind it is a backend that answered more than it was asked.
                let infos = self.dev.command_info(Some(names)).await?;

                let Some(Named { name, value: info }) = infos.into_iter().next() else {
                    return Err(Error::new(ErrorKind::CommandNotFound)
                        .with_description(format!("Command {:?} is not defined on the device", name.value())));
                };

                write_command_info(output, version, name.value(), &info?)?;

            }
            b"command_list_query" | 
            b"command_list_query_2" => {

                // IDL variants:
                // DevCmdInfoList command_list_query()
                // DevCmdInfoList_2 command_list_query_2()

                let version = if operation == b"command_list_query_2" {
                    DevVersion::V2
                } else {
                    DevVersion::V1
                };

                let infos = self.dev.command_info(None).await?;

                // The commands that cannot be described are left out, so how many go out
                // is only known once they have all been written, see [`SequenceLen`].
                let out = &mut **output;
                let len = SequenceLen::start(out)?;
                let mut count = 0;

                for Named { name, value: info } in infos {

                    // One command that cannot be described is not worth failing the whole
                    // listing for, so it is simply left out of it.
                    let info = match info {
                        Ok(info) => info,
                        Err(err) => {
                            warn!(dev = %self.name, command = %name.value(), %err, "Failed to describe the command, leaving it out of the listing");
                            continue;
                        }
                    };

                    write_command_info(out, version, name.value(), &info)?;
                    count += 1;

                }

                len.finish(out, count)?;

            }
            b"set_attribute_config" |
            b"set_attribute_config_3" |
            b"set_attribute_config_4" |
            b"set_attribute_config_5" => {

                // IDL variants:
                // void set_attribute_config(in AttributeConfigList new_conf)
                // void set_attribute_config_3(in AttributeConfigList_3 new_conf)
                // void set_attribute_config_4(in AttributeConfigList_3 new_conf, in ClntIdent cl_ident)
                // void set_attribute_config_5(in AttributeConfigList_5 new_conf, in ClntIdent cl_ident)
                //
                // Note that the configuration of v2 is never sent, v2 reusing the v1 operation,
                // and that the client identification is left unread. What an attribute *is*,
                // which the structure carries along with its configuration, is left out, the
                // configuration being all a client may change of it.

                let version = match operation {
                    b"set_attribute_config" => DevVersion::V1,
                    b"set_attribute_config_5" => DevVersion::V5,
                    _ => DevVersion::V3,
                };

                let len = input.read_sequence_len()?;
                let mut configs = Vec::with_capacity(len);
                for _ in 0..len {
                    let (name, info) = read_attr_info(&mut input, version)?;
                    configs.push(Named::new(Scalar::from(&*name), info?.config));
                }

                // Every operation raises a single DevFailed, so the first attribute that
                // failed is what the whole call fails with, the others having been changed.
                let results = self.dev.set_attr_config(configs).await?;
                if let Some(err) = results.into_iter().find_map(Result::err) {
                    return Err(err);
                }

            }
            b"get_attribute_config" |
            b"get_attribute_config_2" |
            b"get_attribute_config_3" |
            b"get_attribute_config_5" => {

                // IDL variants:
                // AttributeConfigList get_attribute_config(in DevVarStringArray names)
                // AttributeConfigList_2 get_attribute_config_2(in DevVarStringArray names)
                // AttributeConfigList_3 get_attribute_config_3(in DevVarStringArray names)
                // AttributeConfigList_5 get_attribute_config_5(in DevVarStringArray names)
                //
                // Note that v4 has no variant of its own and reuses the v3 one.

                let version = match operation {
                    b"get_attribute_config" => DevVersion::V1,
                    b"get_attribute_config_2" => DevVersion::V2,
                    b"get_attribute_config_3" => DevVersion::V3,
                    _ => DevVersion::V5,
                };

                let names = input.read::<Spectrum<str>>()?;
                let requested = all_attributes(names);
                let named = requested.is_some();

                let infos = self.dev.attr_info(requested).await?;

                let out = &mut **output;
                let len = SequenceLen::start(out)?;
                let mut count = 0;

                for Named { name, value: info } in infos {

                    // An attribute the client named and that cannot be described is what
                    // the whole request fails with, the client having no place to be told
                    // about that one only. One that it did not name, the listing being of
                    // every attribute, is simply left out.
                    let info = match info {
                        Ok(info) => info,
                        Err(err) if named => return Err(err),
                        Err(err) => {
                            warn!(dev = %self.name, attr = %name.value(), %err, "Failed to describe the attribute, leaving it out of the listing");
                            continue;
                        }
                    };

                    write_attr_info(out, version, name.value(), &info)?;
                    count += 1;

                }

                len.finish(out, count)?;

            }
            b"command_inout_history_2" |
            b"command_inout_history_4" => {

                // IDL variants:
                // DevCmdHistoryList command_inout_history_2(in string command, in long n)
                // DevCmdHistory_4 command_inout_history_4(in string command, in long n)

                let version = match operation {
                    b"command_inout_history_2" => DevVersion::V2,
                    _ => DevVersion::V4,
                };

                let name = input.read::<Scalar<str>>()?;
                let max = usize::try_from(input.read_long()?).unwrap_or(0);

                // The structure announces the type of the results even when every record
                // failed, which is the one the command advertises.
                let infos = self.dev.command_info(Some(Spectrum::from(&[name.value()][..]))).await?;
                let info = single_info(infos)?;
                let history = self.dev.command_history(name, max).await?;
                write_command_history(output, version, info.value.out_kind, &history)?;

            }
            b"read_attribute_history_2" |
            b"read_attribute_history_3" |
            b"read_attribute_history_4" |
            b"read_attribute_history_5" => {

                // IDL variants:
                // DevAttrHistoryList read_attribute_history_2(in string name, in long n)
                // DevAttrHistoryList_3 read_attribute_history_3(in string name, in long n)
                // DevAttrHistory_4 read_attribute_history_4(in string name, in long n)
                // DevAttrHistory_5 read_attribute_history_5(in string name, in long n)

                let version = match operation {
                    b"read_attribute_history_2" => DevVersion::V2,
                    b"read_attribute_history_3" => DevVersion::V3,
                    b"read_attribute_history_4" => DevVersion::V4,
                    _ => DevVersion::V5,
                };

                let name = input.read::<Scalar<str>>()?;
                let max = usize::try_from(input.read_long()?).unwrap_or(0);

                // As for a command, the structure announces the type of the values, and
                // the name the attribute is defined with.
                let infos = self.dev.attr_info(Some(Spectrum::from(&[name.value()][..]))).await?;
                let Named { name: attr_name, value: info } = single_info(infos)?;
                let history = self.dev.attr_history(name, max).await?;
                write_attr_history(output, version, attr_name.value(), info.data_kind, info.data_format, &history)?;

            }
            // The operations the IDL declares and that this implementation has nothing to
            // answer with, either because the device backend has no such notion — the
            // pipes and the static information of a device — or because it is read-only,
            // the configuration of an attribute being given by its definition and not by a
            // client.
            //
            // These are told apart from an operation that does not exist at all, so that the
            // DevFailed a client is answered with says that this server does not implement
            // them, and not that they should never have been sent.
            b"get_pipe_config_5" |
            b"set_pipe_config_5" |
            b"read_pipe_5" |
            b"write_pipe_5" |
            b"write_read_pipe_5" => {
                return Err(Error::not_implemented()
                    .with_description(format!("Operation {:?} is not implemented in this server", String::from_utf8_lossy(operation))));
            }
            _ => {
                return Err(Error::new(ErrorKind::NotHandled));
            }
        }

        Ok(reply)

    }

}

/// The one entry of the infos a single name was asked for, failing as that name did.
fn single_info<T>(infos: Vec<NamedResult<T>>) -> Result<Named<T>> {
    let Named { name, value } = infos.into_iter().next()
        .ok_or_else(|| Error::new(ErrorKind::Internal)
            .with_description("The device gave no info for the name it was asked"))?;
    value.map(|value| Named::new(name, value))
}

/// The names of a listing operation as the backend takes them, which is [`None`] for every
/// object of the device: 'all attributes' is a name of its own in the IDL and not the name
/// of an attribute, and the marker to use for it changed in v3, so both are recognized.
fn all_attributes(names: Spectrum<str>) -> Option<Spectrum<str>> {
    if has_all_marker(&names) {
        None
    } else {
        Some(names)
    }
}

/// Whether the 'all attributes' marker is among these names, of either version, see
/// [`all_attributes`].
fn has_all_marker(names: &Spectrum<str>) -> bool {
    names.iter().any(|name| name == ALL_ATTRIBUTES || name == ALL_ATTRIBUTES_3)
}

/// Refuse names given to a request that hold the 'all attributes' marker, which would ask
/// a device for every attribute where specific ones were meant.
fn check_no_all_marker(names: &Spectrum<str>) -> Result<()> {
    if has_all_marker(names) {
        return Err(Error::new(ErrorKind::AttrNameInvalid)
            .with_description("Attribute names must not contain the 'all attributes' marker")
            .with_append_description(format_args!("Got: {names:?}")));
    }
    Ok(())
}

/// Read the list of values of an attribute writing operation, out of the octets it was
/// received in: only what the client wrote is kept of every entry, which is the only thing
/// those operations carry a value for.
fn read_attr_values(input: &mut CdrReader<'_>, version: DevVersion) -> Result<Vec<Named<Value>>> {

    let len = input.read_sequence_len()?;
    let mut values = Vec::with_capacity(len);

    for _ in 0..len {

        let Named { name, value: read } = read_attr(input, version)?;

        // A client that sends an attribute to write with no value in it is one whose
        // request makes no sense, there being nothing to write.
        let value = read?.w_value.ok_or_else(|| Error::new(ErrorKind::InvalidData)
            .with_description(format!("Attribute {:?} has no value to write", name.value())))?;

        values.push(Named::new(name, value));

    }

    Ok(values)

}

/// The length of a sequence, kept for once its entries are written and how many there are
/// is known, which is how a listing that leaves some entries out goes out without being
/// filtered first, see [`CdrWriter::set_position`].
struct SequenceLen(usize);

impl SequenceLen {

    /// Leave the room for the length behind, where the sequence starts.
    fn start(buf: &mut CdrWriter) -> Result<Self> {
        let pos = buf.position();
        buf.write_sequence_len(0)?;
        Ok(Self(pos))
    }

    /// Fill the length in, now that it is known, and carry on writing where the entries
    /// stopped.
    fn finish(self, buf: &mut CdrWriter, len: usize) -> Result<()> {
        let end = buf.position();
        buf.set_position(self.0);
        buf.write_sequence_len(len)?;
        buf.set_position(end);
        Ok(())
    }

}

/// Describe the Tango version of the device interface.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum DevVersion {
    V1,
    V2,
    V3,
    V4,
    V5,
    V6,
}

impl DevVersion {

    /// An array with all IDL versions in order.
    pub const ALL: [Self; 6] = [
        Self::V1,
        Self::V2,
        Self::V3,
        Self::V4,
        Self::V5,
        Self::V6,
    ];

    const REPO_ID: [&str; 6] = [
        tango_repo_id!("Device"),
        tango_repo_id!("Device_2"),
        tango_repo_id!("Device_3"),
        tango_repo_id!("Device_4"),
        tango_repo_id!("Device_5"),
        tango_repo_id!("Device_6"),
    ];

    #[inline]
    pub const fn repo_id(&self) -> &'static str {
        Self::REPO_ID[*self as usize]
    }

    /// The IDL release number this version is known as in the Tango protocol, which every
    /// place the version travels as a number uses, one-based unlike the discriminant.
    #[inline]
    pub const fn number(&self) -> u32 {
        *self as u32 + 1
    }

    /// The version of that IDL release number, see [`Self::number`], none for a release
    /// this crate does not know.
    #[inline]
    pub const fn from_number(number: u32) -> Option<Self> {
        match number {
            1..=6 => Some(Self::ALL[number as usize - 1]),
            _ => None,
        }
    }

}

fn unexpected_user_exception_err(exception_id: &str) -> Error {
    Error::new(ErrorKind::Internal)
        .with_description(format!("User exception {exception_id:?} is unexpected"))
}

// =========================== //
//          Utilities          //
// =========================== //

/// Try to do some binding action against a port range.
/// 
/// The implemented algorithm is to start with a random port, and then increment one by
/// one on retries.
/// 
/// The given function should return a control flow to either continue, or break with a
/// definitive result, which is either a critical error, or a success.
fn try_port_range<F, R>(ports: (u16, u16), mut func: F) -> ControlFlow<Result<R>>
where
    F: FnMut(u16) -> ControlFlow<Result<R>>,
{
    debug_assert!(ports.0 <= ports.1, "invalid port range");
    let mut port = rand::random_range(ports.0..=ports.1);
    let mut retry = ports.1 - ports.0;
    loop {
        func(port)?;
        if retry == 0 {
            break;
        }
        retry -= 1;
        if port == ports.1 {
            port = ports.0;
        } else {
            port += 1;
        }
    }
    // If we did not found a port in range, we return continue to signifies that the 
    // caller should either continue searching, or make its own error!
    ControlFlow::Continue(())
}

#[cfg(test)]
mod tests {

    use std::net::TcpListener;
    use std::time::Duration;

    use crate::future::block_on_timeout;

    use super::*;

    /// A device whose server is not there is not connected, the CORBA failure being the
    /// parent.
    #[test]
    fn connect_refused() {

        let backend = CorbaBackend::new();
        let port = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();

        let err = block_on_timeout(backend.connect_at("127.0.0.1", port, "test/dev/1"), Duration::from_secs(5))
            .unwrap()
            .unwrap_err();

        assert_eq!(err.kind(), ErrorKind::DevNotConnected);
        assert_eq!(err.parent().unwrap().kind(), ErrorKind::Corba);

    }

}