rstango 0.4.0

//! This module defines a process-wide registry 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.
//!
//! A client is given the device itself, which it then holds for as long as it is connected
//! to it: a device that is not served anymore refuses every request, which is what makes
//! the client look it up again, see [`crate::server::Dev::is_withdrawn`].

use std::sync::{Arc, LazyLock, Mutex, Weak};
use std::collections::HashMap;
use std::net::IpAddr;

use crate::name::{ArcName, Name};
use crate::backend::BackendDev;


/// Every device this process serves, weakly, which a device is taken out of once it is not
/// served anymore, see [`LocalDevHandle`]: the registry itself never keeps one alive.
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.
///
/// A client is given the device itself, see [`find`], which it keeps once the handle is
/// dropped: a device that must stop answering then refuses its requests itself, as the
/// devices of a server do, see [`crate::server::Dev::is_withdrawn`].
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.
pub fn find(itr: &str) -> Option<Arc<dyn BackendDev>> {
    REGISTRY.lock().unwrap().itr_mapping.get(itr)?.dev.upgrade()
}

/// The device served by this process under that name by a server listening at that host
/// and port, if one is.
///
/// 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;
    }
    entry.dev.upgrade()
}

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

    }

}

#[cfg(test)]
mod tests {

    use futures::future::{ready, BoxFuture};
    use futures::executor::block_on;
    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};

    use super::*;

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

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

    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 admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> { not_implemented() }
        fn ping(&self) -> BoxFuture<'_, Result<()>> { 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<Scalar<str>>> { not_implemented() }
        fn black_box(&self, _max: Option<usize>) -> BoxFuture<'_, Result<Spectrum<str>>> { 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(&self, _name: Scalar<str>, _max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<Value>>>> { not_implemented() }
        fn attr_history(&self, _name: Scalar<str>, _max: usize) -> BoxFuture<'_, Result<Vec<RecordedResult<AttrRead<Value>>>>> { not_implemented() }
        fn subscribe_attr_value(&self, _admin: Arc<dyn BackendDev>, _name: Scalar<str>, _event: AttrValueEvent) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrRead<Value>>>>> { not_implemented() }
        fn subscribe_attr_ready(&self, _admin: Arc<dyn BackendDev>, _name: Scalar<str>) -> BoxFuture<'_, Result<BoxStream<'static, Result<AttrReady>>>> { not_implemented() }
        fn subscribe_attr_config(&self, _admin: Arc<dyn BackendDev>, _name: Scalar<str>) -> BoxFuture<'_, 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
    /// holds the device it found, whatever its server does of 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, which its server let go of, is kept by the client that found
        // it, until that client lets go of it too.
        let weak = Arc::downgrade(&served);
        drop(served);
        assert_eq!(&*name(&client), "old");
        drop(client);
        assert!(weak.upgrade().is_none());

    }

}