rstango 0.3.0

//! This modules defines a static registry accross the process which stores the devices
//! that are served locally. This is used to avoid going through the backend's protocol
//! in this case.
//!
//! A server registers every device it serves under the reference its backend gave back 
//! for it, see [`register`], and a client looks up the reference of a device before 
//! connecting to it, see [`find`]. The reference is what identifies a device and not its 
//! name: a process may serve several servers, each registered in a database of its own, 
//! two of which may well serve a device of the same name, which are then two devices.
//!
//! A device served without a database is reached by a client at the address of its server,
//! see [`find_at`], which is why it is also registered under the port of its server and its
//! name, the port being what tells two servers apart there.

use std::sync::{Arc, LazyLock, Mutex, Weak};
use std::task::{Context, Poll};
use std::collections::HashMap;
use std::time::SystemTime;
use std::future::Future;
use std::net::IpAddr;
use std::pin::Pin;

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

use crate::dev::{AttrConfig, AttrInfo, AttrRead, AttrValueEvent, CommandInfo, DevInfo, DevSource, DevState, Named, NamedResult};
use crate::error::{Error, ErrorKind, Result};
use crate::value::{Scalar, Spectrum, Value};
use crate::name::{ArcName, Name};
use crate::backend::BackendDev;


/// Every device this process serves, weakly: a server owns its devices, and drops every
/// reference it has to one it no longer serves, which is then one no client reaches.
static REGISTRY: LazyLock<Mutex<Registry>> = LazyLock::new(Default::default);

/// A mutable registry with the entries mapping from either ITR or the port/name.
#[derive(Debug, Default)]
struct Registry {
    itr_mapping: HashMap<Arc<str>, Entry>,
    port_mapping: HashMap<u16, HashMap<ArcName, Entry>>,
}

#[derive(Debug)]
struct Entry {
    /// The weak reference to the device.
    dev: Weak<dyn BackendDev>,
    /// The host this device's server is listening on.
    host: Arc<str>,
}

impl Entry {

    /// Whether this is the entry of that device, which is how a handle tells its own
    /// entries from those of a device registered in their place since: that device is
    /// another one, and so another allocation, which this one cannot be reused for as long
    /// as a weak reference to it is around.
    fn is(&self, dev: &Weak<dyn BackendDev>) -> bool {
        Weak::ptr_eq(&self.dev, dev)
    }

}

/// Register that device, served under that name by a server listening at that host and
/// port, under the reference its backend gave back for it, until the handle given back is
/// dropped.
///
/// A port of zero is a server that is not listening anywhere, whose devices are only
/// reached through their reference. Whatever was registered under the same reference or
/// the same port and name is replaced, being a device that is not served anymore.
pub fn register(dev: &Arc<dyn BackendDev>, itr: Arc<str>, host: Arc<str>, port: u16, name: Arc<str>) -> LocalDevHandle {

    let dev = Arc::downgrade(dev);
    let mut registry = REGISTRY.lock().unwrap();

    registry.itr_mapping.insert(Arc::clone(&itr), Entry { 
        dev: Weak::clone(&dev), 
        host: Arc::clone(&host),
    });

    let port_name = (port != 0).then(|| {
        registry.port_mapping.entry(port).or_default().insert(ArcName::from(Arc::clone(&name)), Entry { 
            dev: Weak::clone(&dev), 
            host,
        });
        (port, name)
    });

    LocalDevHandle { dev, itr, port_name }

}

/// The device served by this process under that reference, if one is, as a client reaches
/// it, see [`LocalBackendDev`].
pub fn find(itr: &str) -> Option<Arc<dyn BackendDev>> {
    let registry = REGISTRY.lock().unwrap();
    LocalBackendDev::new(registry.itr_mapping.get(itr)?)
}

/// The device served by this process under that name by a server listening at that host
/// and port, if one is, as a client reaches it, see [`LocalBackendDev`].
///
/// The host is either the one the server listens at or one of this machine itself, the
/// loopback, which the port alone then tells apart from any other server of it.
pub fn find_at(host: &str, port: u16, name: &str) -> Option<Arc<dyn BackendDev>> {
    let registry = REGISTRY.lock().unwrap();
    let entry = registry.port_mapping.get(&port)?.get(Name::new(name))?;
    if !host.eq_ignore_ascii_case(&entry.host) && !is_loopback(host) {
        return None;
    }
    LocalBackendDev::new(entry)
}

fn is_loopback(host: &str) -> bool {
    host.eq_ignore_ascii_case("localhost") || host.parse::<IpAddr>().is_ok_and(|addr| addr.is_loopback())
}

// =========================== //
//         Local device        //
// =========================== //

/// A device registered as served by this process, from [`register`] until this is
/// dropped, which is what a server holds for as long as it serves that device.
#[derive(Debug)]
pub struct LocalDevHandle {
    dev: Weak<dyn BackendDev>,
    itr: Arc<str>,
    /// The port and the name it is registered under, if it is, see [`register`].
    port_name: Option<(u16, Arc<str>)>,
}

impl Drop for LocalDevHandle {

    /// Remove what was registered, unless another device was registered in its place since.
    fn drop(&mut self) {

        let mut registry = REGISTRY.lock().unwrap();

        if registry.itr_mapping.get(&self.itr).is_some_and(|entry| entry.is(&self.dev)) {
            registry.itr_mapping.remove(&self.itr);
        }

        if let Some((port, name)) = &self.port_name
        && let Some(names) = registry.port_mapping.get_mut(port) {

            if names.get(Name::new(name)).is_some_and(|entry| entry.is(&self.dev)) {
                names.remove(Name::new(name));
            }

            if names.is_empty() {
                registry.port_mapping.remove(port);
            }
            
        }

    }

}

/// A device served by this process, as a client reaches it: every request goes to the
/// device itself, which is only held while it is answering one, so that a client does not
/// keep a server from dropping a device it no longer serves, and fails to reach it once it
/// did, as it would over a connection.
#[derive(Debug)]
struct LocalBackendDev {
    dev: Weak<dyn BackendDev>,
}

impl LocalBackendDev {

    /// A client of that device, if it is still there.
    fn new(entry: &Entry) -> Option<Arc<dyn BackendDev>> {
        entry.dev.upgrade()?;
        Some(Arc::new(Self { 
            dev: Weak::clone(&entry.dev),
        }))
    }

    /// Forward a call to the backend device, note that this implies some indirections
    /// and so latencies, but this latency will never be as big as going through the CORBA
    /// protocol!
    fn call<'a, T: Send + 'a>(&'a self, request: impl FnOnce(&'a dyn BackendDev) -> BoxFuture<'a, Result<T>>) -> BoxFuture<'a, Result<T>> {

        let Some(dev) = self.dev.upgrade() else {
            return Box::pin(ready(Err(Error::new(ErrorKind::DevNotConnected)
                .with_description("The device is not served by this process anymore"))));
        };

        // SAFETY: The request borrows the device the Arc points to, which does not move and
        // is kept alive by that very Arc for as long as the request is, the two being held
        // together and the request being dropped first, see `Holding`. The lifetime given
        // to the borrow is the one of `self`, which the request cannot outlive either.
        let target = unsafe { &*Arc::as_ptr(&dev) };

        Box::pin(Holding { request: request(target), _dev: dev })

    }

}

/// A request, along with the device it borrows, which is dropped after it.
struct Holding<'a, T> {
    request: BoxFuture<'a, T>,
    _dev: Arc<dyn BackendDev>,
}

impl<T> Future for Holding<'_, T> {

    type Output = T;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
        self.request.as_mut().poll(cx)
    }

}

impl BackendDev for LocalBackendDev {

    fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
        self.call(|dev| dev.name())
    }

    fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
        self.call(|dev| dev.admin_dev_name())
    }

    fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
        self.call(|dev| dev.description())
    }

    fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {
        self.call(|dev| dev.info())
    }

    fn ping(&self) -> BoxFuture<'_, Result<()>> {
        self.call(|dev| dev.ping())
    }

    fn state(&self) -> BoxFuture<'_, Result<DevState>> {
        self.call(|dev| dev.state())
    }

    fn status(&self) -> BoxFuture<'_, Result<String>> {
        self.call(|dev| dev.status())
    }

    fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Vec<String>>> {
        self.call(move |dev| dev.black_box(max))
    }

    fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {
        self.call(move |dev| dev.command(name, input, source))
    }

    fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
        self.call(move |dev| dev.read_attr(names, source))
    }

    fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
        self.call(move |dev| dev.write_attr(values))
    }

    fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
        self.call(move |dev| dev.write_read_attr(values))
    }

    fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {
        self.call(move |dev| dev.command_info(names))
    }

    fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {
        self.call(move |dev| dev.attr_info(names))
    }

    fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
        self.call(move |dev| dev.set_attr_config(configs))
    }

    fn command_history<'a>(&'a self, name: &'a str, max: usize) -> BoxFuture<'a, Result<Vec<(SystemTime, Result<Value>)>>> {
        self.call(move |dev| dev.command_history(name, max))
    }

    fn attr_history<'a>(&'a self, name: &'a str, max: usize) -> BoxFuture<'a, Result<Vec<Result<AttrRead<Value>>>>> {
        self.call(move |dev| dev.attr_history(name, max))
    }

    fn subscribe_attr_value<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str, event: AttrValueEvent) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
        self.call(move |dev| dev.subscribe_attr_value(admin, name, event))
    }

    fn subscribe_attr_ready<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<()>>>> {
        self.call(move |dev| dev.subscribe_attr_ready(admin, name))
    }

    fn subscribe_attr_conf<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> {
        self.call(move |dev| dev.subscribe_attr_conf(admin, name))
    }

}

#[cfg(test)]
mod tests {

    use std::sync::atomic::{AtomicUsize, Ordering};

    use futures::executor::block_on;

    use super::*;

    /// A device answering its name and counting its pings, as soon as it is asked.
    #[derive(Debug)]
    struct TestDev {
        name: &'static str,
        pings: AtomicUsize,
    }

    fn dev(name: &'static str) -> Arc<dyn BackendDev> {
        Arc::new(TestDev { name, pings: AtomicUsize::new(0) })
    }

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

    impl BackendDev for TestDev {

        fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
            Box::pin(ready(Ok(Arc::from(self.name))))
        }

        fn ping(&self) -> BoxFuture<'_, Result<()>> {
            self.pings.fetch_add(1, Ordering::Relaxed);
            Box::pin(ready(Ok(())))
        }

        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 state(&self) -> BoxFuture<'_, Result<DevState>> { not_implemented() }
        fn status(&self) -> BoxFuture<'_, Result<String>> { not_implemented() }
        fn black_box(&self, _max: Option<usize>) -> BoxFuture<'_, Result<Vec<String>>> { not_implemented() }
        fn command(&self, _name: Scalar<str>, _input: Value, _source: DevSource) -> BoxFuture<'_, Result<Value>> { not_implemented() }
        fn read_attr(&self, _names: Option<Spectrum<str>>, _source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> { not_implemented() }
        fn write_attr(&self, _values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> { not_implemented() }
        fn write_read_attr(&self, _values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> { not_implemented() }
        fn command_info(&self, _names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> { not_implemented() }
        fn attr_info(&self, _names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> { not_implemented() }
        fn set_attr_config(&self, _configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> { not_implemented() }
        fn command_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<(SystemTime, Result<Value>)>>> { not_implemented() }
        fn attr_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<Result<AttrRead<Value>>>>> { not_implemented() }
        fn subscribe_attr_value<'a>(&'a self, _admin: Arc<dyn BackendDev>, _name: &'a str, _event: AttrValueEvent) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> { not_implemented() }
        fn subscribe_attr_ready<'a>(&'a self, _admin: Arc<dyn BackendDev>, _name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<()>>>> { not_implemented() }
        fn subscribe_attr_conf<'a>(&'a self, _admin: Arc<dyn BackendDev>, _name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> { not_implemented() }

    }

    fn name(dev: &Arc<dyn BackendDev>) -> Arc<str> {
        block_on(dev.name()).unwrap()
    }

    // The registry is the one of the process, shared by every test running at the same
    // time, which is why each test records its own references and ports.

    /// A device found by its reference, and its requests going to it.
    #[test]
    fn by_itr() {

        let served = dev("test/local/1");
        let _registration = register(&served, Arc::from("test:by_itr"), Arc::from("myhost"), 0, Arc::from("test/local/1"));

        let found = find("test:by_itr").unwrap();
        assert_eq!(&*name(&found), "test/local/1");
        assert!(find("test:by_itr/other").is_none());

        // No port, so nothing to find at an address.
        assert!(find_at("localhost", 0, "test/local/1").is_none());

    }

    /// Two servers serving a device of the same name, which are two devices, told apart by
    /// their reference and by the port of their server.
    #[test]
    fn same_name() {

        let first = dev("first");
        let second = dev("second");
        let _first = register(&first, Arc::from("test:same_name/1"), Arc::from("myhost"), 61001, Arc::from("test/same/1"));
        let _second = register(&second, Arc::from("test:same_name/2"), Arc::from("myhost"), 61002, Arc::from("TEST/same/1"));

        assert_eq!(&*name(&find("test:same_name/1").unwrap()), "first");
        assert_eq!(&*name(&find("test:same_name/2").unwrap()), "second");
        assert_eq!(&*name(&find_at("myhost", 61001, "test/same/1").unwrap()), "first");
        assert_eq!(&*name(&find_at("localhost", 61002, "test/SAME/1").unwrap()), "second");

    }

    /// A device at an address is found at the host its server listens at, or at the
    /// loopback, and nowhere else.
    #[test]
    fn at_host() {

        let served = dev("test/at/1");
        let _registration = register(&served, Arc::from("test:at_host"), Arc::from("MyHost"), 61003, Arc::from("test/at/1"));

        for host in ["myhost", "localhost", "127.0.0.1", "::1"] {
            assert!(find_at(host, 61003, "test/at/1").is_some(), "{host}");
        }
        assert!(find_at("otherhost", 61003, "test/at/1").is_none());
        assert!(find_at("myhost", 61004, "test/at/1").is_none());

    }

    /// A handle dropped takes its device out, unless another one replaced it, and a client
    /// of a device its server let go of fails to reach it, without keeping it.
    #[test]
    fn lifetime() {

        let served = dev("old");
        let old = register(&served, Arc::from("test:lifetime"), Arc::from("myhost"), 61005, Arc::from("test/life/1"));
        let client = find("test:lifetime").unwrap();

        let replacing = dev("new");
        let new = register(&replacing, Arc::from("test:lifetime"), Arc::from("myhost"), 61005, Arc::from("test/life/1"));
        drop(old);
        assert_eq!(&*name(&find("test:lifetime").unwrap()), "new");
        assert_eq!(&*name(&find_at("localhost", 61005, "test/life/1").unwrap()), "new");
        drop(new);
        assert!(find("test:lifetime").is_none());
        assert!(find_at("localhost", 61005, "test/life/1").is_none());

        // The old device, once its server let go of it, is neither kept nor reached.
        let weak = Arc::downgrade(&served);
        assert_eq!(&*name(&client), "old");
        drop(served);
        assert!(weak.upgrade().is_none());
        assert_eq!(block_on(client.ping()).unwrap_err().kind(), ErrorKind::DevNotConnected);

    }

    /// A request is issued when it is made, not when it is first polled, and holds the
    /// device until it is answered.
    #[test]
    fn eager() {

        let served = Arc::new(TestDev { name: "test/eager/1", pings: AtomicUsize::new(0) });
        let as_dev = Arc::clone(&served) as Arc<dyn BackendDev>;
        let _registration = register(&as_dev, Arc::from("test:eager"), Arc::from("myhost"), 0, Arc::from("test/eager/1"));
        drop(as_dev);
        let client = find("test:eager").unwrap();

        let ping = client.ping();
        assert_eq!(served.pings.load(Ordering::Relaxed), 1);
        assert_eq!(Arc::strong_count(&served), 2);
        block_on(ping).unwrap();
        assert_eq!(Arc::strong_count(&served), 1);

    }

}