rstango 0.4.0

//! Generic backend interfaces for devices.

use std::ops::{Bound, RangeBounds};
use std::sync::{Arc, Mutex};
use std::fmt::Debug;

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

use crate::dev::{AttrConfig, AttrInfo, AttrRead, AttrReady, AttrValueEvent, CommandInfo, DevInfo, DevSource, DevState, Named, NamedResult, RecordedResult};
use crate::value::{Scalar, Spectrum, Value};
use crate::error::{Error, Result};


// =========================== //
//       Global backend       //
// =========================== //

/// A protocol: how devices are reached, and how they are served.
pub trait Backend: Debug + Send + Sync {

    /// Use this backend to connect to a device given its ITR (Interoperable Tango
    /// Reference). If the backend does not support this type of ITR, then it returns
    /// [`ErrorKind::NotHandled`](crate::error::ErrorKind::NotHandled).
    fn connect(&self, itr: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>>;

    /// Use this backend to connect to a device given its name, and the host/port of the
    /// service running the device server.
    fn connect_at(&self, host: &str, port: u16, name: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>>;

    /// Start serving with this protocol, as configured, resolving to the server once it is
    /// listening: see [`BackendServer::host`] and [`BackendServer::port`] for where it ends
    /// up.
    fn serve<'a>(&'a self, config: &'a BackendServerConfig) -> BoxFuture<'a, Result<Arc<dyn BackendServer>>>;

}

// =========================== //
//       Device backend        //
// =========================== //

/// The goal of a backend is to abstract away the actual technology and protocol used to
/// connect to a specific device. A running backend is a connection to one device, which
/// fails with [`ErrorKind::DevNotConnected`](crate::error::ErrorKind::DevNotConnected) once
/// the device is unreachable.
///
/// No function here waits: a request is issued and a future for its answer is given back,
/// so a caller makes as many requests as it likes before awaiting any of them, and awaits
/// all of them at once through a [`futures::future::join_all`] or a `select!`. Backends are
/// expected to multiplex concurrent requests efficiently, pipelining them on a single
/// connection where the protocol allows it.
///
/// The request goes out when the function is **called** and not when the future it gives
/// back is first polled, which is what makes issuing a batch of them and awaiting the batch
/// afterwards worth anything: a future that is held and never polled is a request that is
/// already on its way.
///
/// The names and the values of a request are given owned, as buffers sharing their octets,
/// so that a backend answering in process, a served device above all, keeps them for as
/// long as it likes, and answers with the very names it was given, rather than copying
/// them. A backend that sends its arguments somewhere marshals them on the call.
///
/// Every future given out here **resolves exactly once**: on the device's answer, on the
/// failure of the connection carrying it, or on the backend giving up on it under whatever
/// timeout it was configured with. There is no deadline in this interface — a caller that
/// will not wait forever bounds its own wait, wrapping the future in the timeout of
/// whichever executor it runs under, and a caller that has stopped caring drops the future,
/// which is what tells the backend to cancel the request rather than pay for its answer.
///
/// The requests resolve to what the device answered, decoded: every [`Value`] shares the
/// octets of the message it was decoded out of, so that reading a spectrum of a million
/// points copies nothing. The requests that answer for several names resolve to one entry
/// per name, see [`Named`].
///
/// A backend answers the requests its protocol or its devices do not have with a
/// [`crate::error::ErrorKind::NotImplemented`] naming what is missing, and every caller has
/// to deal with that anyway. The parts of
/// the Tango interface this crate does not deal with at all — the pipes above all — have no
/// place here: a backend has nothing to answer about what this crate has no notion of.
pub trait BackendDev: Debug + Send + Sync {

    /// Get the full domain/family/member name of this device.
    fn name(&self) -> BoxFuture<'_, Result<Arc<str>>>;

    /// Get the full domain/family/member name of this device's admin device.
    fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>>;

    /// Get the description of the device, by itself.
    fn description(&self) -> BoxFuture<'_, Result<Arc<str>>>;

    /// Static information about the device: which class it is an instance of, which server
    /// runs it and where, see [`DevInfo`].
    ///
    /// None of this changes while the device runs, and none of it is the device's own: it
    /// is what the server it belongs to says of it.
    fn info(&self) -> BoxFuture<'_, Result<DevInfo>>;

    /// Just ping the device.
    fn ping(&self) -> BoxFuture<'_, Result<()>>;

    /// Get the device's state, this should be equivalent to reading the `State` attribute
    /// with [`Self::read_attr`].
    /// Despite being equivalent, this function is provided to allow backends to have
    /// specific implementations for this particular function!
    fn state(&self) -> BoxFuture<'_, Result<DevState>>;

    /// Get the device's status, this should be equivalent to reading the `Status` attribute
    /// with [`Self::read_attr`].
    /// Despite being equivalent, this function is provided to allow backends to have
    /// specific implementations for this particular function!
    fn status(&self) -> BoxFuture<'_, Result<Scalar<str>>>;

    /// Request the black box of this device, listing the actions that have been run on it.
    fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Spectrum<str>>>;

    /// Execute a command on the device, from its name.
    fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>>;

    /// Read an attribute or multiple attributes. Pass None to read all attributes.
    ///
    /// The idea of reading many attributes at once is to ensure that, if any
    /// serialization is enforced on server side, the serialization will be held for
    /// reading all attributes, and then released at the end, thus ensuring atomicity
    /// of the returned values.
    ///
    /// There is one entry per attribute read. If the names are specified, thus
    /// Some(names), then the entries are the attributes in the order they were given, and
    /// there are as many of them, which the caller might check.
    fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>>;

    /// Write the given list of attributes, given as (name, value) pairs. In Tango an
    /// attribute write only carries a value: quality and timestamp cannot be written.
    ///
    /// There is one result per given value, in the order they were given: the outer
    /// result of the call itself is the request failing as a whole, an inner one is that
    /// one attribute refusing the write. Nothing names the attribute back, the caller
    /// having the names it just gave, in that very order.
    fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>>;

    /// Same as [`Self::write_attr`] but immediately followed by a read on the server side.
    fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>>;

    /// Get the command info of the given commands. Pass None to get all of them.
    fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>>;

    /// Get the attribute info (including attr config) of the given attributes. Pass None to
    /// get all of them.
    fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>>;

    /// Change the configuration of the given attributes, each given as the whole
    /// configuration it is to have.
    ///
    /// Only the configuration is writable, which is the part of an [`AttrInfo`] that a
    /// client is allowed to set: what an attribute *is* — its mode, its data kind and its
    /// format — is what its definition says and is never written from the outside.
    ///
    /// A client changes a configuration by reading it and sending it back with what it
    /// means to change, every field it left alone being the one the attribute has already,
    /// which changes nothing. A field with no value brings it back to what the attribute
    /// inherits from its class, if anything.
    ///
    /// One result per given attribute, in the order they were given, see
    /// [`Self::write_attr`].
    fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>>;

    /// The recorded history of a command, oldest first and at most `max` entries of it:
    /// what the polling of the device recorded, or the device itself, which may manage the
    /// history of a command it does not poll.
    ///
    /// A command nothing was recorded of has no history to give, which is a failure and
    /// not an empty history, see [`ErrorKind::CommandNotPolled`] and
    /// [`ErrorKind::AttrNotPolled`].
    ///
    /// [`ErrorKind::CommandNotPolled`]: crate::error::ErrorKind::CommandNotPolled
    /// [`ErrorKind::AttrNotPolled`]: crate::error::ErrorKind::AttrNotPolled
    ///
    /// Every entry is what the command answered then, a run that failed carrying the
    /// failure it failed with, see [`Recorded`](crate::dev::Recorded).
    fn command_history(&self, name: Scalar<str>, max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<Value>>>>;

    /// The recorded history of an attribute, see [`Self::command_history`], a reading
    /// carrying the time it was recorded at as its own.
    fn attr_history(&self, name: Scalar<str>, max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<AttrRead<Value>>>>>;

    // ----------------------- //
    //      Subscriptions      //
    // ----------------------- //

    // A subscription is the one place in this trait where a request does not resolve once:
    // the future resolves once the subscription is in place, to a stream that keeps giving.
    // The stream gives the last event and not the whole history of them, so a subscriber
    // that did not keep up with a device pushing faster than it reads skips to the latest
    // one rather than falling behind forever. Dropping the stream is what unsubscribes,
    // and dropping the future before it resolved cancels the subscription.
    //
    // A subscription that could not be made fails through the future. An item of the
    // stream that is an error is one event that failed, such as the reading the device was
    // to push. Note that a subscription gives no initial value: the first event is the
    // next one the device pushes, so a subscriber that needs to know where the attribute
    // stands reads it, once subscribed.
    //
    // A subscription lives as long as the connection it was made over, just like this
    // device backend does: once it is lost, such as the device becoming unreachable, the
    // stream gives an [`ErrorKind::DevNotConnected`] and ends, the very error a request
    // fails with when its connection is lost. The device may have been restarted,
    // elsewhere even, so making the subscription again is not the backend's to do, but
    // the one's who finds the device again, see [`crate::dev::DevChannel`].
    //
    // The admin device of the device is given, because subscribing goes through it in
    // Tango, and a backend has no way to reach it on its own: where it is, is something
    // only the database knows.

    /// Subscribe to the readings of an attribute, of one of the kinds a device pushes them
    /// under, see [`AttrValueEvent`].
    fn subscribe_attr_value(&self, admin: Arc<dyn BackendDev>, name: Scalar<str>, event: AttrValueEvent) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrRead<Value>>>>>;

    /// Subscribe to an attribute being announced as ready to be read, which carries nothing
    /// of the reading itself: what a device pushes here is that there is something to go
    /// and read, and reading it is the subscriber's to do, see [`AttrReady`].
    fn subscribe_attr_ready(&self, admin: Arc<dyn BackendDev>, name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrReady>>>>;

    /// Subscribe to the configuration of an attribute changing, every event carrying the
    /// configuration as it stands after the change, the same way [`Self::attr_info`] gives
    /// it.
    fn subscribe_attr_config(&self, admin: Arc<dyn BackendDev>, name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrInfo>>>>;

}

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

/// The backend for a server that is served, from the moment [`Backend::serve`] hands it
/// over until it is stopped, see [`Self::stop`], or dropped.
///
/// Nothing here waits but [`Self::wait`]: adding, removing a device and stopping are quick
/// and never block, so they are called from anywhere, a request of one of the devices of
/// the backend included. What they ask for happens at some point after, a device that was
/// removed still answering the requests it was answering, for instance.
///
/// A backend holds the devices it serves, strongly, from [`Self::add`], and, once they
/// were removed or the backend stopped, until their requests in flight are answered, and
/// **never beyond [`Self::wait`]**: once it resolves, no [`Arc`] to any device is left
/// anywhere in the backend. This is what lets the owner of a device rely on the device
/// being dropped with the last reference it has.
///
/// # Dropping
///
/// Dropping the last [`Arc`] to a backend server drops everything of it at once, **with no
/// side effect**: every device it serves is let go of, and every request in flight is
/// cancelled, its future dropped and nothing answered. A server that is to end properly is
/// therefore stopped and waited for before it is dropped.
pub trait BackendServer: Debug + Send + Sync {

    /// Add a device to be served, giving back its ITR, which it is reachable at from then
    /// on or shortly after. The name of the device is given with the device so the server
    /// need not call [`BackendDev::name`] on it. This is quick and never blocks, see the
    /// trait documentation.
    ///
    /// Device names are not case sensitive: a backend lowercases the name as soon as it is
    /// given, which every later lookup of the device, from [`Self::remove`] to the requests
    /// of the clients, then relies on.
    fn add(&self, name: &str, dev: Arc<dyn BackendDev>) -> Result<Arc<str>>;

    /// Stop serving the device of that name, giving back whether one was being served. No
    /// new request reaches it from the call on, or shortly after, and the ones it was
    /// answering are still answered. This is quick and never blocks, see the trait
    /// documentation.
    ///
    /// The device itself is **not** given back: the backend drops what it was holding of
    /// it once its requests are answered, which is what makes it stop being reachable.
    fn remove(&self, name: &str) -> bool;

    /// The host, ip or hostname, where this server is listening.
    fn host(&self) -> Arc<str>;

    /// The TCP port this server is listening to.
    fn port(&self) -> u16;

    /// The version of the protocol the devices are served with, which is what a client is
    /// told of a device in its info and what the database is told when it is exported.
    fn version(&self) -> i32;

    /// Stop serving, which nothing undoes: a backend that was stopped is one no client
    /// reaches again, whatever is added to it afterward. Every device is removed, all at
    /// once, as [`Self::remove`] does: no new request is taken, and the ones being answered
    /// are.
    ///
    /// This does not wait for anything, and can therefore be called from anywhere,
    /// including from a request of one of its devices; see [`Self::wait`] for that.
    fn stop(&self);

    /// Resolve once this backend was stopped, see [`Self::stop`], and holds nothing of any
    /// of its devices anymore, see the trait documentation, after which dropping it is
    /// dropping nothing of them.
    ///
    /// This waits until every request in flight is answered, so it may not be awaited from
    /// a future the backend is driving, that is from a request of one of its devices: at
    /// best that request is left hanging, at worst it waits for itself. Waiting is all it
    /// does, the future it gives back never blocking the thread polling it, as for every
    /// future of a backend.
    fn wait(&self) -> BoxFuture<'_, ()>;

    /// Answer the `ZmqEventSubscriptionChange` command of the admin device, which a client
    /// sends to have an event published and to be told where it is published. The
    /// argument and the result are those of the command, as a client sends and expects
    /// them.
    ///
    /// A backend that publishes no event this way answers that it is not implemented,
    /// which is what this does by default.
    fn zmq_event_subscription_change(&self, _args: Spectrum<str>) -> BoxFuture<'_, Result<(Spectrum<i32>, Spectrum<str>)>> {
        Box::pin(ready(Err(Error::not_implemented()
            .with_description("Publishing events over ZMQ is not implemented by this server"))))
    }

    /// Answer the `EventConfirmSubscription` command of the admin device, which a client
    /// sends periodically to keep the events it subscribed to published. The argument is
    /// that of the command, a list of device name, attribute name and event name triplets.
    ///
    /// A backend that publishes no event this way answers that it is not implemented,
    /// which is what this does by default.
    fn event_confirm_subscription(&self, _args: Spectrum<str>) -> BoxFuture<'_, Result<()>> {
        Box::pin(ready(Err(Error::not_implemented()
            .with_description("Publishing events over ZMQ is not implemented by this server"))))
    }

}

/// How a backend serves, see [`Backend::serve`], whatever is not given being left to the
/// backend, which is what the default configuration is.
///
/// A range of ports is given by both its ends, which are included, so that it holds at
/// least one port. A range including port 0 is the system picking any port, whatever
/// the other end, the same as giving none.
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct BackendServerConfig {
    /// The host to listen at, an address or a name of this machine, every one of its
    /// interfaces if none is given.
    pub host: Option<String>,
    /// The port (or range of ports) to listen at, one the system picks if none is given, 
    /// which is what a Tango device server does: its clients find it through the database, 
    /// where it is exported. A fixed port is what a server without a database needs, its 
    /// clients having nothing else to find it by.
    /// The port range has both ends inclusive.
    pub port: Option<(u16, u16)>,
    /// The name of the admin device of the server being served, `dserver/exe/inst`, which
    /// names the event channel of the server: a backend not given one publishes no event.
    pub admin_name: Option<Arc<str>>,
    /// The host and port of the database the server is registered in, which prefixes the
    /// names of its events. A server without a database prefixes them with its own host and
    /// port instead, marking them as such.
    pub tango_host: Option<(Arc<str>, u16)>,
    /// The port (or range of ports) the ZMQ backend may use for its heartbeat PUB socket.
    /// This should be ignored by backends without ZMQ.
    /// The port range has both ends inclusive.
    pub zmq_heartbeat_port: Option<(u16, u16)>,
    /// The port (or range of ports) the ZMQ backend may use for its event PUB socket.
    /// This should be ignored by backends without ZMQ.
    /// The port range has both ends inclusive.
    pub zmq_event_port: Option<(u16, u16)>,
}

impl BackendServerConfig {

    /// Everything left to the backend.
    pub fn new() -> Self {
        Self::default()
    }

    /// Listen at that host, see [`Self::host`].
    pub fn host(mut self, host: &str) -> Self {
        self.host = Some(host.to_string());
        self
    }

    /// Listen at that port, see [`Self::port`].
    pub fn port<R: RangeBounds<u16>>(mut self, port: R) -> Self {
        self.port = Some(port_range_bounds(port));
        self
    }

    /// Serve the server of that admin device, see [`Self::admin_name`].
    pub fn admin_name(mut self, admin_name: impl Into<Arc<str>>) -> Self {
        self.admin_name = Some(admin_name.into());
        self
    }

    /// Serve a server registered in the database at that host and port, see
    /// [`Self::tango_host`].
    pub fn tango_host(mut self, host: impl Into<Arc<str>>, port: u16) -> Self {
        self.tango_host = Some((host.into(), port));
        self
    }

    /// Make ZMQ listen at this port for its heartbeat, see [`Self::zmq_heartbeat_port`].
    pub fn zmq_heartbeat_port<R: RangeBounds<u16>>(mut self, port: R) -> Self {
        self.zmq_heartbeat_port = Some(port_range_bounds(port));
        self
    }

    /// Make ZMQ listen at this port for its events, see [`Self::zmq_event_port`].
    pub fn zmq_event_port<R: RangeBounds<u16>>(mut self, port: R) -> Self {
        self.zmq_event_port = Some(port_range_bounds(port));
        self
    }

}

fn port_range_bounds<R: RangeBounds<u16>>(range: R) -> (u16, u16) {
    let start = match range.start_bound() {
        Bound::Included(&start) => start,
        Bound::Excluded(&start) => start.checked_add(1).expect("range start overflow"),
        Bound::Unbounded => 0,
    };
    let end = match range.end_bound() {
        Bound::Included(&end) => end,
        Bound::Excluded(&end) => end.checked_sub(1).expect("range end underflow"),
        Bound::Unbounded => u16::MAX,
    };
    assert!(start <= end, "slice index starts at {start} but ends at {end}");
    (start, end)
}

// =========================== //
//       Default backend       //
// =========================== //

static DEFAULT_BACKEND: Mutex<Option<Arc<dyn Backend>>> = Mutex::new(None);

/// The backend used by the proxies created without one, which is the CORBA backend unless
/// another one is set with [`set_default_backend`].
pub fn default_backend() -> Arc<dyn Backend> {
    match &mut *DEFAULT_BACKEND.lock().unwrap() {
        Some(backend) => Arc::clone(backend),
        opt @ None => {
            let backend = crate::corba::CorbaBackend::new();
            *opt = Some(Arc::clone(&backend) as Arc<dyn Backend>);
            backend
        }
    }
}

/// Set the backend used by the proxies created without one from then on, see
/// [`default_backend`].
pub fn set_default_backend(backend: Arc<dyn Backend>) {
    *DEFAULT_BACKEND.lock().unwrap() = Some(backend);
}