rstango 0.1.0

//! Code for the ZMQ event backend.
//! 
//! Note that ZMQ is directly depending on the omniORB backend, because it uses some 
//! custom types for CDR marshalling.

mod convert;

use std::borrow::Cow;
use std::collections::{HashMap, HashSet, VecDeque, hash_map};
use std::time::{Duration, Instant};
use std::fmt::{self, Debug};
use std::sync::{Arc, Weak};

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

use tokio::sync::{mpsc, oneshot, watch};
use tokio::time::sleep_until;

use rzmq::{Context, Msg, Socket, SocketType};
use rzmq::socket::options;

use tango_omniorb as orb;

use crate::omniorb::attr::{attr_read_from_value_v1, attr_read_from_value_v3, attr_read_from_value_v4, attr_read_from_value_v5};
use crate::backend::{self, AttrValueChannelBackend, AttrValueChannelGuard, AttrConfChannelBackend, AttrReadyChannelBackend, EventBackend};
use crate::dev::{AttrValueEvent, AttrRead, DevProxy};
use crate::error::{Error, ErrorKind, Result};
use crate::data::{Data, DataKind, DataLayoutKind, FromData};


/// This value is currently hardcoded into cppTango. After this duration, a client that
/// didn't confirm the subscription will be disconnected and will no longer receive 
/// events!
const MAX_CONFIRM_SUBSCRIPTION_INTERVAL: Duration = Duration::from_secs(600);
/// This is the confirmation interval that we actually use to sending confirmation of
/// event subscription, this allows safe margin with the 600 seconds that the server 
/// is checking.
const CONFIRM_SUBSCRIPTION_INTERVAL: Duration = Duration::from_secs(200);
/// This value is currently hardcoded into cppTango.
const CHECK_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10);


/// The ZeroMQ event backend.
/// 
/// Note that the ZMQ event backend relies on omniORB, so it will internally acquire 
/// a [`orb::CORBA_Orb`] instance by initializing omniORB, thus we are sure that we won't
/// have uninitialized things during execution.
pub struct ZmqEventBackend {
    /// Weak reference back to itself.
    this: Weak<Self>,
    /// The ORB instance that guarantees that the ORB context is initialized, thus allowing
    _orb: orb::CORBA_ORB,
    /// Sender of commands to the worker.
    command_tx: mpsc::UnboundedSender<Command>,
}

impl ZmqEventBackend {

    /// This function is async because it will use the current Tokio context.
    pub async fn init() -> Result<Arc<Self>> {

        let orb = orb::CORBA_ORB::init()?;
        let context = Context::new()?;
        let socket = context.socket(SocketType::Sub)?;

        let (
            command_tx,
            command_rx,
        ) = mpsc::unbounded_channel();

        let ret = Arc::new_cyclic(|this| {
            Self {
                this: this.clone(),
                _orb: orb,
                command_tx,
            }
        });

        let worker = Worker {
            backend: Arc::clone(&ret),
            command_rx,
            socket,
            socket_endpoints: HashMap::new(),
            next_id: 0,
            admins: HashMap::new(),
            channels: HashMap::new(),
            confirm_subscription_queue: ScheduleQueue::new(CONFIRM_SUBSCRIPTION_INTERVAL),
            check_heartbeat_queue: ScheduleQueue::new(CHECK_HEARTBEAT_INTERVAL),
        };

        let _handle = tokio::task::spawn(worker.main());

        Ok(ret)

    }

    /// Register this backend.
    #[inline]
    pub fn register(self: Arc<Self>) {
        backend::register_event_backend(self);
    }

    #[allow(unused)]
    #[inline]
    fn this(&self) -> Arc<Self> {
        // If this function is called, this means that the top-level caller have a 
        // Arc<Self> because we only ever create backends inside Arc<...>, so we can
        // safely unwrap the upgrade.
        self.this.upgrade().unwrap()
    }

}

#[async_trait]
impl EventBackend for ZmqEventBackend {

    async fn subscribe_attr_value(&self, dev: &DevProxy, name: &str, event: AttrValueEvent) -> Result<Box<dyn AttrValueChannelBackend>> {

        let dev = dev.clone();
        let name = name.to_lowercase();

        let (
            result_tx,
            result_rx,
        ) = oneshot::channel();

        self.command_tx.send(Command::SubscribeAttrValue { 
            dev, 
            event, 
            name, 
            result_tx,
        }).map_err(|_| Error::new(ErrorKind::Internal)
            .with_description("The event worker is unexpectedly no longer alive"))?;

        match result_rx.await {
            Err(_) => Err(Error::new(ErrorKind::Internal)
                .with_description("The event worker has unexpectedly dropped the result channel")),
            Ok(res) => res,
        }

    }

    async fn subscribe_attr_ready(&self, dev: &DevProxy, name: &str) -> Result<Box<dyn AttrReadyChannelBackend>> {
        
        let dev = dev.clone();
        let name = name.to_lowercase();

        let (
            result_tx,
            result_rx,
        ) = oneshot::channel();

        self.command_tx.send(Command::SubscribeAttrReady { 
            dev, 
            name, 
            result_tx,
        }).map_err(|_| Error::new(ErrorKind::Internal)
            .with_description("The event worker is unexpectedly no longer alive"))?;

        match result_rx.await {
            Err(_) => Err(Error::new(ErrorKind::Internal)
                .with_description("The event worker has unexpectedly dropped the result channel")),
            Ok(res) => res,
        }

    }
    
    async fn subscribe_attr_conf(&self, dev: &DevProxy, name: &str) -> Result<Box<dyn AttrConfChannelBackend>> {
        todo!()
    }

}

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

/// The worker object that is running the asynchronous polling of the ZMQ sockets. 
/// 
/// We are using a worked that do all the subscription and the polling, because it's 
/// actually bad to put the sockets inside mutexes, it would just not work, and we cant't
/// clone them, so everything is managed by this task.
struct Worker {
    /// The reference to the backend and its shared state.
    backend: Arc<ZmqEventBackend>,
    /// Receiver of commands to the worker.
    command_rx: mpsc::UnboundedReceiver<Command>,
    /// The subscription socket we are using for both admin and event endpoints.
    socket: Socket,
    /// All the endpoints the socket is already connected to, associated to the number of
    /// usage, when we now that it is no longer in use we can disconnect from it!.
    socket_endpoints: HashMap<Arc<str>, usize>,
    /// The ID counter used for assigning IDs to trackers.
    next_id: usize,
    /// Tracking of admin channels and the heartbeat, the key of this map is the base 
    /// topic, not the heartbeat topic!
    admins: HashMap<Arc<str>, AdminTracker>,
    /// Channels for attribute events, mapped by their topic.
    channels: HashMap<Arc<str>, ChannelTracker>,
    /// The schedule queue for confirming event subscription.
    confirm_subscription_queue: ScheduleQueue<(Arc<str>, usize)>,
    /// The schedule queue for checking heartbeat interval.
    check_heartbeat_queue: ScheduleQueue<(Arc<str>, usize)>,
}

impl Worker {

    /// The main entrypoint of this worker.
    async fn main(mut self) {
        while self.poll().await { }
    }

    /// Single poll, return true if the loop should continue.
    async fn poll(&mut self) -> bool {

        tokio::select! {
            biased;
            res = self.command_rx.recv() => {
                // If the channel returns none, it means that it has been closed. 
                // Therefore we also tell the loop to stop, because it means that every
                // instance of the backend has been dropped.
                match res {
                    Some(cmd) => self.process_command(cmd).await,
                    None => return false,
                }
            }
            res = self.socket.recv_multipart() => {
                match res {
                    Ok(messages) => self.process_messages(messages).await,
                    Err(err) => {
                        warn!(?err, "Failed to receive multipart message from socket");
                        return true;
                    }
                }
            }
            (event_topic, id) = self.confirm_subscription_queue.wait() => {
                self.confirm_subscription(event_topic, id).await;
            }
            (admin_topic_base, id) = self.check_heartbeat_queue.wait() => {
                self.check_heartbeat(admin_topic_base, id).await;
            }
        };

        true

    }

    /// Process an internal control command.
    async fn process_command(&mut self, cmd: Command) {
        match cmd {
            Command::SubscribeAttrValue {
                dev,
                event,
                name,
                result_tx,
            } => {
                // Just ignore if the channel is closed, for example if the subscription
                // future has been dropped (cancelled).
                let _ = result_tx.send(self.subscribe_attr_value(dev, event, name).await);
            }
            Command::SubscribeAttrReady { 
                dev, 
                name, 
                result_tx,
            } => {
                let _ = result_tx.send(self.subscribe_attr_ready(dev, name).await);
            }
        }
    }

    /// Common function to subscribe to any event type.
    async fn subscribe_base(&mut self, dev: DevProxy, event_name: &str, name: String) -> Result<&mut ChannelTracker> {

        // Request subscription information
        let dev_name = dev.name().await?;
        let admin_dev = dev.admin_dev().await?;
        let _admin_name = admin_dev.name().await?;
        
        let info_params = [
            &dev_name[..],
            &name[..],
            "subscribe",
            event_name,
            orb::IDL_VERSION,
        ];

        debug!(?info_params, "Requesting subscription info");

        let mut info = match admin_dev.command::<_, SubscriptionInfo>("ZmqEventSubscriptionChange", info_params).await {
            Ok(ret) => ret,
            Err(err) if err.kind() == ErrorKind::CommandNotFound => 
                return Err(Error::new(ErrorKind::NotHandled)
                    .with_parent(err)),
            Err(err) => return Err(err),
        };

        debug!(?info, "Got subscription info");

        // Take the option so we can own the string and transfer it later...
        // Note that here the admin topic we get is the _base_ one, which don't have any
        // event type suffix like ".heartbeat", that explains the naming.
        let (admin_topic_base, event_topic) = match info.topics.take() {
            Some(topics) => (topics.admin, topics.event),
            None => {

                // Adapt code from ZmqEventConsumer::initialize_received_from_admin
                // which guess the event/admin topics depending on the lib version.

                // With legacy version we should guess the topic, depending on the device
                // TRL, this is not trivial for now.
                return Err(Error::new(ErrorKind::Internal)
                    .with_description("Legacy device are not yet supported for event subscription"));

            }
        };

        // Keep the IDL version of the server around.
        let idl_version = info.idl_version;
        debug!(?idl_version, ?admin_topic_base, ?event_topic, "Found topics from subscription info");

        // Now we can check if we already have or not the a channel for that topic, in 
        // case we already have a channel, we really want to check that the values we
        // got with the subscription info are coherent with the cached channel.
        let vacant = match self.channels.entry(event_topic) {
            hash_map::Entry::Occupied(o) => {

                let tracker = o.into_mut();

                if tracker.dev != dev {
                    error!(?tracker, "Existing channel tracker is incoherent with subscription info: invalid device");
                    return Err(Error::new(ErrorKind::Internal)
                        .with_description("Existing channel tracker is incoherent with subscription info: invalid device"));
                }

                if tracker.idl_version != idl_version {
                    error!(?tracker, "Existing channel tracker is incoherent with subscription info: invalid IDL version");
                    return Err(Error::new(ErrorKind::Internal)
                        .with_description("Existing channel tracker is incoherent with subscription info: invalid IDL version"));
                }

                if tracker.name != name {
                    error!(?tracker, "Existing channel tracker is incoherent with subscription info: invalid name");
                    return Err(Error::new(ErrorKind::Internal)
                        .with_description("Existing channel tracker is incoherent with subscription info: invalid name"));
                }

                return Ok(tracker);
                
            }
            hash_map::Entry::Vacant(v) => v,
        };

        // Just remake 'event_topic' from the vacant key...
        let event_topic = vacant.key();

        // Keep track if we connected to admin or event device in this function, so if 
        // anything goes wrong we can go back.
        let mut connected_to_admin = false;
        let mut connected_to_event = false;

        // Internal utility to cleanup
        async fn cleanup(socket: &Socket, admin: bool, event: bool, endpoint: &SubscriptionEndpoint) {
            if admin {
                if let Err(err) = socket.disconnect(&endpoint.admin).await {
                    warn!(?endpoint, ?err, "Failed to disconnect from admin endpoint, ignoring");
                }
            }
            if event {
                if let Err(err) = socket.disconnect(&endpoint.event).await {
                    warn!(?endpoint, ?err, "Failed to disconnect from event endpoint, ignoring");
                }
            }
        }

        // Find the first endpoint that we could connect to!
        let mut endpoints_it = info.endpoints.into_iter();
        let endpoint = loop {
            
            let Some(endpoint) = endpoints_it.next() else {
                return Err(Error::new(ErrorKind::EventNotConnected)
                    .with_append_description("No endpoint was available for connection"));
            };

            if endpoint.event.starts_with("epgm://")
            || endpoint.admin.starts_with("epgm://") {
                debug!(?endpoint, "Got an endpoint with multicast, skipping");
                continue;
            }

            if !self.socket_endpoints.contains_key(&endpoint.admin) {
                match self.socket.connect(&endpoint.admin).await {
                    Ok(()) => connected_to_admin = true,
                    Err(err) => {
                        warn!(?endpoint, ?err, "Cannot connect to admin endpoint, skipping");
                        continue;
                    }
                }
            }

            if !self.socket_endpoints.contains_key(&endpoint.event) {
                match self.socket.connect(&endpoint.event).await {
                    Ok(()) => connected_to_event = true,
                    Err(err) => {
                        warn!(?endpoint, ?err, "Cannot connect to event endpoint, skipping");
                        cleanup(&self.socket, connected_to_admin, false, &endpoint).await;
                        continue;
                    }
                }
            }

            debug!(?endpoint, "Found an endpoint where we could connect");
            break endpoint;
            
        };

        // Start by subscribing to the event topic.
        match self.socket.set_option_raw(options::SUBSCRIBE, event_topic.as_bytes()).await {
            Ok(()) => {}
            Err(err) => {
                error!(?err, "Failed to subscribe to the event topic, cleaning up");
                cleanup(&self.socket, connected_to_admin, connected_to_event, &endpoint).await;
                return Err(Error::new(ErrorKind::EventNotConnected)
                    .with_append_description("Failed to subscribe to the event topic")
                    .with_parent(err.into()));
            }
        }
        
        // Now we just register the last
        let admin_tracker = match self.admins.entry(admin_topic_base.clone()) {
            hash_map::Entry::Occupied(o) => o.into_mut(),
            hash_map::Entry::Vacant(v) => {

                // We subscribe to that admin's heartbeat!
                let heartbeat_topic = format!("{}.heartbeat", v.key());
                match self.socket.set_option_raw(options::SUBSCRIBE, heartbeat_topic.as_bytes()).await {
                    Ok(()) => {}
                    Err(err) => {

                        error!(?err, "Failed to subscribe to the heartbeat topic, cleaning up");
                        cleanup(&self.socket, connected_to_admin, connected_to_event, &endpoint).await;

                        // We subscribed for nothing just above, revert...
                        if let Err(err) = self.socket.set_option_raw(options::UNSUBSCRIBE, event_topic.as_bytes()).await {
                            warn!(?endpoint, ?err, "Failed to unsubscribe from event topic, ignoring");
                        }

                        return Err(Error::new(ErrorKind::EventNotConnected)
                            .with_append_description("Failed to subscribe to the heartbeat topic")
                            .with_parent(err.into()));

                    }
                }

                // Increase the reference count of the endpoint.
                *self.socket_endpoints.entry(Arc::clone(&endpoint.admin)).or_default() += 1;

                let id = self.next_id;
                self.next_id = self.next_id.wrapping_add(1);

                // We created a new admin tracker, so we schedule a check in the future.
                self.check_heartbeat_queue.schedule((Arc::clone(&admin_topic_base), id));

                // Create a new tracker for that admin device.
                v.insert(AdminTracker { 
                    id,
                    last_heartbeat: Instant::now(),
                    admin_endpoint: endpoint.admin,
                    event_topics: HashSet::new(),
                })

            }
        };
        
        // Once we can't have more error, definitely increase sub count of the event 
        // endpoint, and register the event topic.
        *self.socket_endpoints.entry(Arc::clone(&endpoint.event)).or_default() += 1;
        admin_tracker.event_topics.insert(Arc::clone(event_topic));

        let id = self.next_id;
        self.next_id = self.next_id.wrapping_add(1);

        // Push the event topic to the confirmation queue, note that we don't need to 
        // sort that array, since we insert in order and the order only relies on "now"
        // which can only increase.
        self.confirm_subscription_queue.schedule((Arc::clone(event_topic), id));

        Ok(vacant.insert(ChannelTracker {
            id,
            dev,
            idl_version,
            event_endpoint: endpoint.event,
            admin_topic_base,
            name,
            kind: ChannelTrackerKind::Missing,
        }))

    }

    #[instrument(skip(self))]
    async fn subscribe_attr_value(&mut self, dev: DevProxy, event: AttrValueEvent, name: String) -> Result<Box<dyn AttrValueChannelBackend>> {

        let event_name = match event {
            AttrValueEvent::Change => "change",
            AttrValueEvent::Periodic => "periodic",
            AttrValueEvent::Archive => "archive",
            AttrValueEvent::User => "user_event",
            AttrValueEvent::Alarm => "alarm",
        };

        let tracker = self.subscribe_base(dev, event_name, name).await?;

        if let ChannelTrackerKind::Missing = tracker.kind {

            let (
                channel_tx,
                channel_rx,
            ) = watch::channel(orb::CORBA_Any::new());

            tracker.kind = ChannelTrackerKind::AttrValue(AttrValueChannelTracker {
                event,
                tx: channel_tx,
            });

            return Ok(Box::new(ZmqAttrValueChannelBackend {
                idl_version: tracker.idl_version,
                rx: channel_rx,
            }));
            
        }

        let ChannelTrackerKind::AttrValue(kind_tracker) = &tracker.kind else {
            error!(?tracker, "Existing channel tracker is incoherent with subscription info: invalid channel kind");
            return Err(Error::new(ErrorKind::Internal)
                .with_description("Existing channel tracker is incoherent with subscription info: invalid channel kind"));
        };

        if kind_tracker.event != event {
            error!(?tracker, "Existing channel info is incoherent with subscription info: invalid event");
            return Err(Error::new(ErrorKind::Internal)
                .with_description("Existing channel info is incoherent with subscription info: invalid event"));
        }

        Ok(Box::new(ZmqAttrValueChannelBackend {
            idl_version: tracker.idl_version,
            rx: kind_tracker.tx.subscribe(),
        }))

    }
    
    #[instrument(skip_all)]
    async fn subscribe_attr_ready(&mut self, dev: DevProxy, name: String) -> Result<Box<dyn AttrReadyChannelBackend>> {
        
        let tracker = self.subscribe_base(dev, "data_ready", name).await?;

        if let ChannelTrackerKind::Missing = tracker.kind {

            let (
                channel_tx,
                channel_rx,
            ) = watch::channel(());

            tracker.kind = ChannelTrackerKind::AttrReady(AttrReadyChannelTracker {
                tx: channel_tx,
            });

            return Ok(Box::new(ZmqAttrReadyChannelBackend {
                rx: channel_rx,
            }));
            
        }

        let ChannelTrackerKind::AttrReady(kind_tracker) = &tracker.kind else {
            error!(?tracker, "Existing channel tracker is incoherent with subscription info: invalid channel kind");
            return Err(Error::new(ErrorKind::Internal)
                .with_description("Existing channel tracker is incoherent with subscription info: invalid channel kind"));
        };

        Ok(Box::new(ZmqAttrReadyChannelBackend {
            rx: kind_tracker.tx.subscribe(),
        }))

    }

    /// Unsubscribe from an event, thus removing the channel tracking information. You 
    /// should not call this if there the channel is actually still used by some backend.
    /// This function also assumes that the channel exists and will panic if not.
    async fn unsubscribe(&mut self, topic: &str) {

        // Internal function to unsubscribe the main socket from a topic which is known to
        // originate from a given endpoint, thus managing automatic disconnection if no
        // longer used!
        let mut unsubscribe_topic = async |topic: &str, endpoint: &str| {

            match self.socket.set_option_raw(options::UNSUBSCRIBE, topic.as_bytes()).await {
                Ok(()) => {}
                Err(err) => {
                    warn!(?err, "Failed to unsubscribe from topic, ignoring");
                }
            }

            if let Some(count) = self.socket_endpoints.get_mut(endpoint) {
                *count = count.saturating_sub(1);
                if *count == 0 {
                    debug!(?endpoint, "Endpoint no longer have active subscription, disconnecting");
                    let (endpoint, _) = self.socket_endpoints.remove_entry(endpoint).unwrap();
                    if let Err(err) = self.socket.disconnect(&endpoint).await {
                        warn!(?err, "Failed to disconnect from endpoint, ignoring");
                    }
                }
            } else {
                warn!(?endpoint, "Missing information about the endpoint");
            }
            
        };

        debug!(?topic, "Unsubscribing from event topic");
        let channel_tracker = self.channels.remove(topic).unwrap();
        unsubscribe_topic(topic, &channel_tracker.event_endpoint).await;

        let admin_topic_base = &channel_tracker.admin_topic_base;
        let Some(admin_tracker) = self.admins.get_mut(admin_topic_base) else {
            error!(?admin_topic_base, "The admin topic is not linked to any tracker, cannot fully unsubscribe");
            return;
        };
        
        debug!(?topic, ?admin_topic_base, "Removing event topic from admin tracker");
        if !admin_tracker.event_topics.remove(topic) {
            error!(?admin_topic_base, "The event topic is missing from the admin tracker, cannot fully unsubscribe");
            return;
        }

        // If we no longer have any event subscribed for that admin device, unsubscribe
        // the heartbeat topic from the socket and maybe disconnect the admin endpoint.
        if admin_tracker.event_topics.is_empty() {
            debug!(?topic, ?admin_topic_base, "Admin tracker is empty, cleaning it and unsubscribing from heartbeat topic");
            let admin_tracker = self.admins.remove(admin_topic_base).unwrap();
            let heartbeat_topic = format!("{}.heartbeat", channel_tracker.admin_topic_base);
            unsubscribe_topic(&heartbeat_topic, &admin_tracker.admin_endpoint).await;
        }

        debug!(?topic, "Unsubscribed from event");

    }
    
    /// Process an incoming multipart messages.
    #[instrument(skip_all)]
    async fn process_messages(&mut self, messages: Vec<Msg>) {
        
        for msg in &messages {
            if msg.data().is_none() {
                warn!("Ignoring message: no data in msg");
                return;
            }
        }

        let Some(([
            msg_topic, 
            msg_endian, 
            msg_info
        ], messages)) = messages.split_first_chunk::<3>() else {
            warn!("Ignoring message: wrong message count");
            return;
        };

        let Ok(topic) = str::from_utf8(msg_topic.data().unwrap()) else {
            warn!("Ignoring message: invalid topic");
            return;
        };

        if msg_endian.size() != 1 {
            warn!("Ignoring message: invalid endian");
            return;
        }
        let little_endian = msg_endian.data().unwrap()[0] != 0;

        let mut any = orb::CORBA_Any::new();
        match any.load_cdr(orb::Tango_ZmqCallInfo_tc(), little_endian, msg_info.data().unwrap()) {
            Ok(()) => (),
            Err(err) => {
                warn!(err = ?orb::CORBA_SystemException::downcast(err), "Ignoring message: failed to decode");
                return;
            }
        }

        let Some(call_info) = any.extract_borrow::<orb::Tango_ZmqCallInfo>() else {
            warn!("Ignoring message: failed to extract call info");
            return;
        };

        trace!(?topic, ?little_endian, ?call_info, "Message from socket");

        if let Some(topic_base) = topic.strip_suffix(".heartbeat") {

            // This is a heartbeat topic.
            self.process_heartbeat(topic_base).await;

        } else {
            
            // This should be an event topic.
            let [msg_data] = messages else {
                warn!("Ignoring message: wrong message count");
                return;
            };

            self.process_event(topic, little_endian, msg_data.data().unwrap()).await;

        }

    }

    async fn process_heartbeat(&mut self, topic_base: &str) {

        let Some(tracker) = self.admins.get_mut(topic_base) else {
            warn!(?topic_base, "Ignoring message: no admin tracker for topic");
            return;
        };

        let now = Instant::now();
        let interval = now - tracker.last_heartbeat;
        trace!(?interval, "Received heartbeat");
        tracker.last_heartbeat = Instant::now();

    }

    async fn process_event(&mut self, topic: &str, little_endian: bool, msg_data: &[u8]) {

        let Some(tracker) = self.channels.get(topic) else {
            warn!(?topic, "Ignoring message: no channel tracker for topic");
            return;
        };

        // When receiving an event, check in advance if we will be able to send on the
        // channel; if not we directly unsubscribe internally.
        let unsubscribe = match &tracker.kind {
            ChannelTrackerKind::Missing => {
                error!(?topic, "Invalid channel state, unsubscribing");
                true
            }
            ChannelTrackerKind::AttrValue(kind_tracker) => kind_tracker.tx.is_closed(),
            ChannelTrackerKind::AttrReady(kind_tracker) => kind_tracker.tx.is_closed(),
        };

        if unsubscribe {
            debug!(?topic, "No consumers while receiving event, unsubscribing");
            self.unsubscribe(&topic).await;
            return;
        }

        if msg_data.len() < 4 {
            warn!(?msg_data, "Ignoring message: missing 4 byte padding at the beginning");
            return;
        }

        let msg_data = &msg_data[4..];

        let unsubscribe = match &tracker.kind {
            ChannelTrackerKind::Missing => unreachable!(),
            ChannelTrackerKind::AttrValue(kind_tracker) => {

                // NOTE: The cppTango implementation is using a custom type 
                // "ZmqAttributeValue_5" which reduces the copy of bytes, here we 
                // just use omniORB stuff as a simplification! See later.
                let tc = match tracker.idl_version {
                    IdlVersion::V1 |
                    IdlVersion::V2 => orb::Tango_AttributeValue_tc(),
                    IdlVersion::V3 => orb::Tango_AttributeValue_3_tc(),
                    IdlVersion::V4 => orb::Tango_AttributeValue_4_tc(),
                    IdlVersion::V5 |
                    IdlVersion::V6 => orb::Tango_AttributeValue_5_tc(),
                };

                let mut any = orb::CORBA_Any::new();
                match any.load_cdr(tc, little_endian, msg_data) {
                    Ok(()) => (),
                    Err(err) => {
                        warn!(err = ?orb::CORBA_SystemException::downcast(err), "Ignoring message: failed to decode attr value");
                        return;
                    }
                }

                kind_tracker.tx.send(any).is_err()

            }
            ChannelTrackerKind::AttrReady(kind_tracker) => {

                let mut any = orb::CORBA_Any::new();
                match any.load_cdr(orb::Tango_AttDataReady_tc(), little_endian, msg_data) {
                    Ok(()) => (),
                    Err(err) => {
                        warn!(err = ?orb::CORBA_SystemException::downcast(err), "Ignoring message: failed to decode attr ready");
                        return;
                    }
                }

                kind_tracker.tx.send(()).is_err()

            }
        };

        // If we add an error while sending our data to a receiver, it means that the 
        // channel has been closed in the meantime, so we can unsubscribe!
        if unsubscribe {
            debug!(?topic, "No consumers for the parsed data, unsubscribing");
            self.unsubscribe(&topic).await;
        }

    }

    /// Process an event topic confirmation.
    async fn confirm_subscription(&mut self, event_topic: Arc<str>, id: usize) {

        let Some(tracker) = self.channels.get_mut(&event_topic) else {
            debug!(?event_topic, "No more channel tracker for confirming event topic, skipping");
            return;
        };

        if tracker.id != id {
            debug!(?event_topic, "Incoherent channel tracker id, skipping");
            return;
        }

        debug!(?event_topic, "Processing confirmation of event topic");

        // Before confirming, we check that this channel is actually still used
        let unsubscribe = match &tracker.kind {
            ChannelTrackerKind::Missing => {
                error!(?event_topic, "Invalid channel state, unsubscribing");
                true
            }
            ChannelTrackerKind::AttrValue(kind_tracker) => kind_tracker.tx.is_closed(),
            ChannelTrackerKind::AttrReady(kind_tracker) => kind_tracker.tx.is_closed(),
        };

        if unsubscribe {
            debug!(?event_topic, "No consumers while confirming subscription, unsubscribing");
            self.unsubscribe(&event_topic).await;
            return;
        }

        // let admin_device = match tracker.dev.admin_dev().await {
        //     Ok(dev) => dev,
        //     Err(err) => {
        //         todo!();
        //     }
        // };

        // let confirm_params = [
        //     &dev_name[..],
        //     &name[..],
        //     "subscribe",
        //     event_name,
        //     orb::IDL_VERSION,
        // ];
        
        // tracker.dev.command::<_, >("EventConfirmSubscription", confirm_params)


        // TODO: Don't forget to re-add the checking!

    }

    /// Check heartbeat for the given admin tracker.
    async fn check_heartbeat(&mut self, admin_topic_base: Arc<str>, id: usize) {
        
        let Some(tracker) = self.admins.get_mut(&admin_topic_base) else {
            debug!(?admin_topic_base, "No more admin tracker for check heartbeat, skipping");
            return;
        };

        if tracker.id != id {
            debug!(?admin_topic_base, "Incoherent admins tracker id, skipping");
            return;
        }

        if tracker.last_heartbeat.elapsed() <= CHECK_HEARTBEAT_INTERVAL {
            trace!(?admin_topic_base, "Last heartbeat is in interval");
            self.check_heartbeat_queue.schedule((admin_topic_base, id));
            return;
        }

        warn!(?admin_topic_base, "No heartbeat for more than {CHECK_HEARTBEAT_INTERVAL:?}");

    }

}

/// A queue for scheduling actions at fixed duration in the future.
#[derive(Debug)]
struct ScheduleQueue<T> {
    inner: VecDeque<(Instant, T)>,
    duration: Duration,
}

impl<T> ScheduleQueue<T> {

    pub fn new(duration: Duration) -> Self {
        Self {
            inner: VecDeque::new(),
            duration,
        }
    }

    pub fn schedule(&mut self, item: T) {
        self.inner.push_back((Instant::now() + self.duration, item));
    }

    /// Wait for the next ready item, if no item is available this will wait indefinitely
    /// and the task is expected to be cancelled at some point.
    pub async fn wait(&mut self) -> T {

        let Some(&(next_time, _)) = self.inner.front() else {
            std::future::pending::<()>().await;
            unreachable!();
        };

        sleep_until(next_time.into()).await;
        
        // Only pop after sleep is complete, so we never pop if the task is cancelled.
        let (_, item) = self.inner.pop_front().unwrap();
        item

    }

}

/// The command that go to the worker.
#[derive(Debug)]
enum Command {
    /// Subscribe to an attribute value event.
    SubscribeAttrValue {
        /// The device proxy to the device we are registering.
        dev: DevProxy,
        /// Type of event.
        event: AttrValueEvent,
        /// Lowercased attr name that we want to subscribe to.
        name: String,
        /// The sender that the worker should use to answer
        result_tx: oneshot::Sender<Result<Box<dyn AttrValueChannelBackend>>>,
    },
    /// Subscribe to and attribute ready event.
    SubscribeAttrReady {
        /// The device proxy to the device we are registering.
        dev: DevProxy,
        /// Lowercased attr name that we want to subscribe to.
        name: String,
        /// The sender that the worker should use to answer
        result_tx: oneshot::Sender<Result<Box<dyn AttrReadyChannelBackend>>>,
    },
}

/// Information about an admin endpoint, tracking the last heartbeat received from it.
#[derive(Debug)]
struct AdminTracker {
    /// A unique ID for that tracker, this is used to ignoring old scheduled actions.
    id: usize,
    /// The last time a heartbeat was received from this admin device.
    last_heartbeat: Instant,
    /// The endpoint that we are connected to.
    admin_endpoint: Arc<str>,
    /// The list of topics registered onto this admin device.
    event_topics: HashSet<Arc<str>>,
}

/// A channel information associated to a subscription topic.
#[derive(Debug)]
struct ChannelTracker {
    /// A unique ID for that tracker, this is used to ignoring old scheduled actions.
    id: usize,
    /// The device that should be associated to that channel.
    dev: DevProxy,
    /// The IDL version that the device server advertised!
    idl_version: IdlVersion,
    /// The endpoint that we are connected to for this event.
    event_endpoint: Arc<str>,
    /// The base admin topic this channel is linked to.
    admin_topic_base: Arc<str>,
    /// The name of the target object.
    name: String,
    // The kind of channel.
    kind: ChannelTrackerKind,
}

/// The different kind of channels that exists.
#[derive(Debug)]
enum ChannelTrackerKind {
    /// This value is used when subscribing for specifying that the tracker was not 
    /// existing and it was just subscribed to.
    Missing,
    /// The channel tracks some attribute value event.
    AttrValue(AttrValueChannelTracker),
    /// The channel tracks some attribute ready event.
    AttrReady(AttrReadyChannelTracker),
}

/// Tracking for attribute value channel.
#[derive(Debug)]
struct AttrValueChannelTracker {
    /// The kind of event for this specific channel.
    event: AttrValueEvent,
    tx: watch::Sender<orb::CORBA_Any>,
}

/// Tracking for attribute data ready channel.
#[derive(Debug)]
struct AttrReadyChannelTracker {
    tx: watch::Sender<()>,
}

/// The returned implementation of [`AttrValueChannelBackend`].
#[derive(Debug)]
struct ZmqAttrValueChannelBackend {
    /// Received for the watch channel of the event data.
    rx: watch::Receiver<orb::CORBA_Any>,
    /// The IDL version, used to know which attribute value to extract.
    idl_version: IdlVersion,
}

#[async_trait::async_trait]
impl AttrValueChannelBackend for ZmqAttrValueChannelBackend {
    async fn wait(&mut self) -> Result<Box<dyn AttrValueChannelGuard + '_>> {

        self.rx.changed().await
            .map_err(|_| Error::new(ErrorKind::Internal)
                .with_description("Event channel has been closed unexpectedly"))?;

        Ok(Box::new(ZmqAttrValueChannelGuard {
            any: self.rx.borrow(),
            idl_version: self.idl_version,
        }))
        
    }
}

/// The returned implementation of [`AttrValueChannelGuard`].
#[derive(Debug)]
struct ZmqAttrValueChannelGuard<'a> {
    /// The ref guarding the watch from being updated at the same time!
    any: watch::Ref<'a, orb::CORBA_Any>,
    /// The IDL version, used to know which attribute value to extract.
    idl_version: IdlVersion,
}

impl AttrValueChannelGuard for ZmqAttrValueChannelGuard<'_> {
    fn get(&self) -> Result<AttrRead<'_, Data<'_>>> {
        
        let res = match self.idl_version {
            IdlVersion::V1 |
            IdlVersion::V2 => self.any.extract_borrow::<orb::Tango_AttributeValue>()
                .map(attr_read_from_value_v1),
            IdlVersion::V3 => self.any.extract_borrow::<orb::Tango_AttributeValue_3>()
                .map(attr_read_from_value_v3),
            IdlVersion::V4 => self.any.extract_borrow::<orb::Tango_AttributeValue_4>()
                .map(attr_read_from_value_v4),
            IdlVersion::V5 |
            IdlVersion::V6 => self.any.extract_borrow::<orb::Tango_AttributeValue_5>()
                .map(attr_read_from_value_v5),
        };

        match res {
            Some(Ok(val)) => Ok(val),
            Some(Err(err)) => Err(err),
            None => Err(Error::new(ErrorKind::InvalidData)
                .with_append_description("Incoherent value in the internal CORBA::Any"))
        }

    }
}

/// The returned implementation of [`AttrChannel`].
#[derive(Debug)]
struct ZmqAttrReadyChannelBackend {
    /// Received for the watch channel of the event data.
    rx: watch::Receiver<()>,
}

#[async_trait::async_trait]
impl AttrReadyChannelBackend for ZmqAttrReadyChannelBackend {
    async fn wait(&mut self) -> Result<()> {
        
        self.rx.changed().await
            .map_err(|_| Error::new(ErrorKind::Internal)
                .with_description("Event channel has been closed unexpectedly"))?;

        Ok(())

    }
}

/// An enumeration of the supported IDL versions for ZMQ event subscribing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum IdlVersion {
    V1,
    V2,
    V3,
    V4,
    V5,
    V6,
}

/// A subscription information returned by a device server upon subscribing.
#[derive(Debug)]
struct SubscriptionInfo {
    /// The decimal Tango library version.
    lib_version: u32,
    /// The IDL version of the device server.
    idl_version: IdlVersion,
    /// If `lib_version >= 930` then the topics are given by the server for the channel 
    /// and the object events.
    topics: Option<SubscriptionEndpoint>,
    /// The endpoints for the sockets of both channel and object 
    endpoints: Vec<SubscriptionEndpoint>,
    /// The long options for the subscription.
    longs: Vec<i32>,
}

/// Value pair corresponding to the two pub/sub endpoints, for the admin (heartbeat) and
/// the event itself.
#[derive(Debug)]
struct SubscriptionEndpoint {
    /// Value associated to the admin endpoint.
    admin: Arc<str>,
    /// Value associated to the event endpoint.
    event: Arc<str>,
}

impl<'d> FromData<'d> for SubscriptionInfo {
    
    const DATA_KIND: DataKind = DataKind::Struct(Cow::Borrowed(&[
        DataKind::Long(DataLayoutKind::Spectrum(None)),
        DataKind::String(DataLayoutKind::Spectrum(None)),
    ]));

    type Err = String;

    fn from_data(data: Data<'d>) -> std::result::Result<Self, Self::Err> {

        let (
            lvalue, 
            svalue,
        ) = <(Vec<i32>, Vec<Cow<'d, str>>)>::from_data(data)?;

        if lvalue.len() < 2 {
            return Err(format!("Data.0 is too little for a valid SubscriptionInfo"));
        }

        if svalue.len() % 2 != 0 {
            return Err(format!("Data.1 length should be even for a valid SubscriptionInfo"));
        }

        let lib_version = u32::try_from(lvalue[0])
            .map_err(|_| "Data.0[0] does not fit in a u32")?;

        let idl_version = u32::try_from(lvalue[1])
            .map_err(|_| "Data.0[1] does not fit in a u32")?;

        let idl_version = match idl_version {
            1 => IdlVersion::V1,
            2 => IdlVersion::V2,
            3 => IdlVersion::V3,
            4 => IdlVersion::V4,
            5 => IdlVersion::V5,
            6 => IdlVersion::V6,
            _ => return Err("Data.0[1] is not a valid IDL version".to_string()),
        };

        // We collect all endpoints!
        let mut endpoints = (0..(svalue.len() / 2))
            .map(|i| {
                SubscriptionEndpoint {
                    admin: (&*svalue[i * 2 + 0]).into(),
                    event: (&*svalue[i * 2 + 1]).into(),
                }
            })
            .collect::<Vec<_>>();

        // If version is sufficient, gather the topic filters.
        let topics;
        if lib_version >= 930 {
            let endpoint = endpoints.pop()
                .ok_or("Data.1 should end with the event and admin topics")?;
            topics = Some(SubscriptionEndpoint {
                admin: endpoint.event, // NOTE: We are just swapping both, because
                event: endpoint.admin, // this is how it is done...
            });
        } else {
            topics = None;
        }

        Ok(Self {
            lib_version,
            idl_version,
            topics,
            endpoints,
            longs: lvalue[2..].to_vec(),
        })
        
    }

}