rstango 0.4.0

//! The ZMQ event system, which every Tango device server of release 8 or later publishes
//! its events with, from the point of view of a client subscribing to them, see
//! [`Consumer`], and of a server publishing them, see [`Supplier`].
//!
//! This is part of the CORBA backend and not a backend of its own: subscribing goes through
//! the admin device of the server, which is a CORBA request, and the events carry the very
//! same IDL structures the requests do, decoded by the same CDR layer.
//!
//! The protocol is the one `ZMQ.md` specifies, at the root of the repository, which the
//! `§` references in this module point into.
//!
//! What this module does not implement of it:
//! - multicast endpoints (§7), which are deprecated, the TCP ones a server offers along
//!   being used instead;
//! - servers older than Tango 9.3, which leave the client to derive the topics (§3, §4);
//! - the topic variants of a control system of several Tango hosts (§6), this client
//!   subscribing the prefix the server sent and never asking `DbGetCSDbServerList`;
//! - the active renegotiation of a subscription whose heartbeat stopped (§11), the loss
//!   being reported and the subscription only coming back if the server does so on the
//!   same endpoints.

mod convert;
mod consumer;
mod supplier;

use std::future::Future;
use std::time::Duration;
use std::sync::Arc;

use futures::future::BoxFuture;
use futures::stream::BoxStream;
use tracing::debug;

use crate::dev::{AttrInfo, AttrRead, AttrReady, AttrValueEvent, DevSource};
use crate::error::{Error, ErrorKind, Result, reason};
use crate::value::{Scalar, Value};
use crate::backend::BackendDev;
use crate::buf::Buf;

pub use consumer::Consumer;
pub use supplier::Supplier;
use consumer::{decode_config, decode_ready, decode_value, Decode, SubscribeRequest};
use super::orb::cdr::CdrReader;
use super::orb::endian::Endian;
use super::DevVersion;


/// The IDL release this client advertises when subscribing, the release the events are
/// published in being the lowest of this one and the device's (§9).
const IDL_VERSION: DevVersion = DevVersion::V6;

/// The suffix appended to the channel name to make the topic the heartbeat of a device
/// server is published on (§3).
const HEARTBEAT_SUFFIX: &str = ".heartbeat";
/// The prefix of the name of an event that also exists in an earlier form, from IDL v5 on
/// (§9).
const IDL_COMPAT_PREFIX: &str = "idl5_";
/// The marker of a device reached without a database, in the topics of its events and of
/// its heartbeat (§3).
const NO_DATABASE_MARKER: &str = "#dbase=no";
/// The scheme of the multicast endpoints, which are not supported (§7).
const MULTICAST_SCHEME: &str = "epgm://";

/// The events a client may ask for, by the name it asks for them with (§3).
const EVENT_NAMES: [&str; 9] = [
    "change",
    "periodic",
    "archive",
    "user_event",
    "attr_conf",
    "data_ready",
    "intr_change",
    "pipe",
    "alarm",
];
/// The events that have no earlier form than IDL v5, and therefore no IDL compatibility
/// prefix (§9).
const UNPREFIXED_EVENT_NAMES: [&str; 4] = ["alarm", "data_ready", "intr_change", "pipe"];

/// The release of `ZmqCallInfo` the messages are encoded in (§8).
const CALL_INFO_VERSION: i32 = 1;
/// The filler the payload of an event starts with (§8).
const PAYLOAD_FILLER: u32 = 0xDEC0DEC0;

/// The first Tango library release that sends the topics in the negotiation reply, older
/// ones leaving them to be derived (§4).
const MIN_LIB_VERSION: u32 = 930;

/// The interval a device server publishes its heartbeat at, after which a missing one means
/// its event system is no longer reachable (§11).
const HEARTBEAT_PERIOD: Duration = Duration::from_secs(10);
/// The interval after which a server drops a subscription that was not confirmed (§11).
const RESUBSCRIBE_PERIOD: Duration = Duration::from_secs(600);
/// The interval the subscriptions are confirmed at, a third of the one the server drops
/// them after, so that a lost confirmation is harmless (§11).
const CONFIRM_PERIOD: Duration = Duration::from_secs(RESUBSCRIBE_PERIOD.as_secs() / 3);

/// The time a subscription that was just declared is given to reach the publisher before
/// it is handed out (§5.1). Nothing tells when it has, so this is a guess, the one cppTango
/// makes on Windows, its larger one.
const SUBSCRIBE_PROPAGATION: Duration = Duration::from_millis(20);

/// Linger of the sockets, the same as cppTango's: a socket closed with messages still to
/// send drops them instead of holding the context.
const DEFAULT_LINGER: i32 = 0;
/// Send high water mark of the subscriber sockets, the same as cppTango's, which bounds the
/// subscriptions waiting to go out, the only thing a subscriber sends.
const SUB_SEND_HWM: i32 = 10_000;
/// The receive high water mark a server tells its clients to set on their event socket,
/// cppTango's default (§4).
const SUB_HWM: i32 = 1000;
/// Send high water mark of the event publisher socket, cppTango's default.
const PUB_HWM: i32 = 1000;

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

/// Subscribe to the readings of an attribute, see [`BackendDev::subscribe_attr_value`].
///
/// The name of the device is the request asking the device for it, the name the server
/// knows the device as, which the negotiation names (§4).
pub fn subscribe_attr_value<'a>(
    consumer: &'a Consumer,
    dev_name: BoxFuture<'a, Result<Arc<str>>>,
    admin: Arc<dyn BackendDev>,
    name: Scalar<str>,
    event: AttrValueEvent,
) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
    Box::pin(async move { subscribe(consumer, dev_name, admin, name.value(), EventKind::Value(event), decode_value).await })
}

/// Subscribe to an attribute being ready, see [`BackendDev::subscribe_attr_ready`] and
/// [`subscribe_attr_value`].
pub fn subscribe_attr_ready<'a>(
    consumer: &'a Consumer,
    dev_name: BoxFuture<'a, Result<Arc<str>>>,
    admin: Arc<dyn BackendDev>,
    name: Scalar<str>,
) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrReady>>>> {
    Box::pin(async move { subscribe(consumer, dev_name, admin, name.value(), EventKind::Ready, decode_ready).await })
}

/// Subscribe to the configuration of an attribute, see [`BackendDev::subscribe_attr_config`]
/// and [`subscribe_attr_value`].
pub fn subscribe_attr_config<'a>(
    consumer: &'a Consumer,
    dev_name: BoxFuture<'a, Result<Arc<str>>>,
    admin: Arc<dyn BackendDev>,
    name: Scalar<str>,
) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> {
    Box::pin(async move { subscribe(consumer, dev_name, admin, name.value(), EventKind::Config, decode_config).await })
}

/// Ask the server to publish the event, then have the consumer subscribe to it.
async fn subscribe<T: Send + 'static>(
    consumer: &Consumer,
    dev_name: impl Future<Output = Result<Arc<str>>>,
    admin: Arc<dyn BackendDev>,
    name: &str,
    kind: EventKind,
    decode: Decode<T>,
) -> Result<BoxStream<'static, Result<T>>> {

    let dev_name = dev_name.await?;
    let obj_name = Arc::<str>::from(name.to_lowercase());

    // The plain name of the event goes out, the server adding the IDL compatibility prefix
    // itself, and this client advertises the highest release it knows (§4, §9). This is
    // asked for every subscription, including one to an event already subscribed: it is
    // what makes the server publish it at all.
    let idl_version = IDL_VERSION.number().to_string();
    let input = Value::from([&*dev_name, &*obj_name, "subscribe", kind.name(), &*idl_version]);

    debug!(dev = %dev_name, attr = %obj_name, event = kind.name(), "Negotiating the event subscription");

    let reply = admin.command(Scalar::<str>::from("ZmqEventSubscriptionChange"), input, DevSource::Dev).await;
    let info = match reply {
        Ok(reply) => SubscriptionInfo::from_value(reply)?,
        // A device of IDL v3 or older publishes no event this way (§4), so another way of
        // subscribing gets the chance to.
        Err(err) if matches!(err.kind(), ErrorKind::CommandNotFound | ErrorKind::Reason(reason::API_CommandNotFound)) =>
            return Err(Error::new(ErrorKind::NotHandled)
                .with_description(format!("Server of device {dev_name:?} does not publish events over ZMQ"))
                .with_parent(err)),
        Err(err) => return Err(err),
    };

    debug!(dev = %dev_name, attr = %obj_name, ?info, "Negotiated the event subscription");
    let request = SubscribeRequest { admin, dev_name, obj_name, kind, info };
    let sub = consumer.subscribe(request, decode)?.await?;
    Ok(Box::pin(sub))

}

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

/// The kinds of event that can be subscribed to, which decide both the name the event is
/// known as and the payload it carries (§8).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventKind {
    /// An attribute value event, of one of the kinds a value is pushed for.
    Value(AttrValueEvent),
    /// An attribute data ready event.
    Ready,
    /// An attribute configuration event.
    Config,
}

impl EventKind {

    /// The name of this event, as the negotiation sends it (§4).
    fn name(self) -> &'static str {
        match self {
            Self::Value(AttrValueEvent::Change) => "change",
            Self::Value(AttrValueEvent::Periodic) => "periodic",
            Self::Value(AttrValueEvent::Archive) => "archive",
            Self::Value(AttrValueEvent::User) => "user_event",
            Self::Value(AttrValueEvent::Alarm) => "alarm",
            Self::Ready => "data_ready",
            Self::Config => "attr_conf",
        }
    }

    /// The name of this event as the topic carries it and as the confirmation sends it,
    /// which has the IDL compatibility prefix from IDL v5 on, except for the events that
    /// have no earlier form (§9).
    fn topic_name(self, version: DevVersion) -> Arc<str> {
        let compat = !matches!(self, Self::Value(AttrValueEvent::Alarm) | Self::Ready);
        if compat && version >= DevVersion::V5 {
            Arc::from(format!("{IDL_COMPAT_PREFIX}{}", self.name()))
        } else {
            Arc::from(self.name())
        }
    }

}

// =========================== //
//         Negotiation         //
// =========================== //

/// What a device server answers the negotiation with, the `DevVarLongStringArray` of its
/// `ZmqEventSubscriptionChange` command (§4).
#[derive(Debug)]
pub struct SubscriptionInfo {
    /// The release the events are published in, which is the lowest of the device's and
    /// this client's (§9).
    version: DevVersion,
    /// The receive high water mark to set on the event socket, absent when the server is
    /// too old to send one, `1000` by default in cppTango.
    sub_hwm: Option<i32>,
    /// The endpoints the server publishes on, one pair per network interface of its host,
    /// such as `tcp://10.0.0.1:41235` and `tcp://10.0.0.1:41236`.
    endpoints: Vec<Endpoints>,
    /// The topic the event is published on, such as
    /// `tango://db.example.org:10000/sys/tg_test/1/double_scalar.idl5_change`, or
    /// `tango://host:1234/sys/tg_test/1/double_scalar#dbase=no.idl5_change` for a device
    /// reached without a database (§3).
    event_topic: Arc<str>,
    /// The name of the channel of the device server, the base of its heartbeat topic, such
    /// as `tango://db.example.org:10000/dserver/tangotest/test`, or
    /// `tango://host:1234/dserver/tangotest/test#dbase=no` for a device reached without a
    /// database, the marker being added back here if the server left it out (§3).
    channel_name: Arc<str>,
}

/// The pair of endpoints of one network interface of the host of a device server (§5).
#[derive(Debug, Clone)]
struct Endpoints {
    /// Where the heartbeat is published, such as `tcp://10.0.0.1:41235`.
    heartbeat: Arc<str>,
    /// Where the events are published, such as `tcp://10.0.0.1:41236`.
    event: Arc<str>,
}

impl SubscriptionInfo {

    /// Decode the answer of the `ZmqEventSubscriptionChange` command.
    fn from_value(value: Value) -> Result<Self> {

        let Value::LongString(numbers, strings) = value else {
            return Err(Error::new(ErrorKind::InvalidData)
                .with_description("The device server did not answer the event negotiation with a DevVarLongStringArray"));
        };

        let [lib_version, dev_version, rest @ ..] = numbers.as_slice() else {
            return Err(Error::new(ErrorKind::InvalidData)
                .with_description("The device server sent neither its library release nor the IDL release of the device"));
        };

        if *lib_version < MIN_LIB_VERSION as i32 {
            return Err(Error::not_implemented()
                .with_description("Events of a device server older than Tango 9.3 are not implemented")
                .with_append_description(format_args!("Got: library release {lib_version}")));
        }

        // The device's release and not the negotiated one, which is applied here (§9).
        let version = u32::try_from(*dev_version).ok()
            .map(|number| number.min(IDL_VERSION.number()))
            .and_then(DevVersion::from_number)
            .ok_or_else(|| Error::new(ErrorKind::InvalidData)
                .with_description("The IDL release the device server sent is not a known one")
                .with_append_description(format_args!("Got: {dev_version}")))?;

        // The multicast parameters and the ZMQ release that follow are of no use here.
        let sub_hwm = rest.first().copied();

        // The topics are the last two strings, every one before them being an endpoint.
        let Some(endpoints_len) = strings.len().checked_sub(2) else {
            return Err(Error::new(ErrorKind::InvalidData)
                .with_description("The device server sent neither the event topic nor the channel name"));
        };

        if endpoints_len == 0 || endpoints_len % 2 != 0 {
            return Err(Error::new(ErrorKind::InvalidData)
                .with_description("The device server sent an odd or empty count of endpoints")
                .with_append_description(format_args!("Got: {endpoints_len}"))
                .with_append_description("Expected: pairs of a heartbeat and an event endpoint"));
        }

        // Within bounds, the length was just checked.
        let string = |index: usize| strings.get([index]).unwrap();

        let endpoints = (0..endpoints_len).step_by(2)
            .map(|index| Endpoints {
                heartbeat: Arc::from(string(index)),
                event: Arc::from(string(index + 1)),
            })
            .collect();

        let event_topic = string(endpoints_len);
        let channel_name = string(endpoints_len + 1);

        // The server builds the channel name from the admin device name, which lacks the
        // marker of a device reached without a database, while it publishes its heartbeat
        // on a topic that has it (§3). The event topic has it whenever that is the case.
        let channel_name = if event_topic.contains(NO_DATABASE_MARKER) && !channel_name.ends_with(NO_DATABASE_MARKER) {
            Arc::from(format!("{channel_name}{NO_DATABASE_MARKER}"))
        } else {
            Arc::from(channel_name)
        };

        Ok(Self {
            version,
            sub_hwm,
            endpoints,
            event_topic: Arc::from(event_topic),
            channel_name,
        })

    }

    /// The topic the heartbeat of the channel of this event is published on (§3).
    fn heartbeat_topic(&self) -> Arc<str> {
        Arc::from(format!("{}{HEARTBEAT_SUFFIX}", self.channel_name))
    }

}

impl Endpoints {

    /// Whether this pair is published over multicast (§7).
    fn is_multicast(&self) -> bool {
        self.heartbeat.starts_with(MULTICAST_SCHEME) || self.event.starts_with(MULTICAST_SCHEME)
    }

}

// =========================== //
//          Messages           //
// =========================== //

/// The `ZmqCallInfo` every message carries (§8).
#[derive(Debug)]
struct CallInfo {
    /// The counter of the event within its topic (§10).
    counter: u32,
    /// Whether the payload is the error stack of a failed reading instead of a value.
    is_except: bool,
}

impl CallInfo {

    fn read(octets: &[u8], endian: Endian) -> Result<Self> {

        let buf = Buf::from(octets);
        let mut reader = CdrReader::new(&buf).with_endian(endian);

        let _version = reader.read_long()?;
        let counter = reader.read_ulong()?;
        let _method_name = reader.read_string()?;

        let oid_len = reader.read_sequence_len()?;
        reader.skip(oid_len)?;

        let is_except = reader.read_bool()?;

        Ok(Self { counter, is_except })

    }

}

#[cfg(test)]
mod tests {

    use super::*;

    use crate::value::Spectrum;

    /// A reply the way a cppTango 9.3+ server sends it, for a device with a database, on a
    /// host with two network interfaces.
    fn reply(dev_version: i32, strings: &[&str]) -> Value {
        Value::LongString(
            Spectrum::from(vec![1000, dev_version, 1000, 0, 0, 40305]),
            Spectrum::<str>::from(strings),
        )
    }

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

    #[test]
    fn subscription_info() {

        let info = SubscriptionInfo::from_value(reply(6, &[
            "tcp://10.0.0.1:5001", "tcp://10.0.0.1:5002",
            "tcp://192.168.0.1:5001", "tcp://192.168.0.1:5002",
            TOPIC, CHANNEL,
        ])).unwrap();

        assert_eq!(info.version, DevVersion::V6);
        assert_eq!(info.sub_hwm, Some(1000));
        assert_eq!(info.endpoints.len(), 2);
        assert_eq!(&*info.endpoints[1].heartbeat, "tcp://192.168.0.1:5001");
        assert_eq!(&*info.endpoints[1].event, "tcp://192.168.0.1:5002");
        assert_eq!(&*info.event_topic, TOPIC);
        assert_eq!(&*info.channel_name, CHANNEL);
        assert_eq!(&*info.heartbeat_topic(), "tango://db.example.org:10000/dserver/tangotest/test.heartbeat");

    }

    #[test]
    fn subscription_info_no_database() {

        let info = SubscriptionInfo::from_value(reply(5, &[
            "tcp://10.0.0.1:5001", "tcp://10.0.0.1:5002",
            "tango://host:1234/sys/tg_test/1/double_scalar#dbase=no.idl5_change",
            "tango://host:1234/dserver/tangotest/test",
        ])).unwrap();

        assert_eq!(&*info.channel_name, "tango://host:1234/dserver/tangotest/test#dbase=no");

    }

    #[test]
    fn subscription_info_negotiated_version() {
        // A device of a release this client does not know yet publishes in this client's.
        let info = SubscriptionInfo::from_value(reply(7, &["tcp://a:1", "tcp://a:2", TOPIC, CHANNEL])).unwrap();
        assert_eq!(info.version, IDL_VERSION);
    }

    #[test]
    fn subscription_info_invalid() {

        // Too old to send the topics.
        let mut old = reply(4, &["tcp://a:1", "tcp://a:2"]);
        if let Value::LongString(numbers, _) = &mut old {
            *numbers = Spectrum::from(vec![920, 4]);
        }
        assert_eq!(SubscriptionInfo::from_value(old).unwrap_err().kind(), ErrorKind::NotImplemented);

        // Odd count of endpoints.
        let odd = reply(6, &["tcp://a:1", TOPIC, CHANNEL]);
        assert_eq!(SubscriptionInfo::from_value(odd).unwrap_err().kind(), ErrorKind::InvalidData);

        // Not even the topics.
        let empty = reply(6, &[]);
        assert_eq!(SubscriptionInfo::from_value(empty).unwrap_err().kind(), ErrorKind::InvalidData);

        assert_eq!(SubscriptionInfo::from_value(Value::Void).unwrap_err().kind(), ErrorKind::InvalidData);

    }

    #[test]
    fn topic_name() {
        let change = EventKind::Value(AttrValueEvent::Change);
        assert_eq!(&*change.topic_name(DevVersion::V4), "change");
        assert_eq!(&*change.topic_name(DevVersion::V5), "idl5_change");
        assert_eq!(&*EventKind::Config.topic_name(DevVersion::V6), "idl5_attr_conf");
        assert_eq!(&*EventKind::Ready.topic_name(DevVersion::V6), "data_ready");
        assert_eq!(&*EventKind::Value(AttrValueEvent::Alarm).topic_name(DevVersion::V6), "alarm");
    }

    #[test]
    fn call_info() {

        // version 1, counter 42, "push_zmq_event", empty oid, not an exception: big endian.
        let mut octets = Vec::new();
        octets.extend(1i32.to_be_bytes());
        octets.extend(42u32.to_be_bytes());
        octets.extend(15u32.to_be_bytes());
        octets.extend(b"push_zmq_event\0");
        octets.push(0); // alignment of the sequence length
        octets.extend(0u32.to_be_bytes());
        octets.push(0);

        let call = CallInfo::read(&octets, Endian::Big).unwrap();
        assert_eq!(call.counter, 42);
        assert!(!call.is_except);

    }

}