rstango 0.4.0

//! The supplying side of the event system: the publishers of a device server, and the
//! negotiation that hands them out, the equivalent of cppTango's `ZmqEventSupplier` and of
//! the event commands of its `DServer`.
//!
//! The publishers are only made once a client first asks for an event, a server nobody
//! subscribes to having no use for them. They are then owned by a single thread, the
//! [`Worker`], as the sockets of the consumer are: it publishes the heartbeat of the server
//! (§11), and the events subscribed to, which the requests of the clients tell it of
//! through a [`Command`].
//!
//! An event is what the served device gives: a client subscribing has the backend subscribe
//! to the device, see [`crate::backend::BackendDev::subscribe_attr_value`], and the stream
//! it gets is handed to the worker, which polls it and publishes every item of it. One
//! stream is kept per event, whatever the number of clients, and dropped once no client
//! confirmed the event for long enough (§11).

use std::collections::hash_map::{Entry, HashMap};
use std::task::{Context, Poll, Wake, Waker};
use std::time::{Duration, Instant};
use std::sync::{Arc, Mutex};
use std::ops::ControlFlow;
use std::fmt::Write as _;
use std::net::IpAddr;
use std::thread;
use std::mem;

use futures::stream::{self, AbortHandle, BoxStream, SelectAll, StreamExt};
use futures::channel::oneshot;
use futures::future;

use tracing::{debug, error, trace, warn};

use crate::corba::tango::{write_attr, write_attr_info, write_attr_ready};
use crate::dev::{AttrInfo, AttrRead, AttrReady, AttrValueEvent};
use crate::error::{Error, ErrorKind, Result, reason};
use crate::corba::{LIB_VERSION, DevVersion};
use crate::corba::orb::endian::Endian;
use crate::corba::orb::cdr::CdrWriter;
use crate::value::{Scalar, Spectrum, Value};
use crate::backend::BackendDev;

use super::{
    CALL_INFO_VERSION, DEFAULT_LINGER, EVENT_NAMES, HEARTBEAT_SUFFIX,
    IDL_COMPAT_PREFIX, NO_DATABASE_MARKER, PAYLOAD_FILLER, PUB_HWM, RESUBSCRIBE_PERIOD,
    SUB_HWM, UNPREFIXED_EVENT_NAMES,
};


/// The interval the heartbeat is published at, which cppTango publishes from a polling
/// thread every 9 seconds, clients giving up on a server after 10 (§11).
const HEARTBEAT_PERIOD: Duration = Duration::from_secs(9);

/// The message cppTango publishes on the event socket along with every heartbeat, on no
/// topic any client subscribes to, which keeps the connections of the event socket alive
/// when no event is published.
const KEEP_ALIVE_MESSAGE: &[u8] = b"I like beer";

/// The items the worker takes from the streams of the devices at once, before it looks at
/// its commands again, so that a device publishing without a pause holds nothing up.
const MAX_ITEMS_AT_ONCE: usize = 256;

// =========================== //
//          Supplier           //
// =========================== //

/// What a device server publishes its events with, see the module documentation.
#[derive(Debug)]
pub struct Supplier {
    /// What the names of the events of this server start with, `tango://host:port/` (§3).
    prefix: Arc<str>,
    /// Whether the server runs without a database, which marks the names of its events.
    no_database: bool,
    /// The name of the channel of this server, see [`Self::channel_name`].
    channel_name: Arc<str>,
    /// The name of the admin device of this server, lowercased, which is given to the
    /// devices subscribed to.
    admin_name: Box<str>,
    /// The address the publishers are bound to, every interface if unspecified.
    bind_ip: IpAddr,
    bind_heartbeat_ports: (u16, u16),
    bind_event_ports: (u16, u16),
    /// The addresses the publishers are told to be reachable at, the first one first.
    addrs: Box<[IpAddr]>,
    /// The publishers, once they are made, and none again once the server stopped.
    state: Mutex<State>,
}

/// Whether the publishers of a supplier are made, see [`Supplier::state`].
#[derive(Debug)]
enum State {
    Idle,
    Running(Running),
    Stopped,
}

/// What the requests of the clients hold of the publishers once they are made, the
/// publishers themselves being the worker's.
#[derive(Debug)]
struct Running {
    /// The endpoints the publishers are reachable at, heartbeat and event in turn, one pair
    /// per address, which is how the negotiation lists them (§4).
    endpoints: Vec<Arc<str>>,
    /// The commands of the worker, which stops once this is dropped.
    command_tx: crossbeam_channel::Sender<Command>,
}

/// What the requests of the clients tell the worker.
enum Command {
    /// A client subscribed to that event, by its name without the IDL compatibility prefix,
    /// to be published on that topic, in that version of its structure, from what the
    /// device gives. The device is only kept for an event not published yet.
    Subscribe {
        name: Arc<str>,
        obj_name: Arc<str>,
        topic: Arc<str>,
        version: DevVersion,
        source: BoxStream<'static, Payload>,
    },
    /// A client confirmed those events, by their name without the IDL compatibility prefix,
    /// to be told back the ones that are not published.
    Confirm {
        names: Vec<Arc<str>>,
        reply: oneshot::Sender<Vec<Arc<str>>>,
    },
}

/// What a device gives for an event, which is what goes out in its payload.
enum Payload {
    Value(Result<AttrRead<Value>>),
    Config(Result<AttrInfo>),
    Ready(Result<AttrReady>),
}

/// An item of the stream of a device, tagged with the event and the subscription to the
/// device it is of, or the end of that stream when there is no payload.
struct Tagged {
    /// The event, by its name without the IDL compatibility prefix.
    name: Arc<str>,
    /// The subscription to the device, see [`Published::id`].
    id: u64,
    /// What the device gave, none at the end of its stream.
    payload: Option<Payload>,
}

/// The thread owning the publishers of a server, see the module documentation.
struct Worker {
    command_rx: crossbeam_channel::Receiver<Command>,
    /// What the streams of the devices wake the worker up with, see [`WakeWorker`].
    wake_rx: crossbeam_channel::Receiver<()>,
    waker: Waker,
    heartbeat: zmq::Socket,
    event: zmq::Socket,
    /// The topic the heartbeat is published on.
    heartbeat_topic: Box<str>,
    /// Where the `ZmqCallInfo` of every message, and the payload of every event, are
    /// encoded, cleared before each of them so that their allocation is reused.
    call_info_writer: CdrWriter,
    payload_writer: CdrWriter,
    /// Whether the next heartbeat is published twice, which cppTango does after a new
    /// subscription, the first message on a new connection being sometimes lost (§10).
    double_heartbeat: bool,
    /// The events published, by their name without the IDL compatibility prefix.
    events: HashMap<Arc<str>, Published>,
    /// The streams of the devices of every event published.
    sources: SelectAll<BoxStream<'static, Tagged>>,
    /// The identifier of the next subscription to a device, which tells the items of a
    /// stream that was dropped from the ones of the stream that replaced it.
    next_id: u64,
}

/// An event published.
struct Published {
    /// The subscription to the device the event is published from.
    id: u64,
    /// What stops the stream of that subscription, which it is not by being dropped, the
    /// stream being the worker's, see [`Worker::sources`].
    abort: AbortHandle,
    /// The name of the attribute, as the payload carries it.
    obj_name: Arc<str>,
    /// The topics the event is published on, each with the version of the structure of its
    /// payload, which is one per release of the clients that subscribed (§9).
    topics: Vec<(Arc<str>, DevVersion)>,
    /// The counter of the next event, which cppTango starts at 1 and never resets (§10).
    counter: u32,
    /// Whether the next event is published twice, which cppTango does after a new
    /// subscription, as for the heartbeat (§10).
    double: bool,
    /// When a client last subscribed to it or confirmed it, which it is published for
    /// until long after (§11).
    confirmed: Instant,
}

/// What wakes the worker up when a stream of a device has something.
struct WakeWorker(crossbeam_channel::Sender<()>);

impl Supplier {

    /// A supplier for the server of that admin device, registered in the database at that
    /// host and port, or served without one at the given host and port, its own, when there
    /// is none.
    ///
    /// The publishers are bound to the given address, and told reachable at the given ones,
    /// which are the ones the server itself is reachable at.
    pub fn new(
        admin_name: &str,
        tango_host: Option<(&str, u16)>,
        corba_host: &str,
        corba_port: u16,
        bind_ip: IpAddr,
        bind_heartbeat_ports: (u16, u16),
        bind_event_ports: (u16, u16),
        addrs: &[IpAddr],
    ) -> Self {

        let no_database = tango_host.is_none();
        let (prefix_host, prefix_port) = tango_host.unwrap_or((corba_host, corba_port));
        let prefix = format!("tango://{prefix_host}:{prefix_port}/").to_lowercase();

        // cppTango names the channel after the admin device, without the marker of a
        // server without a database, which the heartbeat topic has (§3).
        let mut channel_name = String::with_capacity(prefix.len() + admin_name.len());
        channel_name.push_str(&prefix);
        channel_name.push_str(&admin_name.to_lowercase());

        Self {
            prefix: Arc::from(prefix),
            no_database,
            channel_name: Arc::from(channel_name),
            admin_name: admin_name.to_ascii_lowercase().into_boxed_str(),
            bind_ip,
            bind_heartbeat_ports,
            bind_event_ports,
            addrs: Box::from(addrs),
            state: Mutex::new(State::Idle),
        }

    }

    /// Answer the `ZmqEventSubscriptionChange` command, given the devices of this server by
    /// their lowercased name, see
    /// [`crate::backend::BackendServer::zmq_event_subscription_change`].
    ///
    /// A subscription subscribes to the device, which is what checks that it has the
    /// attribute and publishes that event: its refusal is the answer to the client.
    pub async fn subscription_change(
        &self,
        args: &Spectrum<str>,
        served: impl Fn(&str) -> Option<Arc<dyn BackendDev>>,
        dev_version: u32,
    ) -> Result<(Spectrum<i32>, Spectrum<str>)> {

        // Only the first five arguments mean anything, the others being ignored as cppTango
        // does, and the ones missing are left empty.
        let mut fields = [""; 5];
        for (field, arg) in fields.iter_mut().zip(args.iter()) {
            *field = arg;
        }

        let [
            dev_name, 
            obj_name, 
            action, 
            event_name, 
            version_arg,
        ] = fields;

        match args.len() {
            0 | 1 => {
                if dev_name.eq_ignore_ascii_case("info") {
                    return Ok(self.info());
                }
                return Err(Error::new(ErrorKind::Reason(reason::API_WrongNumberOfArgs))
                    .with_description("Input has too few arguments")
                    .with_append_description("Expected: device name, attribute/pipe name, action, event name"));
            }
            2 | 3 => {
                return Err(Error::new(ErrorKind::Reason(reason::API_WrongNumberOfArgs))
                    .with_description("Input has too few arguments")
                    .with_append_description("Expected: device name, attribute/pipe name, action, \
                        event name, <Tango lib release>"));
            }
            _ => {}
        }

        let raw_event = event_name.to_lowercase();
        let event = strip_idl_prefix(&raw_event);

        if !EVENT_NAMES.contains(&event) {
            return Err(Error::new(ErrorKind::Reason(reason::API_WrongNumberOfArgs))
                .with_description(format!("Event type {raw_event:?} is invalid")));
        }

        // The release the client advertises, else the one its event name has, else the
        // one cppTango assumes of a client that says nothing (§4).
        let client_version = version_arg.parse::<u32>().ok()
            .filter(|&number| number != 0)
            .or_else(|| raw_event.starts_with(IDL_COMPAT_PREFIX).then_some(5))
            .unwrap_or(match event {
                "attr_conf" => 3,
                "alarm" => 6,
                _ => 4,
            });

        let version = client_version.min(dev_version);

        let Some(dev) = served(&dev_name.to_ascii_lowercase()) else {
            return Err(Error::new(ErrorKind::Reason(reason::API_DeviceNotFound))
                .with_description(format!("Device {dev_name:?} is not served here")));
        };

        // The lock is not held across the subscription to the device below.
        let (endpoints, command_tx) = {
            let mut state = self.state.lock().unwrap();
            let running = self.ensure_running(&mut state)?;
            (running.endpoints.clone(), running.command_tx.clone())
        };

        let numbers = [
            LIB_VERSION,
            dev_version as i32,
            SUB_HWM,
            0,
            0,
            zmq_version(),
        ];

        let compat = version >= 5 && !UNPREFIXED_EVENT_NAMES.contains(&event);
        let topic = self.event_name_with(dev_name, obj_name, compat.then_some(IDL_COMPAT_PREFIX), event);

        // Any other action is answered all the same, which is what cppTango does, but
        // only a subscription is recorded: nothing is ever unsubscribed, an event no
        // client confirms anymore being dropped after a while instead (§12).
        if action == "subscribe" {

            let admin = served(&self.admin_name).ok_or_else(|| Error::new(ErrorKind::Internal)
                .with_description("The admin device of this server is not served"))?;

            let source = subscribe_dev(&*dev, admin, obj_name, event).await?;

            // The structure of the payload is the one of the release the client and the
            // device have in common, which has no form of its own past v5 (§8).
            let version = match (event, version) {
                (_, 5..) => DevVersion::V5,
                ("attr_conf", _) => DevVersion::V3,
                _ => DevVersion::V4,
            };

            // The worker only stops once the supplier did, which nothing is told of then.
            let _ = command_tx.send(Command::Subscribe {
                name: self.event_name(dev_name, obj_name, event),
                obj_name: Arc::from(obj_name),
                topic: Arc::clone(&topic),
                version,
                source,
            });

            debug!(dev = %dev_name, attr = %obj_name, %event, "A client subscribed to the event");

        }

        let mut strings = Vec::with_capacity(endpoints.len() + 2);
        strings.extend(endpoints);
        strings.push(topic);
        strings.push(Arc::clone(&self.channel_name));

        Ok((Spectrum::from(&numbers[..]), Spectrum::from(&strings[..])))

    }

    /// Answer the `EventConfirmSubscription` command, given whether the devices it names,
    /// lowercased, are of this server, see
    /// [`crate::backend::BackendServer::event_confirm_subscription`].
    ///
    /// Only an event published is confirmed: one that is not anymore, because no client
    /// confirmed it for too long or the device stopped publishing it, takes a new
    /// subscription to be published again.
    pub async fn confirm_subscription(&self, args: &Spectrum<str>, is_served: impl Fn(&str) -> bool) -> Result<()> {

        // What is left of an incomplete triplet is ignored, as cppTango does.
        let mut names = Vec::new();
        let mut args = args.iter();
        while let (Some(dev_name), Some(obj_name), Some(event_name)) = (args.next(), args.next(), args.next()) {

            if !is_served(&dev_name.to_ascii_lowercase()) {
                return Err(Error::new(ErrorKind::Reason(reason::API_DeviceNotFound))
                    .with_description(format!("Device {dev_name:?} is not served here")));
            }

            let event_name = event_name.to_lowercase();
            names.push(self.event_name(dev_name, obj_name, strip_idl_prefix(&event_name)));

        }

        if names.is_empty() {
            return Ok(());
        }

        // Publishers that are not made yet publish nothing, and are not made for this.
        let command_tx = match &*self.state.lock().unwrap() {
            State::Running(running) => running.command_tx.clone(),
            State::Idle => return Err(not_published_err(&names)),
            State::Stopped => return Err(shutting_down_err()),
        };

        let (reply_tx, reply_rx) = oneshot::channel();
        let _ = command_tx.send(Command::Confirm { names, reply: reply_tx });

        // The worker only drops the reply once the supplier stopped.
        let missing = reply_rx.await.map_err(|_| shutting_down_err())?;
        if !missing.is_empty() {
            return Err(not_published_err(&missing));
        }

        Ok(())

    }

    /// Stop publishing, for good, closing the publishers if they were made.
    pub fn stop(&self) {
        *self.state.lock().unwrap() = State::Stopped;
    }

    /// The answer to the `info` form of the `ZmqEventSubscriptionChange` command, which
    /// says where the publishers are, in words.
    fn info(&self) -> (Spectrum<i32>, Spectrum<str>) {

        let strings = match &*self.state.lock().unwrap() {
            State::Running(running) => running.endpoints.as_chunks::<2>().0.iter()
                .enumerate()
                .flat_map(|(index, pair)| {
                    let alternate = if index == 0 { "" } else { "Alternate " };
                    let heartbeat = if index == 0 { "Heartbeat" } else { "heartbeat" };
                    let event = if index == 0 { "Event" } else { "event" };
                    [
                        format!("{alternate}{heartbeat}: {}", pair[0]),
                        format!("{alternate}{event}: {}", pair[1]),
                    ]
                })
                .collect(),
            State::Idle | State::Stopped => vec!["No ZMQ event yet!".to_string(), String::new()],
        };

        (Spectrum::from(&[LIB_VERSION][..]), Spectrum::from(&strings[..]))

    }

    /// The full name of an event of this server, which is what its counter is kept under,
    /// see [`Self::event_name_with`].
    fn event_name(&self, dev_name: &str, obj_name: &str, event: &str) -> Arc<str> {
        self.event_name_with(dev_name, obj_name, None, event)
    }

    /// The full name of an event of this server, which is the topic it is published on
    /// when given the IDL compatibility prefix where it applies (§3).
    fn event_name_with(&self, dev_name: &str, obj_name: &str, compat: Option<&str>, event: &str) -> Arc<str> {

        let mut name = String::with_capacity(
            self.prefix.len() + 
            dev_name.len() + 
            obj_name.len() + 
            NO_DATABASE_MARKER.len() + 
            IDL_COMPAT_PREFIX.len() + 
            event.len() + 2);  // The '/' before the attribute and the '.' before the event.

        name.push_str(&self.prefix);
        name.push_str(dev_name);
        if event != "intr_change" {
            name.push('/');
            name.push_str(obj_name);
        }
        name.make_ascii_lowercase();

        if self.no_database {
            name.push_str(NO_DATABASE_MARKER);
        }

        name.push('.');
        name.extend(compat);
        name.push_str(event);

        Arc::from(name)

    }

    /// The publishers, made if they are not yet, failing if the server stopped.
    fn ensure_running<'a>(&self, state: &'a mut State) -> Result<&'a Running> {

        if let State::Idle = state {
            *state = State::Running(self.start()?);
        }

        match state {
            State::Running(running) => Ok(running),
            State::Idle | State::Stopped => Err(shutting_down_err()),
        }

    }

    /// Make the publishers, bound and handed to their worker.
    fn start(&self) -> Result<Running> {

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

        let heartbeat = context.socket(zmq::PUB)?;
        heartbeat.set_linger(DEFAULT_LINGER)?;
        let heartbeat_port = bind(&heartbeat, self.bind_ip, self.bind_heartbeat_ports)?;

        let event = context.socket(zmq::PUB)?;
        event.set_linger(DEFAULT_LINGER)?;
        event.set_sndhwm(PUB_HWM)?;
        let event_port = bind(&event, self.bind_ip, self.bind_event_ports)?;

        let endpoints = self.addrs.iter()
            .flat_map(|addr| [endpoint(*addr, heartbeat_port), endpoint(*addr, event_port)])
            .collect::<Vec<_>>();

        debug!(?endpoints, "Publishing the events of the server");

        let mut heartbeat_topic = String::with_capacity(
            self.channel_name.len() + 
            NO_DATABASE_MARKER.len() + 
            HEARTBEAT_SUFFIX.len()
        );
        heartbeat_topic.push_str(&self.channel_name);
        if self.no_database {
            heartbeat_topic.push_str(NO_DATABASE_MARKER);
        }
        heartbeat_topic.push_str(HEARTBEAT_SUFFIX);

        let (command_tx, command_rx) = crossbeam_channel::unbounded();

        // A single wake up pending is all it takes, the worker polling every stream then.
        let (wake_tx, wake_rx) = crossbeam_channel::bounded(1);

        let worker = Worker {
            command_rx,
            wake_rx,
            waker: Waker::from(Arc::new(WakeWorker(wake_tx))),
            heartbeat,
            event,
            heartbeat_topic: heartbeat_topic.into_boxed_str(),
            call_info_writer: CdrWriter::new(),
            payload_writer: CdrWriter::new(),
            double_heartbeat: false,
            events: HashMap::new(),
            sources: SelectAll::new(),
            next_id: 0,
        };

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

        Ok(Running {
            endpoints,
            command_tx,
        })

    }

}

impl Wake for WakeWorker {

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

    fn wake_by_ref(self: &Arc<Self>) {
        // Full means a wake up is already pending, and disconnected that the worker is gone.
        let _ = self.0.try_send(());
    }

}

impl Worker {

    /// Publish the heartbeat and the events, and take the commands, until the supplier
    /// stops.
    fn main(mut self) {

        // Owned here, the receivers being borrowed by the selection below while the worker
        // is taken mutably by what it selected.
        let command_rx = self.command_rx.clone();
        let wake_rx = self.wake_rx.clone();

        let mut heartbeat_deadline = Instant::now() + HEARTBEAT_PERIOD;

        loop {

            self.poll_sources();

            let timeout = heartbeat_deadline.saturating_duration_since(Instant::now());

            crossbeam_channel::select! {
                recv(command_rx) -> command => match command {
                    Ok(command) => self.process_command(command),
                    Err(_) => break,
                },
                recv(wake_rx) -> _ => {}
                default(timeout) => {
                    if let Err(err) = self.send_heartbeat() {
                        warn!(%err, "Failed to publish the heartbeat, ignoring");
                    }
                    self.expire();
                    heartbeat_deadline += HEARTBEAT_PERIOD;
                }
            }

        }

        debug!("Stopped publishing the events of the server");

    }

    fn process_command(&mut self, command: Command) {
        match command {
            Command::Subscribe { name, obj_name, topic, version, source } => {
                self.subscribe(name, obj_name, topic, version, source);
            }
            Command::Confirm { names, reply } => {

                let now = Instant::now();
                let missing = names.into_iter()
                    .filter(|name| match self.events.get_mut(name) {
                        Some(published) => {
                            published.confirmed = now;
                            trace!(event = %name, "A client confirmed its subscription to the event");
                            false
                        }
                        None => {
                            debug!(event = %name, "A client confirmed its subscription to an event that is not published");
                            true
                        }
                    })
                    .collect();

                let _ = reply.send(missing);

            }
        }
    }

    /// Publish that event on that topic from now on, from that stream if it is not
    /// published yet, which is otherwise dropped.
    fn subscribe(&mut self, name: Arc<str>, obj_name: Arc<str>, topic: Arc<str>, version: DevVersion, source: BoxStream<'static, Payload>) {

        let now = Instant::now();
        self.double_heartbeat = true;

        match self.events.entry(name) {
            Entry::Occupied(mut entry) => {
                let published = entry.get_mut();
                if !published.topics.iter().any(|(published, _)| *published == topic) {
                    published.topics.push((topic, version));
                }
                published.double = true;
                published.confirmed = now;
            }
            Entry::Vacant(entry) => {

                let id = self.next_id;
                self.next_id += 1;

                let (source, abort) = stream::abortable(source);
                let tag = Arc::clone(entry.key());
                let end = Tagged {
                    name: Arc::clone(entry.key()),
                    id,
                    payload: None,
                };

                self.sources.push(source
                    .map(move |payload| Tagged { name: Arc::clone(&tag), id, payload: Some(payload) })
                    .chain(stream::once(future::ready(end)))
                    .boxed());

                entry.insert(Published {
                    id,
                    abort,
                    obj_name,
                    topics: vec![(topic, version)],
                    counter: 1,
                    double: true,
                    confirmed: now,
                });

            }
        }

    }

    /// Publish what the streams of the devices have, up to [`MAX_ITEMS_AT_ONCE`].
    fn poll_sources(&mut self) {

        let waker = self.waker.clone();
        let mut cx = Context::from_waker(&waker);

        for _ in 0..MAX_ITEMS_AT_ONCE {
            match self.sources.poll_next_unpin(&mut cx) {
                Poll::Ready(Some(Tagged { name, id, payload: Some(payload) })) => self.publish(&name, id, payload),
                Poll::Ready(Some(Tagged { name, id, payload: None })) => self.end(&name, id),
                Poll::Ready(None) | Poll::Pending => return,
            }
        }

        // Some may be left, which the next round of the loop takes.
        waker.wake();

    }

    /// Publish an item of the stream of the device of that event, on every topic of it.
    fn publish(&mut self, name: &Arc<str>, id: u64, payload: Payload) {

        // An item of a stream that was dropped, still in flight.
        let Some(published) = self.events.get_mut(name).filter(|published| published.id == id) else {
            return;
        };

        let counter = published.counter;
        published.counter = published.counter.wrapping_add(1);
        let count = if mem::take(&mut published.double) { 2 } else { 1 };
        let endian = [(Endian::NATIVE == Endian::Little) as u8];

        for (topic, version) in &published.topics {

            self.payload_writer.clear();
            self.call_info_writer.clear();

            let encoded = write_payload(&mut self.payload_writer, &payload, &published.obj_name, *version)
                .and_then(|is_except| write_call_info(&mut self.call_info_writer, counter, is_except));

            if let Err(err) = encoded {
                error!(%topic, %err, "Failed to encode the event, not publishing it");
                continue;
            }

            // The first filler is dropped, see [`write_payload`].
            let call_info = self.call_info_writer.buffer();
            let data = &self.payload_writer.buffer()[4..];

            for _ in 0..count {
                let sent = self.event.send(topic.as_bytes(), zmq::SNDMORE)
                    .and_then(|()| self.event.send(&endian[..], zmq::SNDMORE))
                    .and_then(|()| self.event.send(call_info, zmq::SNDMORE))
                    .and_then(|()| self.event.send(data, 0));
                if let Err(err) = sent {
                    warn!(%topic, %err, "Failed to publish the event, ignoring");
                }
            }

            trace!(%topic, counter, count, "Published the event");

        }

    }

    /// The stream of the device of that event ended, which is no longer published.
    fn end(&mut self, name: &Arc<str>, id: u64) {
        if let Entry::Occupied(entry) = self.events.entry(Arc::clone(name)) && entry.get().id == id {
            entry.remove();
            debug!(event = %name, "The device stopped publishing the event");
        }
    }

    /// Stop publishing the events no client confirmed for too long (§11).
    fn expire(&mut self) {
        self.events.retain(|name, published| {
            let alive = published.confirmed.elapsed() < RESUBSCRIBE_PERIOD;
            if !alive {
                published.abort.abort();
                debug!(event = %name, "No client confirmed the event for too long, no longer publishing it");
            }
            alive
        });
    }

    /// Publish the heartbeat, twice if a client subscribed since the last one.
    fn send_heartbeat(&mut self) -> Result<()> {

        let count = if std::mem::take(&mut self.double_heartbeat) { 2 } else { 1 };
        let endian = [(Endian::NATIVE == Endian::Little) as u8];

        self.call_info_writer.clear();
        write_call_info(&mut self.call_info_writer, 0, false)?;

        for _ in 0..count {
            self.heartbeat.send(self.heartbeat_topic.as_bytes(), zmq::SNDMORE)?;
            self.heartbeat.send(&endian[..], zmq::SNDMORE)?;
            self.heartbeat.send(self.call_info_writer.buffer(), 0)?;
        }

        self.event.send(KEEP_ALIVE_MESSAGE, 0)?;
        trace!(topic = %self.heartbeat_topic, count, "Published the heartbeat");

        Ok(())

    }

}

/// Bind that socket to that address, at the port the given variable sets if it does, giving
/// back the port it is bound to.
fn bind(socket: &zmq::Socket, addr: IpAddr, ports: (u16, u16)) -> Result<u16> {

    let host = match addr {
        addr if addr.is_unspecified() => "*".to_string(),
        IpAddr::V4(addr) => addr.to_string(),
        IpAddr::V6(addr) => format!("[{addr}]"),
    };

    if addr.is_ipv6() {
        socket.set_ipv6(true)?;
    }

    let mut endpoint = format!("tcp://{host}:*");
    let endpoint_prefix_len = endpoint.len() - 1;

    if ports.0 == 0 {
        socket.bind(&endpoint)?;
    } else {

        let res = super::super::try_port_range(ports, |port| {
            
            endpoint.truncate(endpoint_prefix_len);
            write!(endpoint, "{port}").unwrap();

            match socket.bind(&endpoint) {
                Ok(()) => ControlFlow::Break(Ok(())),
                Err(zmq::Error::EADDRINUSE) => ControlFlow::Continue(()),
                Err(err) => ControlFlow::Break(Err(Error::new(ErrorKind::Zmq)
                    .with_description(format!("Failed to bind a ZMQ socket to {endpoint}"))
                    .with_source(Box::new(err)))),
            }

        });

        match res {
            ControlFlow::Break(Ok(())) => (),
            ControlFlow::Break(Err(err)) => return Err(err),
            ControlFlow::Continue(()) => return Err(Error::new(ErrorKind::Zmq)
                .with_description(format!("Failed to bind a ZMQ socket to {addr}, every port being in use"))
                .with_append_description(format_args!("Tried: ports {}..={}", ports.0, ports.1))),
        }

    }

    // What ZMQ bound to, the port included when it picked it, as `tcp://host:port`.
    let bound = socket.get_last_endpoint()?
        .map_err(|_| Error::new(ErrorKind::Internal).with_description("The endpoint ZMQ bound to is not text"))?;

    bound.rsplit_once(':')
        .and_then(|(_, port)| port.parse().ok())
        .ok_or_else(|| Error::new(ErrorKind::Internal)
            .with_description(format!("Endpoint {bound:?} that ZMQ bound to has no port")))

}

/// The endpoint of a publisher at that address and port.
fn endpoint(addr: IpAddr, port: u16) -> Arc<str> {
    Arc::from(match addr {
        IpAddr::V4(addr) => format!("tcp://{addr}:{port}"),
        IpAddr::V6(addr) => format!("tcp://[{addr}]:{port}"),
    })
}

/// Write a `ZmqCallInfo`, which a heartbeat has with no counter (§8).
///
/// Its method name is always empty, and so is its object id: cppTango never sets them, and
/// its consumer fails to read a call info whose method name is not empty, rejecting the
/// event as malformed.
fn write_call_info(writer: &mut CdrWriter, counter: u32, is_except: bool) -> Result<()> {
    writer.write_long(CALL_INFO_VERSION)?;
    writer.write_ulong(counter)?;
    writer.write_string("")?;
    writer.write_sequence_len(0)?;
    writer.write_bool(is_except)
}

/// Write the payload of an event, in that version of its structure, or its error stack
/// when the device failed, giving back which of the two it is (§8).
///
/// The filler is written twice, as cppTango does, the first one to be dropped from what is
/// sent: what follows is then aligned as it would be with no filler at all.
fn write_payload(writer: &mut CdrWriter, payload: &Payload, obj_name: &str, version: DevVersion) -> Result<bool> {

    writer.write_ulong(PAYLOAD_FILLER)?;
    writer.write_ulong(PAYLOAD_FILLER)?;

    let is_except = match payload {
        Payload::Value(Err(err)) | 
        Payload::Config(Err(err)) | 
        Payload::Ready(Err(err)) => {
            writer.write(err)?;
            true
        }
        Payload::Value(read) => {
            write_attr(writer, version, obj_name, read)?;
            false
        }
        Payload::Config(Ok(info)) => {
            write_attr_info(writer, version, obj_name, info)?;
            false
        }
        Payload::Ready(Ok(ready)) => {
            write_attr_ready(writer, obj_name, ready)?;
            false
        }
    };

    Ok(is_except)

}

/// Subscribe to that event of an attribute of that device, by the name a client asks it
/// with, the one without the IDL compatibility prefix.
async fn subscribe_dev(dev: &dyn BackendDev, admin: Arc<dyn BackendDev>, obj_name: &str, event: &str) -> Result<BoxStream<'static, Payload>> {

    let event = match event {
        "change" => AttrValueEvent::Change,
        "periodic" => AttrValueEvent::Periodic,
        "archive" => AttrValueEvent::Archive,
        "user_event" => AttrValueEvent::User,
        "alarm" => AttrValueEvent::Alarm,
        "attr_conf" => return Ok(dev.subscribe_attr_config(admin, Scalar::from(obj_name)).await?.map(Payload::Config).boxed()),
        "data_ready" => return Ok(dev.subscribe_attr_ready(admin, Scalar::from(obj_name)).await?.map(Payload::Ready).boxed()),
        _ => return Err(Error::not_implemented()
            .with_description(format!("Publishing event {event:?} is not implemented"))),
    };

    Ok(dev.subscribe_attr_value(admin, Scalar::from(obj_name), event).await?.map(Payload::Value).boxed())

}

/// The error of a request the supplier cannot answer anymore, the server having stopped.
fn shutting_down_err() -> Error {
    Error::new(ErrorKind::Reason(reason::API_ShutdownInProgress))
        .with_description("The device server is shutting down and accepts no more event subscriptions")
}

/// The error of a confirmation that named those events, which are not published: no client
/// confirmed them for too long, or the device stopped publishing them, and a client only
/// gets them again by subscribing again.
fn not_published_err(names: &[Arc<str>]) -> Error {

    Error::new(ErrorKind::Reason(reason::API_EventNotFound))
        .with_description("Events are not published and must be subscribed to again")
        .with_append_description(format_args!("Got: {names:?}"))

}

/// The release of the ZMQ library, in the encoding cppTango uses,
/// `major * 100 + minor * 10 + patch`.
fn zmq_version() -> i32 {
    let (major, minor, patch) = zmq::version();
    major * 100 + minor * 10 + patch
}

/// The event name without its IDL compatibility prefix, if it has one (§9).
fn strip_idl_prefix(event_name: &str) -> &str {
    event_name.strip_prefix(IDL_COMPAT_PREFIX).unwrap_or(event_name)
}

#[cfg(test)]
mod tests {

    use std::net::Ipv4Addr;
    use std::time::SystemTime;

    use futures::channel::mpsc;
    use futures::executor::block_on;
    use futures::future::{ready, BoxFuture};

    use crate::dev::{AttrConfig, AttrQuality, CommandInfo, DevInfo, DevSource, DevState, Named, NamedResult, RecordedResult};
    use crate::corba::orb::cdr::CdrReader;
    use crate::value::Scalar;
    use crate::buf::Buf;

    use super::super::{decode_value, CallInfo};
    use super::*;

    /// The call info of an event has an empty method name, as cppTango's has, whose consumer
    /// rejects one that does not.
    #[test]
    fn call_info() {

        let mut writer = CdrWriter::new();
        write_call_info(&mut writer, 42, true).unwrap();

        let buf = Buf::from(writer.buffer());
        let mut reader = CdrReader::new(&buf).with_endian(Endian::NATIVE);
        assert_eq!(reader.read_long().unwrap(), CALL_INFO_VERSION);
        assert_eq!(reader.read_ulong().unwrap(), 42);
        assert_eq!(reader.read_string().unwrap(), "");
        assert_eq!(reader.read_sequence_len().unwrap(), 0);
        assert!(reader.read_bool().unwrap());

    }

    /// A served device, also serving as the admin device, whose value events are what the
    /// test sends through the channel it is made with, a single subscription getting them,
    /// and whose data ready events never come. It has no other event.
    #[derive(Debug)]
    struct Dev {
        values: Mutex<Option<mpsc::UnboundedReceiver<Result<AttrRead<Value>>>>>,
    }

    impl Dev {
        fn new() -> (Arc<Self>, mpsc::UnboundedSender<Result<AttrRead<Value>>>) {
            let (values_tx, values_rx) = mpsc::unbounded();
            (Arc::new(Self { values: Mutex::new(Some(values_rx)) }), values_tx)
        }
    }

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

    impl BackendDev for Dev {
        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, _: Scalar<str>, _: Value, _: DevSource) -> BoxFuture<'_, Result<Value>> { not_implemented() }
        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>>>>> {
            let values = self.values.lock().unwrap().take();
            Box::pin(ready(Ok(match values {
                Some(values) => values.boxed(),
                None => stream::pending().boxed(),
            })))
        }
        fn subscribe_attr_ready(&self, _: Arc<dyn BackendDev>, _: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrReady>>>> {
            Box::pin(ready(Ok(stream::pending().boxed())))
        }
        fn subscribe_attr_config(&self, _: Arc<dyn BackendDev>, _: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrInfo>>>> { not_implemented() }
    }

    fn supplier(tango_host: Option<(&str, u16)>) -> Supplier {
        Supplier::new(
            "dserver/exe/inst",
            tango_host,
            "10.0.0.1",
            1234,
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            (0, 0),
            (0, 0),
            &[IpAddr::V4(Ipv4Addr::LOCALHOST)],
        )
    }

    /// The devices of the server, `sys/tg_test/1` and the admin device, both that device.
    fn served(dev: &Arc<Dev>) -> impl Fn(&str) -> Option<Arc<dyn BackendDev>> {
        move |name| matches!(name, "sys/tg_test/1" | "dserver/exe/inst")
            .then(|| Arc::clone(dev) as Arc<dyn BackendDev>)
    }

    fn args(args: &[&str]) -> Spectrum<str> {
        Spectrum::from(args)
    }

    /// Receive an event on that topic, giving its counter, whether it is an error, and its
    /// payload without the filler.
    fn recv_event(sub: &zmq::Socket, topic: &str) -> (u32, bool, Buf<u8, 1>) {

        let parts = sub.recv_multipart(0).unwrap();
        assert_eq!(parts.len(), 4);
        assert_eq!(parts[0], topic.as_bytes());
        assert_eq!(parts[1], [(Endian::NATIVE == Endian::Little) as u8]);

        let call = CallInfo::read(&parts[2], Endian::NATIVE).unwrap();
        assert_eq!(&parts[3][..4], PAYLOAD_FILLER.to_ne_bytes());

        (call.counter, call.is_except, Buf::from(&parts[3][4..]))

    }

    #[test]
    fn idl_prefix() {
        assert_eq!(strip_idl_prefix("idl5_change"), "change");
        assert_eq!(strip_idl_prefix("change"), "change");
    }

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

        let (dev, _values) = Dev::new();
        let supplier = supplier(Some(("DB.example.org", 10000)));
        let (numbers, strings) = block_on(supplier.subscription_change(
            &args(&["sys/TG_test/1", "Double_Scalar", "subscribe", "change", "6"]),
            served(&dev),
            6,
        )).unwrap();

        assert_eq!(numbers.as_slice()[..5], [LIB_VERSION, 6, SUB_HWM, 0, 0]);

        let strings = strings.iter().collect::<Vec<_>>();
        assert_eq!(strings.len(), 4);
        assert!(strings[0].starts_with("tcp://127.0.0.1:"));
        assert!(strings[1].starts_with("tcp://127.0.0.1:"));
        assert_eq!(strings[2], "tango://db.example.org:10000/sys/tg_test/1/double_scalar.idl5_change");
        assert_eq!(strings[3], "tango://db.example.org:10000/dserver/exe/inst");

        supplier.stop();

    }

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

        let (dev, _values) = Dev::new();
        let supplier = supplier(None);
        let (_, strings) = block_on(supplier.subscription_change(
            &args(&["sys/tg_test/1", "double_scalar", "subscribe", "data_ready", "5"]),
            served(&dev),
            6,
        )).unwrap();

        let strings = strings.iter().collect::<Vec<_>>();
        assert_eq!(strings[2], "tango://10.0.0.1:1234/sys/tg_test/1/double_scalar#dbase=no.data_ready");
        assert_eq!(strings[3], "tango://10.0.0.1:1234/dserver/exe/inst");

        supplier.stop();

    }

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

        let (dev, _values) = Dev::new();
        let supplier = supplier(Some(("db", 10000)));
        let (_, strings) = block_on(supplier.subscription_change(
            &args(&["sys/tg_test/1", "double_scalar", "subscribe", "change"]),
            served(&dev),
            6,
        )).unwrap();

        assert_eq!(strings.get([2]).unwrap(), "tango://db:10000/sys/tg_test/1/double_scalar.change");

        supplier.stop();

    }

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

        let (dev, _values) = Dev::new();
        let supplier = supplier(Some(("db", 10000)));
        let change = |input: &[&str]| block_on(supplier.subscription_change(&args(input), served(&dev), 6)).unwrap_err();

        assert_eq!(change(&["a", "b"]).kind(), ErrorKind::Reason(reason::API_WrongNumberOfArgs));
        assert_eq!(change(&["sys/tg_test/1", "d", "subscribe", "nope"]).kind(), ErrorKind::Reason(reason::API_WrongNumberOfArgs));
        assert_eq!(change(&["a/b/c", "d", "subscribe", "change"]).kind(), ErrorKind::Reason(reason::API_DeviceNotFound));

        // What the device refuses, or has no way to publish.
        assert_eq!(change(&["sys/tg_test/1", "d", "subscribe", "attr_conf"]).kind(), ErrorKind::NotImplemented);
        assert_eq!(change(&["sys/tg_test/1", "d", "subscribe", "intr_change"]).kind(), ErrorKind::NotImplemented);

        supplier.stop();

        assert_eq!(change(&["sys/tg_test/1", "d", "subscribe", "change"]).kind(), ErrorKind::Reason(reason::API_ShutdownInProgress));

    }

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

        let (dev, _values) = Dev::new();
        let supplier = supplier(Some(("db", 10000)));

        let (_, strings) = block_on(supplier.subscription_change(&args(&["info"]), served(&dev), 6)).unwrap();
        assert_eq!(strings.get([0]).unwrap(), "No ZMQ event yet!");

        block_on(supplier.subscription_change(&args(&["sys/tg_test/1", "d", "subscribe", "change", "6"]), served(&dev), 6)).unwrap();

        let (_, strings) = block_on(supplier.subscription_change(&args(&["INFO"]), served(&dev), 6)).unwrap();
        assert!(strings.get([0]).unwrap().starts_with("Heartbeat: tcp://127.0.0.1:"));

        supplier.stop();

    }

    /// Only the events published are confirmed, the others failing the confirmation, so
    /// that a client knows to subscribe to them again.
    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn confirm() {

        let (dev, _values) = Dev::new();
        let supplier = supplier(Some(("db", 10000)));
        let confirm = |input: &[&str]| block_on(supplier.confirm_subscription(&args(input), |name| name == "sys/tg_test/1"));

        // Nothing is published before a subscription.
        assert_eq!(confirm(&["sys/tg_test/1", "d", "idl5_change"]).unwrap_err().kind(), ErrorKind::Reason(reason::API_EventNotFound));

        block_on(supplier.subscription_change(&args(&["sys/tg_test/1", "d", "subscribe", "change", "6"]), served(&dev), 6)).unwrap();

        confirm(&["sys/tg_test/1", "d", "idl5_change"]).unwrap();
        confirm(&["SYS/tg_test/1", "D", "change"]).unwrap();
        confirm(&[]).unwrap();

        assert_eq!(confirm(&["sys/tg_test/1", "d", "idl5_change", "sys/tg_test/1", "e", "idl5_change"]).unwrap_err().kind(), ErrorKind::Reason(reason::API_EventNotFound));
        assert_eq!(confirm(&["a/b/c", "d", "idl5_change"]).unwrap_err().kind(), ErrorKind::Reason(reason::API_DeviceNotFound));

        supplier.stop();

        assert_eq!(confirm(&["sys/tg_test/1", "d", "idl5_change"]).unwrap_err().kind(), ErrorKind::Reason(reason::API_ShutdownInProgress));

    }

    /// What the device gives is published, the first event twice, and a failure as the
    /// error stack.
    #[test]
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn publish() {

        let (dev, values) = Dev::new();
        let supplier = supplier(Some(("db", 10000)));
        let (_, strings) = block_on(supplier.subscription_change(
            &args(&["sys/tg_test/1", "double_scalar", "subscribe", "change", "6"]),
            served(&dev),
            6,
        )).unwrap();

        let strings = strings.iter().collect::<Vec<_>>();
        let topic = strings[2];

        let context = zmq::Context::new();
        let sub = context.socket(zmq::SUB).unwrap();
        sub.set_rcvtimeo(5000).unwrap();
        sub.connect(strings[1]).unwrap();
        sub.set_subscribe(topic.as_bytes()).unwrap();

        // The subscription has to reach the publisher before anything is published (§5.1).
        thread::sleep(Duration::from_millis(200));

        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),
        };

        values.unbounded_send(Ok(sent.clone())).unwrap();

        for _ in 0..2 {
            let (counter, is_except, payload) = recv_event(&sub, topic);
            assert_eq!((counter, is_except), (1, false));
            let mut reader = CdrReader::new(&payload).with_endian(Endian::NATIVE);
            assert_eq!(decode_value(&mut reader, DevVersion::V5).unwrap(), sent);
        }

        values.unbounded_send(Err(Error::new(ErrorKind::Reason(reason::API_AttrNotAllowed)))).unwrap();

        let (counter, is_except, payload) = recv_event(&sub, topic);
        assert_eq!((counter, is_except), (2, true));
        let err = CdrReader::new(&payload).with_endian(Endian::NATIVE).read::<Error>().unwrap();
        assert_eq!(err.kind(), ErrorKind::Reason(reason::API_AttrNotAllowed));

        supplier.stop();

    }

}