rstango 0.4.0

//! The consuming side of the event system: subscribing, and receiving the events of the
//! subscriptions, the equivalent of cppTango's `ZmqEventConsumer`.
//!
//! The ZMQ sockets are owned by a single thread, the [`Worker`]: a ZMQ socket may only be
//! used by one thread at a time, and the one polling them is busy polling them all the
//! time, so this thread is the only one touching them, and with them the state of the
//! subscriptions, which is therefore behind no lock. The rest of the crate talks to it
//! through a [`Consumer`], which queues a [`Command`] and wakes it up.
//!
//! Everything that happens on a schedule happens there too, the poll giving up at the next
//! deadline: the connection of a channel giving up, the reply to a subscription that was
//! given time to reach the publisher, the check of the heartbeats and the confirmation of
//! the subscriptions. A confirmation is a request to the admin device of the server, whose
//! future the worker polls itself, its waker being the same wake up the commands use, so
//! a request in flight never holds the events up.
//!
//! What a subscriber is handed is first the future of its subscription being made,
//! [`Subscribing`], then the stream of its events, [`Subscription`]. Neither decodes
//! anything until the events are read: the worker hands over the payload as it arrived,
//! which a subscriber decodes into whatever its kind of event carries, see [`Decode`].

use std::collections::{HashMap, HashSet, VecDeque};
use std::task::{Context, Poll, Wake, Waker};
use std::mem::{self, ManuallyDrop};
use std::time::{Duration, Instant};
use std::sync::{Arc, Mutex};
use std::future::Future;
use std::pin::Pin;
use std::thread;
use std::fmt;

use futures::stream::{FuturesUnordered, Stream, StreamExt};
use tracing::{debug, error, trace, warn};
use futures::future::BoxFuture;
use futures::channel::oneshot;

use crate::corba::tango::{read_attr, read_attr_info, read_attr_ready};
use crate::dev::{AttrInfo, AttrRead, AttrReady, DevSource};
use crate::error::{Error, ErrorKind, Result};
use crate::value::{Scalar, Spectrum, Value};
use crate::corba::orb::cdr::CdrReader;
use crate::corba::orb::endian::Endian;
use crate::backend::BackendDev;
use crate::corba::DevVersion;
use crate::future::Slot;
use crate::buf::Buf;

use super::{
    CallInfo, Endpoints, EventKind, SubscriptionInfo,
    CONFIRM_PERIOD, DEFAULT_LINGER, HEARTBEAT_PERIOD, HEARTBEAT_SUFFIX,
    SUBSCRIBE_PROPAGATION, SUB_SEND_HWM,
};


/// The endpoint the worker is woken up on, internal to the ZMQ context of a worker, so two
/// of them do not share it.
const CONTROL_ENDPOINT: &str = "inproc://rstango-zmq-control";
/// The endpoint the heartbeat socket reports the connections it made on, see
/// [`Worker::process_monitor`].
const MONITOR_ENDPOINT: &str = "inproc://rstango-zmq-monitor";
/// The time a new channel is given for one of the endpoints its server offers to answer,
/// after which the subscriptions waiting for it fail. Longer than the 100 ms cppTango
/// probes an endpoint for, because this is not a bare TCP connection but the ZMTP
/// handshake on top of it.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(1);

// =========================== //
//           Consumer            //
// =========================== //

/// What the rest of the crate holds of the worker, through which it subscribes and
/// unsubscribes. The worker stops once the last of them is dropped, which no subscription
/// outlives, every one of them holding a consumer of its own.
pub struct Consumer {
    /// Dropped by hand, before the worker is woken up, see the [`Drop`] implementation.
    command_tx: ManuallyDrop<crossbeam_channel::Sender<Command>>,
    notify: Arc<Notify>,
}

/// A subscription for the worker to make, once the server was asked to publish the event,
/// see [`Consumer::subscribe`].
pub struct SubscribeRequest {
    /// The admin device of the server, which the subscription is confirmed through.
    pub admin: Arc<dyn BackendDev>,
    /// The name of the device as the server knows it, which the confirmation names.
    pub dev_name: Arc<str>,
    /// The lowercased name of the attribute.
    pub obj_name: Arc<str>,
    pub kind: EventKind,
    /// What the server answered the negotiation with.
    pub info: SubscriptionInfo,
}

impl Consumer {

    /// Create the sockets and start the worker thread.
    pub fn start() -> Result<Self> {

        debug!("Starting the ZMQ event worker");

        let context = zmq::Context::new();

        // The commands travel over a channel and not over the socket, they are made of 
        // Rust values, so the socket only carries the wake up, and nothing travels back.
        let control = context.socket(zmq::PULL)?;
        control.set_linger(DEFAULT_LINGER)?;
        control.bind(CONTROL_ENDPOINT)?;

        let notify = context.socket(zmq::PUSH)?;
        notify.set_linger(DEFAULT_LINGER)?;
        notify.connect(CONTROL_ENDPOINT)?;

        let heartbeat = context.socket(zmq::SUB)?;
        heartbeat.set_linger(DEFAULT_LINGER)?;
        heartbeat.set_sndhwm(SUB_SEND_HWM)?;

        // The connections of the heartbeat socket are what tells which of the endpoints a
        // server offers can be reached, see [`Worker::process_monitor`].
        heartbeat.monitor(MONITOR_ENDPOINT, zmq::SocketEvent::HANDSHAKE_SUCCEEDED as i32)?;
        let monitor = context.socket(zmq::PAIR)?;
        monitor.set_linger(DEFAULT_LINGER)?;
        monitor.connect(MONITOR_ENDPOINT)?;

        let event = context.socket(zmq::SUB)?;
        event.set_linger(DEFAULT_LINGER)?;
        event.set_sndhwm(SUB_SEND_HWM)?;

        let (command_tx, command_rx) = crossbeam_channel::unbounded();
        let notify = Arc::new(Notify(Mutex::new(notify)));

        // Note that the context is not kept: every socket holds a reference to it, so it
        // is terminated once the last of them is closed.
        let worker = Worker {
            command_rx,
            control,
            monitor,
            heartbeat: Sub::new(heartbeat),
            event: Sub::new(event),
            waker: Waker::from(Arc::clone(&notify)),
            channels: HashMap::new(),
            events: HashMap::new(),
            declared: VecDeque::new(),
            confirmations: FuturesUnordered::new(),
        };

        thread::Builder::new()
            .name("zmq-consumer".to_string())
            .spawn(move || worker.main())
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to start the ZMQ event worker thread")
                .with_source(err))?;

        Ok(Self {
            command_tx: ManuallyDrop::new(command_tx),
            notify,
        })

    }

    fn send(&self, command: Command) -> Result<()> {
        self.command_tx.send(command)
            .map_err(|_| Error::new(ErrorKind::Internal)
                .with_description("The ZMQ event worker is unexpectedly no longer alive"))?;
        self.notify.notify();
        Ok(())
    }

    /// Have the worker subscribe, the future resolving to the stream of the events once
    /// the subscription is in place and was given time to reach the publisher (§5.1), see
    /// [`Subscribing`], each event being decoded with the given function.
    pub fn subscribe<T>(&self, request: SubscribeRequest, decode: Decode<T>) -> Result<Subscribing<T>> {
        let topic = Arc::clone(&request.info.event_topic);
        let version = request.info.version;
        let slot = Arc::new(Slot::new());
        let reply = self.send_subscribe(request, Arc::clone(&slot))?;
        Ok(Subscribing {
            inner: Some(SubscribingInner { reply, consumer: self.clone(), topic, slot, version, decode }),
        })
    }

    /// Hand a subscription over to the worker, its events going to that slot, and give back
    /// where its outcome is replied.
    ///
    /// Dropping the receiver is cancelling the subscription: the worker undoes what it
    /// subscribed for it once it finds nobody is waiting anymore.
    fn send_subscribe(&self, request: SubscribeRequest, slot: Arc<Slot<Delivery>>) -> Result<oneshot::Receiver<Result<()>>> {
        let (reply, reply_rx) = oneshot::channel();
        self.send(Command::Subscribe(Pending { request, slot, reply }))?;
        Ok(reply_rx)
    }

    /// Have the worker drop that subscriber of that topic.
    fn unsubscribe(&self, topic: Arc<str>, slot: Arc<Slot<Delivery>>) {
        if let Err(err) = self.send(Command::Unsubscribe { topic, slot }) {
            warn!(%err, "Failed to hand an unsubscription to the ZMQ event worker, ignoring");
        }
    }

}

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

impl Clone for Consumer {
    fn clone(&self) -> Self {
        Self {
            command_tx: ManuallyDrop::new((*self.command_tx).clone()),
            notify: Arc::clone(&self.notify),
        }
    }
}

impl Drop for Consumer {
    fn drop(&mut self) {
        // The sender goes first, so that the worker woken up below finds the channel
        // disconnected if this was the last consumer, instead of polling forever.
        // SAFETY: Never used again, this being the drop of its owner.
        unsafe { ManuallyDrop::drop(&mut self.command_tx) };
        self.notify.notify();
    }
}

/// The socket waking the worker up out of its poll, shared by every consumer and by the
/// waker of the futures the worker polls. We have to use a Mutex despite .send taking
/// &self, because [`zmq::Socket`] isn't Sync.
struct Notify(Mutex<zmq::Socket>);

impl Notify {

    fn notify(&self) {
        // An empty message is enough, the worker only takes it as the signal to look at
        // what is pending. A full queue is not a failure either: wake ups are already on
        // their way.
        match self.0.lock().unwrap().send(&[][..], zmq::DONTWAIT) {
            Ok(()) | Err(zmq::Error::EAGAIN) => {}
            Err(err) => warn!(%err, "Failed to wake the ZMQ event worker up, ignoring"),
        }
    }

}

impl Wake for Notify {

    fn wake(self: Arc<Self>) {
        self.notify();
    }

    fn wake_by_ref(self: &Arc<Self>) {
        self.notify();
    }

}

enum Command {
    Subscribe(Pending),
    Unsubscribe { 
        topic: Arc<str>, 
        slot: Arc<Slot<Delivery>>,
    },
}

/// A subscription the worker was handed and has not replied to yet.
struct Pending {
    request: SubscribeRequest,
    /// Where the events of this subscription are handed over.
    slot: Arc<Slot<Delivery>>,
    reply: oneshot::Sender<Result<()>>,
}

/// How the payload of an event is decoded, given the release it was published in (§9).
pub type Decode<T> = fn(&mut CdrReader<'_>, DevVersion) -> Result<T>;

// =========================== //
//         Subscribing         //
// =========================== //

/// A subscription being made by the worker, which resolves to its stream once it is in
/// place, see [`Consumer::subscribe`].
///
/// Dropping it cancels the subscription, including when the worker already succeeded but
/// this future was not polled since: see the [`Drop`] implementation.
pub struct Subscribing<T> {
    /// Taken once the future resolved, which is what tells the drop there is nothing to
    /// cancel.
    inner: Option<SubscribingInner<T>>,
}

/// What the stream is made of once the subscription is in place.
struct SubscribingInner<T> {
    reply: oneshot::Receiver<Result<()>>,
    consumer: Consumer,
    topic: Arc<str>,
    slot: Arc<Slot<Delivery>>,
    version: DevVersion,
    decode: Decode<T>,
}

impl<T> Future for Subscribing<T> {

    type Output = Result<Subscription<T>>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {

        let this = self.get_mut();
        let inner = this.inner.as_mut().expect("polled after it resolved");

        let res = match Pin::new(&mut inner.reply).poll(cx) {
            Poll::Pending => return Poll::Pending,
            Poll::Ready(res) => res,
        };

        let SubscribingInner { 
            consumer, 
            topic, 
            slot, 
            version, 
            decode, 
            ..
        } = this.inner.take().unwrap();

        Poll::Ready(match res {
            // From here on, it is the stream that unsubscribes when it is dropped.
            Ok(Ok(())) => Ok(Subscription { 
                consumer, 
                topic, 
                slot, 
                version, 
                decode,
                ended: false,
            }),
            // Nothing was left subscribed for a subscription that failed.
            Ok(Err(err)) => Err(err),
            Err(oneshot::Canceled) => Err(Error::new(ErrorKind::Internal)
                .with_description("The ZMQ event worker dropped the subscription")),
        })

    }

}

impl<T> Drop for Subscribing<T> {
    fn drop(&mut self) {

        let Some(mut inner) = self.inner.take() else { return };

        // Closed first, so that the worker can no longer succeed behind our back: a reply
        // it sends from here on fails, and the worker undoes the subscription itself. A
        // success it sent before is the one left to undo here.
        inner.reply.close();
        if let Ok(Some(Ok(()))) = inner.reply.try_recv() {
            inner.consumer.unsubscribe(inner.topic, inner.slot);
        }

    }
}

// =========================== //
//        Subscription         //
// =========================== //

/// The stream of the events of a subscription, which unsubscribes when it is dropped.
///
/// It gives the latest event and not every one of them (§10). It ends once the
/// subscription is lost, a server whose heartbeat stopped giving an
/// [`ErrorKind::DevNotConnected`] before it does: the server may have been restarted,
/// elsewhere even, and forgot every subscription, so it is the subscriber's to make it
/// again, through a negotiation of its own.
pub struct Subscription<T> {
    consumer: Consumer,
    topic: Arc<str>,
    slot: Arc<Slot<Delivery>>,
    version: DevVersion,
    decode: Decode<T>,
    /// Set once the loss of the subscription was given, the stream having ended.
    ended: bool,
}

impl<T> Subscription<T> {

    fn decode(&self, delivery: Delivery) -> Result<T> {
        match delivery {
            Delivery::Lost => {
                Err(Error::new(ErrorKind::DevNotConnected)
                    .with_description(format!("Topic {:?} received no heartbeat from the device server for more than {HEARTBEAT_PERIOD:?}, losing the channel", self.topic)))
            }
            Delivery::Event { 
                payload, 
                endian, 
                failed,
            } => {

                let mut reader = CdrReader::new(&payload)
                    .with_endian(endian);

                // The reading the server was about to publish failed, and the error stack
                // of that failure travels in place of the value (§8).
                if failed {
                    return Err(match reader.read::<Error>() {
                        Ok(err) => err,
                        Err(err) => err,
                    });
                }

                (self.decode)(&mut reader, self.version)

            }
        }
    }

}

impl<T> Stream for Subscription<T> {

    type Item = Result<T>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        if this.ended {
            return Poll::Ready(None);
        }
        this.slot.poll_take(cx).map(|delivery| {
            let delivery = delivery.unwrap_or(Delivery::Lost);
            this.ended = matches!(delivery, Delivery::Lost);
            Some(this.decode(delivery))
        })
    }

}

impl<T> Drop for Subscription<T> {
    fn drop(&mut self) {
        // A subscription that was lost was already dropped by the worker.
        if !self.slot.is_ended() {
            self.consumer.unsubscribe(Arc::clone(&self.topic), Arc::clone(&self.slot));
        }
    }
}

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

// =========================== //
//          Decoding           //
// =========================== //

/// An attribute value event, of the very structure a reading answers with: cppTango's
/// `ZmqAttributeValue_4` and `ZmqAttributeValue_5` have the fields of the plain ones, in the
/// same order (§8). The name it carries is dropped, the subscriber knowing it.
pub fn decode_value(reader: &mut CdrReader<'_>, version: DevVersion) -> Result<AttrRead<Value>> {
    read_attr(reader, version)?.value
}

/// An attribute configuration event, an `AttributeConfig_3` before IDL v5 and an
/// `AttributeConfig_5` from then on (§8).
pub fn decode_config(reader: &mut CdrReader<'_>, version: DevVersion) -> Result<AttrInfo> {
    let structure = if version >= DevVersion::V5 { DevVersion::V5 } else { DevVersion::V3 };
    read_attr_info(reader, structure)?.1
}

/// An attribute data ready event, an `AttDataReady`, which has a single version (§8).
pub fn decode_ready(reader: &mut CdrReader<'_>, _version: DevVersion) -> Result<AttrReady> {
    Ok(read_attr_ready(reader)?.value)
}

// =========================== //
//          Delivery           //
// =========================== //

/// What a subscriber is handed through its [`Slot`], of which an event has one per
/// subscriber: only the latest delivery is kept, this protocol delivering state and not a
/// history (§10), and the loss of the subscription is the end of the slot.
#[derive(Clone)]
enum Delivery {
    /// An event, left undecoded: the worker decodes nothing on behalf of a subscriber that
    /// may never read it.
    Event {
        /// The payload, its four leading filler octets dropped (§8), shared by every
        /// subscriber of the topic.
        payload: Buf<u8, 1>,
        /// The byte order the payload is in.
        endian: Endian,
        /// Whether the payload is the error stack of a reading that failed.
        failed: bool,
    },
    /// The subscription was lost, the server having stopped sending its heartbeat (§11),
    /// which is the last delivery: what the end of the slot is given as.
    Lost,
}

// =========================== //
//           Worker            //
// =========================== //

struct Worker {
    command_rx: crossbeam_channel::Receiver<Command>,
    /// The socket the wake ups are received on.
    control: zmq::Socket,
    /// The socket the heartbeat socket reports its connections on.
    monitor: zmq::Socket,
    /// Subscribed to the heartbeat of every channel (§5).
    heartbeat: Sub,
    /// Subscribed to every event (§5).
    event: Sub,
    /// The waker of the confirmations, which wakes this thread up.
    waker: Waker,
    /// The channels, one per device server, keyed by their name.
    channels: HashMap<Arc<str>, Channel>,
    /// The events, keyed by their topic.
    events: HashMap<Arc<str>, Event>,
    /// The subscriptions in place whose reply waits for them to reach the publisher, in 
    /// the order they come due: they all wait the same [`SUBSCRIBE_PROPAGATION`] from 
    /// when they are pushed.
    declared: VecDeque<Declared>,
    /// The confirmations in flight, each giving back the channel it confirms.
    confirmations: FuturesUnordered<BoxFuture<'static, (Arc<str>, Result<Value>)>>,
}

/// One of the subscriber sockets, with the endpoints it is connected to.
struct Sub {
    socket: zmq::Socket,
    /// How many users each endpoint has: connecting twice to the same endpoint would
    /// deliver every message twice (§5), so it is only connected for its first user and
    /// disconnected after its last.
    connected: HashMap<Arc<str>, usize>,
    /// The receive high water mark set, so that it is only set when it changes.
    rcvhwm: Option<i32>,
}

/// The event channel of one device server, carrying its heartbeat and all its events.
struct Channel {
    /// The admin device of the server, the subscriptions are confirmed through.
    admin: Arc<dyn BackendDev>,
    heartbeat_topic: Arc<str>,
    state: ChannelState,
    /// The topics of the events subscribed on this channel, which tells when it is no
    /// longer needed.
    events: HashSet<Arc<str>>,
    last_heartbeat: Instant,
    next_confirm: Instant,
}

enum ChannelState {
    /// The heartbeat socket is connected to every candidate pair, the first to complete its
    /// handshake being the one kept, see [`Worker::process_monitor`].
    Connecting {
        candidates: Vec<Endpoints>,
        deadline: Instant,
        /// The subscriptions waiting for the channel to be connected.
        waiting: Vec<Pending>,
    },
    /// The pair of endpoints is chosen, and reused by every later subscription (§5).
    Connected {
        endpoints: Endpoints,
    },
}

/// One subscribed event, shared by every subscriber of its topic.
struct Event {
    channel_name: Arc<str>,
    /// The endpoint the event socket is connected to for it.
    endpoint: Arc<str>,
    dev_name: Arc<str>,
    obj_name: Arc<str>,
    /// The name the confirmation sends, with the IDL compatibility prefix if it has one.
    event_name: Arc<str>,
    version: DevVersion,
    kind: EventKind,
    /// The counter of the last event (§10).
    counter: u32,
    /// Set when the last event was a dropped duplicate (§10).
    discarded: bool,
    slots: Vec<Arc<Slot<Delivery>>>,
}

/// A subscription in place whose reply waits until it can be expected to have reached the
/// publisher (§5.1).
struct Declared {
    at: Instant,
    reply: oneshot::Sender<Result<()>>,
    topic: Arc<str>,
    slot: Arc<Slot<Delivery>>,
}

impl Worker {

    fn main(mut self) {

        debug!("Started the ZMQ event worker");

        loop {

            if !self.process_commands() {
                debug!("Stopping the ZMQ event worker, nothing can reach it anymore");
                return;
            }

            self.process_timers();
            self.poll_confirmations();

            let timeout = match self.next_deadline() {
                // Rounded up, so that the deadline has passed when the poll comes back.
                Some(deadline) => deadline.saturating_duration_since(Instant::now())
                    .as_micros().div_ceil(1000) as i64,
                None => -1,
            };

            let mut items = [
                self.control.as_poll_item(zmq::POLLIN),
                self.monitor.as_poll_item(zmq::POLLIN),
                self.heartbeat.socket.as_poll_item(zmq::POLLIN),
                self.event.socket.as_poll_item(zmq::POLLIN),
            ];

            match zmq::poll(&mut items, timeout) {
                Ok(_) => {}
                Err(zmq::Error::EINTR) => continue,
                Err(err) => {
                    error!(%err, "Failed to poll the ZMQ event sockets, stopping the worker");
                    return;
                }
            }

            let readable = items.map(|item| item.is_readable());

            // The wake ups only say that something is pending, which is looked at on every
            // turn anyway, so they are only drained.
            if readable[0] {
                while self.control.recv_bytes(zmq::DONTWAIT).is_ok() {}
            }

            // The connections come first, so that the channel an event or a heartbeat is
            // for is known when it arrives.
            if readable[1] {
                self.process_monitor();
            }

            if readable[2] {
                self.process_heartbeat();
            }

            if readable[3] {
                self.process_event();
            }

        }

    }

    /// Run the pending commands, giving false when no consumer is left, and therefore no
    /// subscription.
    fn process_commands(&mut self) -> bool {
        loop {
            match self.command_rx.try_recv() {
                Ok(Command::Subscribe(pending)) => self.subscribe(pending),
                Ok(Command::Unsubscribe { topic, slot }) => self.unsubscribe(&topic, &slot),
                Err(crossbeam_channel::TryRecvError::Empty) => return true,
                Err(crossbeam_channel::TryRecvError::Disconnected) => return false,
            }
        }
    }

    // ======================= //
    //       Subscribing       //
    // ======================= //

    /// Subscribe, attaching to the event if it is already subscribed, and connecting its
    /// channel first if it is not.
    fn subscribe(&mut self, pending: Pending) {

        let request = &pending.request;
        let topic = &request.info.event_topic;

        if pending.reply.is_canceled() {
            debug!(%topic, "Dropping a subscription nobody waits for anymore");
            return;
        }

        if let Some(event) = self.events.get_mut(topic) {

            let coherent =
                event.kind == request.kind &&
                event.version == request.info.version &&
                *event.dev_name == *request.dev_name &&
                *event.obj_name == *request.obj_name &&
                *event.channel_name == *request.info.channel_name;

            if !coherent {
                error!(%topic, "Failed to subscribe to the event, its topic is already subscribed for another one");
                let err = Error::new(ErrorKind::Internal)
                    .with_description(format!("Topic {topic:?} is already subscribed to for another device, attribute, event kind or IDL release"));
                let _ = pending.reply.send(Err(err));
                return;
            }

            // The topic reached the publisher long ago, there is nothing to wait for (§5.1).
            event.slots.push(Arc::clone(&pending.slot));
            debug!(%topic, "Attached to the event already subscribed to");

            let topic = Arc::clone(topic);
            if pending.reply.send(Ok(())).is_err() {
                self.unsubscribe(&topic, &pending.slot);
            }

            return;

        }

        match self.channels.get_mut(&request.info.channel_name) {
            Some(Channel { state: ChannelState::Connecting { waiting, .. }, .. }) => {
                waiting.push(pending);
            }
            Some(Channel { state: ChannelState::Connected { endpoints }, .. }) => {
                let endpoint = Arc::clone(&endpoints.event);
                self.declare(pending, endpoint);
            }
            None => self.open_channel(pending),
        }

    }

    /// Connect the channel the request is for, which it then waits for.
    fn open_channel(&mut self, pending: Pending) {

        let request = &pending.request;
        let channel_name = Arc::clone(&request.info.channel_name);
        let heartbeat_topic = request.info.heartbeat_topic();

        let candidates = match candidates(&request.info) {
            Ok(candidates) => candidates,
            Err(err) => {
                let _ = pending.reply.send(Err(err));
                return;
            }
        };

        // An endpoint the heartbeat socket is already connected to has completed its
        // handshake long ago, so there is nothing to wait for if one of them is.
        let known = candidates.iter()
            .find(|endpoints| self.heartbeat.is_connected(&endpoints.heartbeat))
            .cloned();

        let to_connect = match &known {
            Some(endpoints) => std::slice::from_ref(endpoints),
            None => &candidates[..],
        };

        if let Err(err) = self.heartbeat.subscribe(&heartbeat_topic) {
            let _ = pending.reply.send(Err(err));
            return;
        }

        // Connect the heartbeat socket to every candidate at once: ZMQ connects in the
        // background, and the monitor reports each handshake as it completes, the first one
        // choosing the pair, see [`Self::process_monitor`]. A connection that fails here is a
        // malformed endpoint, not an unreachable one, so the ones made so far are undone.
        for (index, endpoints) in to_connect.iter().enumerate() {
            if let Err(err) = self.heartbeat.connect(&endpoints.heartbeat) {
                for endpoints in &to_connect[..index] {
                    self.heartbeat.disconnect(&endpoints.heartbeat);
                }
                self.heartbeat.unsubscribe(&heartbeat_topic);
                let _ = pending.reply.send(Err(err));
                return;
            }
        }

        let now = Instant::now();

        // The negotiation that came with the request confirmed the subscription.
        let mut channel = Channel {
            admin: Arc::clone(&request.admin),
            heartbeat_topic,
            state: ChannelState::Connected { endpoints: candidates[0].clone() },
            events: HashSet::new(),
            last_heartbeat: now,
            next_confirm: now + CONFIRM_PERIOD,
        };

        match known {
            Some(endpoints) => {
                debug!(channel = %channel_name, ?endpoints, "Opened the channel on endpoints already connected");
                channel.state = ChannelState::Connected { endpoints };
                self.channels.insert(channel_name, channel);
                self.subscribe(pending);
            }
            None => {
                debug!(channel = %channel_name, ?candidates, "Connecting the channel");
                channel.state = ChannelState::Connecting {
                    candidates,
                    deadline: now + CONNECT_TIMEOUT,
                    waiting: vec![pending],
                };
                self.channels.insert(channel_name, channel);
            }
        }

    }

    /// Subscribe the event on the connected channel it belongs to, the reply going out once
    /// the subscription had time to reach the publisher.
    fn declare(&mut self, pending: Pending, endpoint: Arc<str>) {

        let Pending { request: SubscribeRequest { dev_name, obj_name, kind, info, .. }, slot, reply } = pending;
        let topic = info.event_topic;

        // Only taken into account for the connections made afterward (§5).
        if !self.event.is_connected(&endpoint) 
        && let Some(hwm) = info.sub_hwm
        && let Err(err) = self.event.set_rcvhwm(hwm) {
            let _ = reply.send(Err(err));
            return;
        }

        if let Err(err) = self.event.connect(&endpoint) {
            let _ = reply.send(Err(err));
            return;
        }

        if let Err(err) = self.event.subscribe(&topic) {
            self.event.disconnect(&endpoint);
            let _ = reply.send(Err(err));
            return;
        }

        self.channels.get_mut(&info.channel_name)
            .expect("the channel of a declared event is connected")
            .events.insert(Arc::clone(&topic));

        self.events.insert(Arc::clone(&topic), Event {
            channel_name: info.channel_name,
            endpoint,
            dev_name,
            obj_name,
            event_name: kind.topic_name(info.version),
            version: info.version,
            kind,
            counter: 0,
            discarded: false,
            slots: vec![Arc::clone(&slot)],
        });

        debug!(%topic, "Subscribed to the event");

        self.declared.push_back(Declared {
            at: Instant::now() + SUBSCRIBE_PROPAGATION,
            reply,
            topic,
            slot,
        });

    }

    /// Drop a subscriber of an event, unsubscribing the event, and closing its channel, once
    /// they have no user left.
    fn unsubscribe(&mut self, topic: &Arc<str>, slot: &Arc<Slot<Delivery>>) {

        let Some(event) = self.events.get_mut(topic) else {
            warn!(%topic, "Unsubscribing from an event that is not subscribed, ignoring");
            return;
        };

        event.slots.retain(|other| !Arc::ptr_eq(other, slot));
        if !event.slots.is_empty() {
            debug!(%topic, remaining = event.slots.len(), "Detached from the event, which still has subscribers");
            return;
        }

        debug!(%topic, "Unsubscribing from the event, which has no subscriber left");

        let event = self.events.remove(topic).unwrap();
        self.event.unsubscribe(topic);
        self.event.disconnect(&event.endpoint);

        let Some(channel) = self.channels.get_mut(&event.channel_name) else {
            error!(channel = %event.channel_name, "Failed to find the channel of the event");
            return;
        };

        channel.events.remove(topic);
        if channel.events.is_empty() {
            self.close_channel(&event.channel_name);
        }

    }

    /// Close a channel, failing whatever was waiting for it to connect.
    fn close_channel(&mut self, channel_name: &Arc<str>) {

        let Some(channel) = self.channels.remove(channel_name) else { return };
        debug!(channel = %channel_name, "Closing the channel");

        self.heartbeat.unsubscribe(&channel.heartbeat_topic);

        match channel.state {
            ChannelState::Connected { endpoints } => {
                self.heartbeat.disconnect(&endpoints.heartbeat);
            }
            ChannelState::Connecting { candidates, waiting, .. } => {
                
                for endpoints in &candidates {
                    self.heartbeat.disconnect(&endpoints.heartbeat);
                }

                let err = Error::new(ErrorKind::EventNotConnected)
                    .with_description(format!("No endpoint the device server offers answers within {CONNECT_TIMEOUT:?}"))
                    .with_append_description(format_args!("Tried: {candidates:?}"));
                for pending in waiting {
                    let _ = pending.reply.send(Err(err.clone()));
                }

            }
        }

    }

    /// Drop a channel whose server is gone, along with every event subscribed on it, each
    /// subscriber being told that its subscription was lost.
    ///
    /// Nothing of it is kept for when the server comes back: it may come back elsewhere,
    /// and it forgot every subscription anyway, so a subscriber makes its own again, which
    /// opens the channel anew, see [`crate::dev::DevChannel`].
    fn lose_channel(&mut self, channel_name: &Arc<str>) {

        let Some(channel) = self.channels.get_mut(channel_name) else { return };
        let topics = mem::take(&mut channel.events);

        for topic in &topics {

            let Some(event) = self.events.remove(topic) else { continue };
            self.event.unsubscribe(topic);
            self.event.disconnect(&event.endpoint);

            for slot in &event.slots {
                slot.end();
            }

        }

        debug!(channel = %channel_name, events = topics.len(), "Lost the channel");
        self.close_channel(channel_name);

    }

    // ======================= //
    //         Timers          //
    // ======================= //

    /// Run everything that is due at that time, see [`Self::next_deadline`] for when that
    /// is.
    fn process_timers(&mut self) {

        let now = Instant::now();

        // Reply to the subscriptions that had time to reach the publisher (§5.1), a
        // receiver that is gone cancelling its subscription.
        while let Some(declared) = self.declared.pop_front_if(|declared| declared.at <= now) {
            if declared.reply.send(Ok(())).is_err() {
                debug!(topic = %declared.topic, "Dropping a subscription nobody waits for anymore");
                self.unsubscribe(&declared.topic, &declared.slot);
            }
        }

        // Go through the channels, only deciding what is to be done to them: closing one
        // and confirming one both need the whole worker, which this loop borrows. No
        // vector allocates on a turn where nothing is due, which is most of them.
        let mut abandoned = Vec::new();
        let mut lost = Vec::new();
        let mut confirm = Vec::new();
        for (channel_name, channel) in &mut self.channels {
            match channel.state {
                // A channel none of whose endpoints completed its handshake in time.
                ChannelState::Connecting { deadline, .. } => {
                    if deadline <= now {
                        warn!(channel = %channel_name, "Failed to connect the channel, none of its endpoints answered in time");
                        abandoned.push(Arc::clone(channel_name));
                    }
                }
                ChannelState::Connected { .. } => {

                    // A server whose heartbeat stopped (§11), which may have been
                    // restarted, elsewhere even, and forgot every subscription.
                    if now.duration_since(channel.last_heartbeat) >= HEARTBEAT_PERIOD {
                        warn!(channel = %channel_name, "Received no heartbeat from the device server for more than {HEARTBEAT_PERIOD:?}, losing the channel");
                        lost.push(Arc::clone(channel_name));
                        continue;
                    }

                    // A channel whose subscriptions are due to be confirmed (§11).
                    if channel.next_confirm <= now {
                        // Counted from now and not from the answer, so that a server
                        // failing to answer is not asked again on every turn.
                        channel.next_confirm = now + CONFIRM_PERIOD;
                        confirm.push(Arc::clone(channel_name));
                    }

                }
            }
        }

        // Close the channels given up on, failing the subscriptions waiting for them.
        for channel_name in abandoned {
            self.close_channel(&channel_name);
        }

        // Drop the channels whose server is gone, and every subscription with them.
        for channel_name in lost {
            self.lose_channel(&channel_name);
        }

        // Send the confirmations, which are polled with the others from then on, see
        // `Self::poll_confirmations`.
        for channel_name in confirm {
            self.confirm(channel_name);
        }

    }

    /// The next time [`Self::process_timers`] has something to do.
    fn next_deadline(&self) -> Option<Instant> {

        let declared = self.declared
            .front()
            .map(|declared| declared.at)
            .into_iter();

        let channels = self.channels
            .values()
            .flat_map(|channel| match channel.state {
                ChannelState::Connecting { deadline, .. } => [Some(deadline), None],
                ChannelState::Connected { .. } => [
                    Some(channel.last_heartbeat + HEARTBEAT_PERIOD),
                    Some(channel.next_confirm),
                ],
            })
            .flatten();

        declared.chain(channels).min()

    }

    /// Confirm every event of a channel, in one request (§11).
    fn confirm(&mut self, channel_name: Arc<str>) {

        let Some(channel) = self.channels.get(&channel_name) else { return };

        let params = channel.events.iter()
            .filter_map(|topic| self.events.get(topic))
            .flat_map(|event| [&*event.dev_name, &*event.obj_name, &*event.event_name]);

        let params = Spectrum::<str>::from_strs(params).expect("event names of more than 4 GiB");

        if params.is_empty() {
            return;
        }

        debug!(channel = %channel_name, ?params, "Confirming the subscriptions of the channel");

        let admin = Arc::clone(&channel.admin);
        let input = Value::from(params);

        self.confirmations.push(Box::pin(async move {
            let res = admin.command(Scalar::<str>::from("EventConfirmSubscription"), input, DevSource::Dev).await;
            (channel_name, res)
        }));

    }

    fn poll_confirmations(&mut self) {
        let mut cx = Context::from_waker(&self.waker);
        while let Poll::Ready(Some((channel_name, res))) = self.confirmations.poll_next_unpin(&mut cx) {
            match res {
                Ok(_) => trace!(channel = %channel_name, "Confirmed the subscriptions of the channel"),
                // The next one comes well before the server drops the subscriptions, and a
                // server of Tango 8 has no such command and drops nothing either (§11).
                Err(err) => warn!(channel = %channel_name, %err, "Failed to confirm the subscriptions of the channel"),
            }
        }
    }

    // ======================= //
    //        Receiving        //
    // ======================= //

    /// Process one connection the heartbeat socket reports, the first of the candidates of
    /// a connecting channel to complete its handshake being the pair that channel keeps.
    ///
    /// This is the reachability check of §5 made on the connection itself: the one that
    /// proved the endpoint reachable is the one used, and nothing blocks.
    fn process_monitor(&mut self) {

        // Two parts, the event and its value, then the endpoint it is about.
        let parts = match recv_message(&self.monitor, 2) {
            Ok(Some(parts)) => parts,
            Ok(None) => return,
            Err(err) => {
                warn!(%err, "Failed to receive a ZMQ monitor message, ignoring");
                return;
            }
        };

        let event = match &*parts[0] {
            [a, b, ..] => u16::from_ne_bytes([*a, *b]),
            _ => {
                warn!("Received a ZMQ monitor message that is too short, ignoring");
                return;
            }
        };

        if event != zmq::SocketEvent::HANDSHAKE_SUCCEEDED as u16 {
            return;
        }

        let Ok(endpoint) = str::from_utf8(&parts[1]) else {
            warn!("Received a ZMQ monitor message whose endpoint is not valid text, ignoring");
            return;
        };

        let found = self.channels
            .iter()
            .find_map(|(channel_name, channel)| match &channel.state {
                ChannelState::Connecting { candidates, .. } => candidates
                    .iter()
                    .find(|endpoints| *endpoints.heartbeat == *endpoint)
                    .map(|endpoints| (Arc::clone(channel_name), endpoints.clone())),
                ChannelState::Connected { .. } => None,
            });

        // A reconnection of a channel already connected, which needs nothing.
        let Some((channel_name, chosen)) = found else {
            trace!(%endpoint, "Ignoring the handshake of an endpoint no channel is waiting for");
            return;
        };

        debug!(channel = %channel_name, endpoints = ?chosen, "Connected the channel");

        let channel = self.channels.get_mut(&channel_name).unwrap();
        channel.last_heartbeat = Instant::now();

        // Replace the old state, which MUST BE a Connecting state.
        let state = mem::replace(&mut channel.state, ChannelState::Connected { 
            endpoints: chosen.clone(),
        });

        let ChannelState::Connecting { 
            candidates, 
            waiting, 
            .. 
        } = state else { unreachable!() };

        // The other candidates may complete their handshake too in the meantime, which only
        // ever duplicates heartbeats, never events, the event socket only being connected
        // to the chosen pair.
        for endpoints in candidates {
            if *endpoints.heartbeat != *chosen.heartbeat {
                self.heartbeat.disconnect(&endpoints.heartbeat);
            }
        }

        for pending in waiting {
            self.subscribe(pending);
        }

        // Every waiting subscription may have been given up on.
        if self.channels.get(&channel_name).is_some_and(|channel| channel.events.is_empty()) {
            self.close_channel(&channel_name);
        }

    }

    /// Process one heartbeat (§8, §11).
    fn process_heartbeat(&mut self) {

        let parts = match recv_message(&self.heartbeat.socket, 3) {
            Ok(Some(parts)) => parts,
            Ok(None) => return,
            Err(err) => {
                warn!(%err, "Failed to receive a heartbeat message, ignoring");
                return;
            }
        };

        let Some((topic, endian, call)) = split_message(&parts) else { return };

        // Nothing of it is used, but a heartbeat that does not decode is not from a Tango
        // device server (§8).
        if let Err(err) = CallInfo::read(call, endian) {
            warn!(%topic, %err, "Received a malformed heartbeat, ignoring");
            return;
        }

        let Some(channel_name) = topic.strip_suffix(HEARTBEAT_SUFFIX) else {
            warn!(%topic, "Received a heartbeat on a topic that is not a heartbeat one, ignoring");
            return;
        };

        let Some(channel) = self.channels.get_mut(channel_name) else {
            warn!(channel = %channel_name, "Received a heartbeat for an unknown channel, ignoring");
            return;
        };

        let interval = channel.last_heartbeat.elapsed();
        trace!(channel = %channel_name, ?interval, "Received a heartbeat");

        channel.last_heartbeat = Instant::now();

    }

    /// Process one event and hand it over to every subscriber of its topic (§8, §10).
    fn process_event(&mut self) {

        let parts = match recv_message(&self.event.socket, 4) {
            Ok(Some(parts)) => parts,
            Ok(None) => return,
            Err(err) => {
                warn!(%err, "Failed to receive an event message, ignoring");
                return;
            }
        };

        let Some((topic, endian, call)) = split_message(&parts) else { return };

        let call = match CallInfo::read(call, endian) {
            Ok(call) => call,
            Err(err) => {
                warn!(%topic, %err, "Received a malformed event call information, ignoring");
                return;
            }
        };

        trace!(%topic, ?endian, ?call, "Received an event");

        // The four filler octets, `0xDEC0DEC0` in the publisher's byte order, 
        // only realign the payload for cppTango's own decoding (§8).
        let Some(payload) = parts[3].get(4..) else {
            warn!(%topic, len = parts[3].len(), "Received an event whose payload is too short for its filler, ignoring");
            return;
        };

        let Some(event) = self.events.get_mut(topic) else {
            warn!(%topic, "Received an event for an unknown topic, ignoring");
            return;
        };

        // The first event of a subscription is accepted whatever its counter (§10).
        if call.counter > 1 && event.counter == 0 {
            event.counter = call.counter - 1;
        }

        // Signed, the counter going backward when the server restarted (§10).
        let missed = i64::from(call.counter) - i64::from(event.counter);
        event.counter = call.counter;

        if missed >= 2 {
            // Not reported to the subscribers: they get the latest event, and not every
            // event, so a dropped one is no different from a superseded one.
            warn!(%topic, missed = missed - 1, "Missed events, a ZMQ high water mark was probably reached");
            event.discarded = false;
        } else if missed == 0 {
            // One repeat is tolerated and dropped, but only one (§10).
            if !event.discarded {
                debug!(%topic, counter = call.counter, "Discarding a duplicated event");
                event.discarded = true;
                return;
            }
            event.discarded = false;
        } else {
            event.discarded = false;
        }

        let delivery = Delivery::Event {
            payload: Buf::from(payload),
            endian,
            failed: call.is_except,
        };

        // The last subscriber takes the delivery itself, the common case of a single one
        // not cloning it at all.
        if let Some((last, others)) = event.slots.split_last() {
            for slot in others {
                slot.push(delivery.clone());
            }
            last.push(delivery);
        }

    }

}

/// Receive the parts of a single multipart message, giving [`None`] when the socket has
/// nothing to give or the message does not have the expected number of parts (§8).
fn recv_message(socket: &zmq::Socket, expected: usize) -> Result<Option<Vec<zmq::Message>>> {

    let mut parts = Vec::with_capacity(expected);

    loop {

        match socket.recv_msg(zmq::DONTWAIT) {
            Ok(part) => parts.push(part),
            // Only the first part may be missing, which means there was nothing after all:
            // ZMQ delivers the parts of a message together.
            Err(zmq::Error::EAGAIN) if parts.is_empty() => return Ok(None),
            Err(err) => return Err(err.into()),
        }

        if !socket.get_rcvmore()? {
            break;
        }

    }

    if parts.len() != expected {
        warn!(got = parts.len(), expected, "Received a message with the wrong number of parts, ignoring");
        return Ok(None);
    }

    Ok(Some(parts))

}

/// Split the three parts every message starts with, the topic, the byte order and the call
/// information, giving [`None`] when one of them does not hold what it should (§8).
fn split_message(parts: &[zmq::Message]) -> Option<(&str, Endian, &[u8])> {

    let Ok(topic) = str::from_utf8(&parts[0]) else {
        warn!("Received a message whose topic is not valid text, ignoring");
        return None;
    };

    let [endian] = *parts[1] else {
        warn!(%topic, "Received a message whose byte order is not a single octet, ignoring");
        return None;
    };

    let endian = if endian == 0 { Endian::Big } else { Endian::Little };

    Some((topic, endian, &parts[2]))

}

/// The pairs of endpoints a channel can be connected through, which are all the ones the
/// server offers but the multicast ones (§5, §7), which of them is reachable being for the
/// connection itself to tell, see [`Worker::process_monitor`].
fn candidates(info: &SubscriptionInfo) -> Result<Vec<Endpoints>> {

    let candidates = info.endpoints.iter()
        .filter(|endpoints| !endpoints.is_multicast())
        .cloned()
        .collect::<Vec<_>>();

    if candidates.is_empty() {
        return Err(Error::not_implemented()
            .with_description("The device server only offers multicast endpoints for this event")
            .with_append_description(format_args!("Got: {:?}", info.endpoints)));
    }

    Ok(candidates)

}

impl Sub {

    fn new(socket: zmq::Socket) -> Self {
        Self {
            socket,
            connected: HashMap::new(),
            rcvhwm: None,
        }
    }

    fn is_connected(&self, endpoint: &str) -> bool {
        self.connected.contains_key(endpoint)
    }

    fn connect(&mut self, endpoint: &Arc<str>) -> Result<()> {

        if let Some(users) = self.connected.get_mut(endpoint) {
            *users += 1;
            return Ok(());
        }

        trace!(%endpoint, "Connecting to the endpoint");
        self.socket.connect(endpoint)?;
        self.connected.insert(Arc::clone(endpoint), 1);
        Ok(())
        
    }

    fn disconnect(&mut self, endpoint: &str) {

        let Some(users) = self.connected.get_mut(endpoint) else {
            error!(%endpoint, "Failed to disconnect from an endpoint that is not connected");
            return;
        };

        *users -= 1;
        if *users > 0 {
            return;
        }

        self.connected.remove(endpoint);
        trace!(%endpoint, "Disconnecting from the endpoint");

        // An endpoint already disconnected because the publisher went away is what was
        // asked for.
        match self.socket.disconnect(endpoint) {
            Ok(()) | Err(zmq::Error::ENOENT) => {}
            Err(err) => warn!(%endpoint, %err, "Failed to disconnect from the endpoint, ignoring"),
        }

    }

    fn subscribe(&self, topic: &str) -> Result<()> {
        trace!(%topic, "Subscribing to the topic");
        Ok(self.socket.set_subscribe(topic.as_bytes())?)
    }

    /// Unsubscribing cannot fail in a way that matters: a topic left subscribed only means
    /// messages that are received and then dropped.
    fn unsubscribe(&self, topic: &str) {
        trace!(%topic, "Unsubscribing from the topic");
        if let Err(err) = self.socket.set_unsubscribe(topic.as_bytes()) {
            warn!(%topic, %err, "Failed to unsubscribe from the topic, ignoring");
        }
    }

    fn set_rcvhwm(&mut self, hwm: i32) -> Result<()> {
        if self.rcvhwm != Some(hwm) {
            self.socket.set_rcvhwm(hwm)?;
            self.rcvhwm = Some(hwm);
        }
        Ok(())
    }

}

#[cfg(test)]
mod tests {

    use super::*;

    use std::time::SystemTime;

    use futures::future::{poll_fn, ready};
    use futures::stream::BoxStream;

    use crate::error::reason;
    use crate::dev::{AttrConfig, AttrMode, AttrQuality, AttrValueEvent, CommandInfo, DevInfo, DevState, DisplayLevel, Named, NamedResult, RecordedResult};
    use crate::value::{AttrDataFormat, AttrDataKind};
    use crate::future::block_on_timeout;

    use crate::corba::orb::cdr::CdrWriter;
    use crate::corba::tango::{write_attr, write_attr_info};

    /// An admin device answering the negotiation with what it is given, failing it as a
    /// device server without the ZMQ event system does when it is given nothing, and
    /// every other command with nothing. It records the commands it was sent.
    #[derive(Debug)]
    struct Admin {
        negotiation: Option<Value>,
        commands: Mutex<Vec<(String, Value)>>,
    }

    impl Admin {
        fn new(negotiation: Option<Value>) -> Self {
            Self { negotiation, commands: Mutex::new(Vec::new()) }
        }
    }

    fn not_implemented<'a, T: Send + 'a>() -> BoxFuture<'a, Result<T>> {
        Box::pin(ready(Err(Error::not_implemented())))
    }

    impl BackendDev for Admin {
        fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> { not_implemented() }
        fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> { not_implemented() }
        fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> { not_implemented() }
        fn info(&self) -> BoxFuture<'_, Result<DevInfo>> { not_implemented() }
        fn ping(&self) -> BoxFuture<'_, Result<()>> { not_implemented() }
        fn state(&self) -> BoxFuture<'_, Result<DevState>> { not_implemented() }
        fn status(&self) -> BoxFuture<'_, Result<Scalar<str>>> { not_implemented() }
        fn black_box(&self, _: Option<usize>) -> BoxFuture<'_, Result<Spectrum<str>>> { not_implemented() }
        fn command(&self, name: Scalar<str>, input: Value, _: DevSource) -> BoxFuture<'_, Result<Value>> {
            let name = name.value().to_string();
            self.commands.lock().unwrap().push((name.clone(), input));
            Box::pin(ready(match (&*name, &self.negotiation) {
                ("ZmqEventSubscriptionChange", Some(reply)) => Ok(reply.clone()),
                ("ZmqEventSubscriptionChange", None) => Err(Error::new(ErrorKind::Reason(reason::API_CommandNotFound))),
                _ => Ok(Value::Void),
            }))
        }
        fn read_attr(&self, _: Option<Spectrum<str>>, _: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> { not_implemented() }
        fn write_attr(&self, _: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> { not_implemented() }
        fn write_read_attr(&self, _: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> { not_implemented() }
        fn command_info(&self, _: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> { not_implemented() }
        fn attr_info(&self, _: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> { not_implemented() }
        fn set_attr_config(&self, _: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> { not_implemented() }
        fn command_history(&self, _: Scalar<str>, _: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<Value>>>> { not_implemented() }
        fn attr_history(&self, _: Scalar<str>, _: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<AttrRead<Value>>>>> { not_implemented() }
        fn subscribe_attr_value(&self, _: Arc<dyn BackendDev>, _: Scalar<str>, _: AttrValueEvent) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrRead<Value>>>>> { not_implemented() }
        fn subscribe_attr_ready(&self, _: Arc<dyn BackendDev>, _: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrReady>>>> { not_implemented() }
        fn subscribe_attr_config(&self, _: Arc<dyn BackendDev>, _: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrInfo>>>> { not_implemented() }
    }

    const TOPIC: &str = "tango://host:10000/sys/tg_test/1/double_scalar.idl5_change";
    const CHANNEL: &str = "tango://host:10000/dserver/tangotest/test";

    /// The publishers of a device server, bound on the loopback. The event one is an XPUB,
    /// which receives the subscriptions, so that a test sees them come and go.
    struct Server {
        heartbeat: zmq::Socket,
        event: zmq::Socket,
    }

    impl Server {

        fn new(context: &zmq::Context) -> Self {
            let bind = |socket: zmq::Socket| {
                socket.bind("tcp://127.0.0.1:*").unwrap();
                socket
            };
            Self {
                heartbeat: bind(context.socket(zmq::PUB).unwrap()),
                event: bind(context.socket(zmq::XPUB).unwrap()),
            }
        }

        fn endpoints(&self) -> [String; 2] {
            [
                self.heartbeat.get_last_endpoint().unwrap().unwrap(),
                self.event.get_last_endpoint().unwrap().unwrap(),
            ]
        }

        fn publish(&self, counter: u32, payload: &[u8]) {
            self.publish_call(counter, false, payload);
        }

        /// Publish an event, whose payload is an error stack if `except` is set.
        fn publish_call(&self, counter: u32, except: bool, payload: &[u8]) {

            // `ZmqCallInfo`, little endian.
            let mut call = Vec::new();
            call.extend(1i32.to_le_bytes());
            call.extend(counter.to_le_bytes());
            call.extend(15u32.to_le_bytes());
            call.extend(b"push_zmq_event\0");
            call.push(0);
            call.extend(0u32.to_le_bytes());
            call.push(except as u8);

            let mut data = 0xDEC0DEC0u32.to_le_bytes().to_vec();
            data.extend(payload);

            self.event.send_multipart([TOPIC.as_bytes(), &[1], &call, &data], 0).unwrap();

        }

        /// The next subscription or unsubscription of the event topic the event socket
        /// receives, true for a subscription.
        fn next_subscription(&self) -> Option<bool> {
            self.event.set_rcvtimeo(5000).unwrap();
            loop {
                let message = self.event.recv_bytes(0).ok()?;
                if let Some((&flag, topic)) = message.split_first() && topic == TOPIC.as_bytes() {
                    return Some(flag == 1);
                }
            }
        }

    }

    /// The negotiation reply of a cppTango 9.3+ server of a v6 device publishing on those
    /// endpoints.
    fn negotiation(endpoints: &[&str]) -> Value {
        let mut strings = endpoints.to_vec();
        strings.extend([TOPIC, CHANNEL]);
        Value::LongString(
            Spectrum::from(vec![1000, 6, 1000, 0, 0, 40305]),
            Spectrum::<str>::from(&strings[..]),
        )
    }

    fn request(endpoints: &[&str]) -> (SubscribeRequest, Arc<Slot<Delivery>>) {

        let info = SubscriptionInfo::from_value(negotiation(endpoints)).unwrap();

        let slot = Arc::new(Slot::new());

        (SubscribeRequest {
            admin: Arc::new(Admin::new(None)),
            dev_name: Arc::from("sys/tg_test/1"),
            obj_name: Arc::from("double_scalar"),
            kind: EventKind::Value(AttrValueEvent::Change),
            info,
        }, slot)

    }

    /// Publish until the slot is handed the event, the subscription only reaching the
    /// publisher some time after it was declared (§5.1).
    fn receive(server: &Server, slot: &Slot<Delivery>, payload: &[u8]) -> Delivery {
        for counter in 1..100 {
            server.publish(counter, payload);
            if let Ok(Some(delivery)) = block_on_timeout(poll_fn(|cx| slot.poll_take(cx)), Duration::from_millis(20)) {
                return delivery;
            }
        }
        panic!("the event never arrived");
    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn subscribe_and_receive() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let [heartbeat, event] = server.endpoints();

        let consumer = Consumer::start().unwrap();
        let (request, slot) = request(&[&heartbeat, &event]);

        let reply = consumer.send_subscribe(request, Arc::clone(&slot)).unwrap();
        block_on_timeout(reply, Duration::from_secs(5)).unwrap().unwrap().unwrap();

        let Delivery::Event { payload, endian, failed } = receive(&server, &slot, b"payload") else {
            panic!("expected an event");
        };

        assert_eq!(payload.as_slice(), b"payload");
        assert_eq!(endian, Endian::Little);
        assert!(!failed);

        consumer.unsubscribe(Arc::from(TOPIC), slot);

    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn subscribe_chooses_reachable_endpoints() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let [heartbeat, event] = server.endpoints();

        let consumer = Consumer::start().unwrap();

        // The first pair is of an interface that cannot be reached from here.
        let (request, slot) = request(&["tcp://127.0.0.1:1", "tcp://127.0.0.1:2", &heartbeat, &event]);

        let reply = consumer.send_subscribe(request, Arc::clone(&slot)).unwrap();
        block_on_timeout(reply, Duration::from_secs(5)).unwrap().unwrap().unwrap();

        assert!(matches!(receive(&server, &slot, b"x"), Delivery::Event { .. }));

    }

    #[test]
    fn candidates_skip_multicast() {

        let (both, _slot) = request(&["tcp://a:1", "tcp://a:2", "tcp://b:1", "epgm://eth0;239.0.0.1:5555"]);
        let found = candidates(&both.info).unwrap();
        assert_eq!(found.len(), 1);
        assert_eq!(&*found[0].heartbeat, "tcp://a:1");

        let (multicast, _slot) = request(&["tcp://a:1", "epgm://eth0;239.0.0.1:5555"]);
        assert_eq!(candidates(&multicast.info).unwrap_err().kind(), ErrorKind::NotImplemented);

    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn subscribe_times_out() {

        let consumer = Consumer::start().unwrap();
        let (request, slot) = request(&["tcp://127.0.0.1:1", "tcp://127.0.0.1:2"]);

        let reply = consumer.send_subscribe(request, Arc::clone(&slot)).unwrap();
        let err = block_on_timeout(reply, Duration::from_secs(5)).unwrap().unwrap().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::EventNotConnected);

    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn attach_to_subscribed_event() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let [heartbeat, event] = server.endpoints();

        let consumer = Consumer::start().unwrap();

        let (first, first_slot) = request(&[&heartbeat, &event]);
        block_on_timeout(consumer.send_subscribe(first, Arc::clone(&first_slot)).unwrap(), Duration::from_secs(5)).unwrap().unwrap().unwrap();

        let (second, second_slot) = request(&[&heartbeat, &event]);
        block_on_timeout(consumer.send_subscribe(second, Arc::clone(&second_slot)).unwrap(), Duration::from_secs(5)).unwrap().unwrap().unwrap();

        // Both are handed the same event.
        receive(&server, &first_slot, b"x");
        assert!(matches!(block_on_timeout(poll_fn(|cx| second_slot.poll_take(cx)), Duration::from_secs(1)), Ok(Some(Delivery::Event { .. }))));

        // The first one leaving does not take the event away from the second.
        consumer.unsubscribe(Arc::from(TOPIC), first_slot);
        receive(&server, &second_slot, b"y");

    }

    fn subscribe_stream<T>(consumer: &Consumer, server: &Server, decode: Decode<T>) -> Subscription<T> {
        let [heartbeat, event] = server.endpoints();
        let (request, _slot) = request(&[&heartbeat, &event]);
        let subscribing = consumer.subscribe(request, decode).unwrap();
        block_on_timeout(subscribing, Duration::from_secs(5)).unwrap().unwrap()
    }

    /// Publish until the stream gives the event, the subscription only reaching the
    /// publisher some time after it was declared (§5.1).
    fn next_event<T>(server: &Server, sub: &mut (impl Stream<Item = Result<T>> + Unpin), except: bool, payload: &[u8]) -> Result<T> {
        for counter in 1..100 {
            server.publish_call(counter, except, payload);
            if let Ok(item) = block_on_timeout(sub.next(), Duration::from_millis(20)) {
                return item.expect("the stream never ends");
            }
        }
        panic!("the event never arrived");
    }

    fn encode(write: impl FnOnce(&mut CdrWriter)) -> Vec<u8> {
        let mut writer = CdrWriter::new();
        write(&mut writer);
        writer.into_buf().freeze().as_slice().to_vec()
    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn value_event() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let consumer = Consumer::start().unwrap();
        let mut sub = subscribe_stream(&consumer, &server, decode_value);

        let sent = AttrRead {
            value: Some(Value::from(3.5f64)),
            w_value: None,
            quality: AttrQuality::Valid,
            time: SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000),
        };

        let payload = encode(|writer| write_attr(writer, DevVersion::V6, "double_scalar", &Ok(sent.clone())).unwrap());
        assert_eq!(next_event(&server, &mut sub, false, &payload).unwrap(), sent);

    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn failed_event() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let consumer = Consumer::start().unwrap();
        let mut sub = subscribe_stream(&consumer, &server, decode_value);

        let sent = Error::new(ErrorKind::Internal).with_description("the reading failed");
        let payload = encode(|writer| writer.write(&sent).unwrap());

        let err = next_event(&server, &mut sub, true, &payload).unwrap_err();
        assert_eq!(err.description(), "the reading failed");

    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn lost() {

        let consumer = Consumer::start().unwrap();
        let slot = Arc::new(Slot::new());

        let mut sub = Subscription {
            consumer,
            topic: Arc::from("topic"),
            slot: Arc::clone(&slot),
            version: DevVersion::V6,
            decode: |_, _| Ok(()),
            ended: false,
        };

        // The last event comes before the loss, which ends the stream.
        slot.push(Delivery::Event { payload: Buf::from(&[][..]), endian: Endian::Big, failed: false });
        slot.end();

        let item = block_on_timeout(sub.next(), Duration::from_secs(1)).unwrap().unwrap();
        assert!(item.is_ok());
        let item = block_on_timeout(sub.next(), Duration::from_secs(1)).unwrap().unwrap();
        assert_eq!(item.unwrap_err().kind(), ErrorKind::DevNotConnected);
        assert!(block_on_timeout(sub.next(), Duration::from_secs(1)).unwrap().is_none());

    }

    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn dropping_unsubscribes() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let consumer = Consumer::start().unwrap();

        let sub = subscribe_stream(&consumer, &server, decode_value);
        assert_eq!(server.next_subscription(), Some(true));

        drop(sub);
        assert_eq!(server.next_subscription(), Some(false));

    }

    /// A subscription the worker made, but whose future was dropped before it was polled
    /// again, is undone all the same.
    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn dropping_subscribing_unsubscribes() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let consumer = Consumer::start().unwrap();

        let [heartbeat, event] = server.endpoints();
        let (request, _slot) = request(&[&heartbeat, &event]);
        let subscribing = consumer.subscribe(request, decode_value).unwrap();

        assert_eq!(server.next_subscription(), Some(true));
        std::thread::sleep(Duration::from_millis(200));

        drop(subscribing);
        assert_eq!(server.next_subscription(), Some(false));

    }

    #[test]
    fn decode_config_versions() {

        let info = AttrInfo {
            config: AttrConfig::default(),
            mode: AttrMode::ReadWrite,
            data_kind: AttrDataKind::Double,
            data_format: AttrDataFormat::Scalar,
            display_level: DisplayLevel::Operator,
            memorized: None,
        };

        // An unset label goes out as the name of the attribute, as cppTango sends it.
        let mut expected = info.clone();
        expected.config.label = Some("double_scalar".into());

        for (version, structure) in [(DevVersion::V4, DevVersion::V3), (DevVersion::V5, DevVersion::V5), (DevVersion::V6, DevVersion::V5)] {
            let mut writer = CdrWriter::new();
            write_attr_info(&mut writer, structure, "double_scalar", &info).unwrap();
            let buf = writer.into_buf().freeze();
            assert_eq!(decode_config(&mut CdrReader::new(&buf), version).unwrap(), expected, "{version:?}");
        }

    }

    #[test]
    fn decode_ready_event() {

        let mut writer = CdrWriter::new();
        writer.write_string("double_scalar").unwrap();
        writer.write_long(5).unwrap();
        writer.write_long(42).unwrap();
        let buf = writer.into_buf().freeze();

        let ready = decode_ready(&mut CdrReader::new(&buf), DevVersion::V6).unwrap();
        assert_eq!(ready, AttrReady { data_kind: AttrDataKind::Double, counter: 42 });
        assert!(decode_ready(&mut CdrReader::empty(), DevVersion::V6).is_err());

    }

    /// The whole of a subscription: the negotiation with the admin device, then the
    /// subscription of what it answered (§4).
    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn negotiate_and_subscribe() {

        let context = zmq::Context::new();
        let server = Server::new(&context);
        let [heartbeat, event] = server.endpoints();

        let consumer = Consumer::start().unwrap();
        let admin = Arc::new(Admin::new(Some(negotiation(&[&heartbeat, &event]))));
        let dev_name = ready(Ok(Arc::<str>::from("sys/tg_test/1")));

        let subscribing = super::super::subscribe(&consumer, dev_name, admin.clone(), "Double_Scalar", EventKind::Value(AttrValueEvent::Change), decode_value);
        let mut sub = block_on_timeout(subscribing, Duration::from_secs(5)).unwrap().unwrap();

        // The plain event name and the lowercased attribute name go out, with the highest
        // release this client knows.
        let commands = admin.commands.lock().unwrap().clone();
        assert_eq!(commands, [(
            "ZmqEventSubscriptionChange".to_string(),
            Value::from(["sys/tg_test/1", "double_scalar", "subscribe", "change", "6"]),
        )]);

        let sent = AttrRead {
            value: Some(Value::from(1.5f64)),
            w_value: None,
            quality: AttrQuality::Valid,
            time: SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000),
        };

        let payload = encode(|writer| write_attr(writer, DevVersion::V6, "double_scalar", &Ok(sent.clone())).unwrap());
        assert_eq!(next_event(&server, &mut sub, false, &payload).unwrap(), sent);

    }

    /// A device server without the ZMQ event system leaves the subscription to another way
    /// of subscribing (§4).
    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn negotiate_without_zmq() {

        let consumer = Consumer::start().unwrap();
        let admin = Arc::new(Admin::new(None));
        let dev_name = ready(Ok(Arc::<str>::from("sys/tg_test/1")));

        let subscribing = super::super::subscribe(&consumer, dev_name, admin, "double_scalar", EventKind::Config, decode_config);
        let Err(err) = block_on_timeout(subscribing, Duration::from_secs(5)).unwrap() else {
            panic!("subscribed to a device server without the ZMQ event system");
        };
        assert_eq!(err.kind(), ErrorKind::NotHandled);

    }

}