tango/server/

server.rs

//! The server itself: what owns the classes, and through them the devices, and the backend
//! serving them.
//!
//! Ownership is deliberately cyclic: the server owns its classes, a class owns its devices,
//! and a device owns its class which owns the server, so that any of them reaches the
//! others as plain field access. The cycles are broken by the [`ServerHandle`] alone, which
//! empties every map when it is dropped, see [`Server::clear`]. Nothing of a server is
//! therefore handed out as an [`Arc`], only as a [`Ref`].

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::collections::hash_map::Entry;
use std::sync::{Arc, Mutex, Weak};
use std::task::{Context, Poll};
use std::collections::HashMap;
use std::time::Duration;
use std::borrow::Cow;
use std::pin::Pin;
use std::mem;

use futures::stream::{BoxStream, Stream, StreamExt};
use futures::executor::block_on;
use tracing::{debug, info, warn};

use crate::backend::{default_backend, Backend, BackendServer, BackendServerConfig};
use crate::database::{DatabaseProxy, ExportDevInfo};
use crate::dev::{AttrInfo, AttrRead, AttrValueEvent};
use crate::error::{Error, ErrorKind, Result, reason};
use crate::future::block_on_timeout;
use crate::name::{ArcName, Name};
use crate::value::Value;
use crate::future::Slot;
use crate::testing::TestContext;

use super::class::{AnyClass, Class};
use super::admin::{self, Admin};
use super::def::ClassDef;
use super::{Model, Ref};
use super::dev::AnyDev;
use super::signal;


/// What the name of the admin device of a server is its `exe/inst` name prefixed with.
const ADMIN_DEV_PREFIX: &str = "dserver/";

/// The shared state of a server, co-owned by the [`ServerHandle`] and by every class of it.
#[derive(Debug)]
pub struct Server {
    /// Self reference, for what a server hands its classes of itself.
    this: Weak<Self>,
    /// The `dserver/exe/inst` name of the admin device of that server, which is where its
    /// own `exe/inst` name is taken from, see [`Self::name`].
    pub(super) admin_dev_name: Arc<str>,
    /// The database proxy used for that server.
    database: DatabaseProxy,
    /// The backend serving the devices of that server, owned by the handle, see
    /// [`ServerHandle`]: the backend holds the devices, which hold the server, so this
    /// owning it would be a cycle nothing breaks.
    backend: Weak<dyn BackendServer>,
    /// The classes of that server, each owning its devices.
    classes: Mutex<HashMap<ArcName, Arc<dyn AnyClass>>>,
    /// The number of classes made for that server so far, which is the index of the next
    /// one, see [`AnyClass::index`].
    class_count: AtomicUsize,
    /// Every device of that server, which only the server owns, device names being unique
    /// server-wide. This is the lock every addition of a device is made under, see
    /// [`Class::add_dev`].
    pub(super) devs: Mutex<HashMap<ArcName, Arc<dyn AnyDev>>>,
    /// What the next [`Self::sync`] applies, which is never locked with any other lock.
    pub(super) pending: Mutex<Pending>,
    /// Held for the whole of a [`Self::sync`], so that two of them never interleave, and
    /// while stopping, see [`Self::stop`].
    sync_lock: async_lock::Mutex<()>,
    /// Whether that server was stopped, which is only ever done once.
    stopped: AtomicBool,
    /// The subscriptions to the events of the attributes of the devices of that server, by
    /// device and attribute name.
    ///
    /// They are the server's and not the devices': a device restarted, or a whole server
    /// restarted, is made anew under the same name, and so are its attributes on every
    /// Init, while its subscribers are kept, the way cppTango keeps them. A client of a
    /// server that is still running sees nothing of a restart, so nothing would make it
    /// subscribe again.
    ///
    /// The streams handed out end, after a [`ErrorKind::DevNotConnected`], once the device
    /// is gone for good, see [`Self::sync`], or the server stops, see [`Self::stop`]: the
    /// end of a subscription of a backend, see [`crate::backend::BackendDev`].
    events: Mutex<HashMap<ArcName, HashMap<ArcName, AttrEvents>>>,
    /// Whether the devices of that server are reached directly by the clients of this
    /// process, see [`ServerBuilder::local`].
    pub(super) local: bool,
}

/// The subscribers of the events of an attribute, see [`Server::events`].
#[derive(Debug, Default)]
struct AttrEvents {
    values: HashMap<AttrValueEvent, Subscribers<AttrRead<Value>>>,
    ready: Subscribers<()>,
    conf: Subscribers<AttrInfo>,
}

/// The subscribers of an event, which only hold their slot weakly: a subscriber dropping
/// its stream is what unsubscribes it, its slot being pruned the next time the event is
/// pushed or subscribed to.
type Subscribers<T> = Vec<Weak<Slot<Result<T>>>>;

/// The operations on the devices of a server that wait for its next [`Server::sync`].
#[derive(Debug, Default)]
pub struct Pending {
    /// The devices added and not initialized yet, in the order they were added.
    pub added: Vec<Arc<dyn AnyDev>>,
    /// The devices initialized and not exported yet, either because they were just
    /// initialized or because serving or exporting them failed, along with the reference
    /// the backend serves them at once it does.
    pub unexported: Vec<(Arc<dyn AnyDev>, Option<Arc<str>>)>,
    /// The devices removed and still served, possibly still exported.
    pub removed: Vec<Arc<dyn AnyDev>>,
}

impl Server {

    /// The `exe/inst` name of that server, under which it is registered in the database.
    #[inline]
    pub fn name(&self) -> &str {
        &self.admin_dev_name[ADMIN_DEV_PREFIX.len()..]
    }

    #[inline]
    pub fn admin_dev_name(&self) -> &str {
        &self.admin_dev_name
    }

    #[inline]
    pub fn database(&self) -> &DatabaseProxy {
        &self.database
    }

    /// The backend serving the devices of that server, [`None`] once its handle is
    /// dropped, see [`ServerHandle`].
    #[inline]
    pub fn backend(&self) -> Option<Arc<dyn BackendServer>> {
        self.backend.upgrade()
    }

    /// The backend serving the devices of that server, failing once its handle is
    /// dropped, see [`Self::backend`].
    pub(super) fn served_backend(&self) -> Result<Arc<dyn BackendServer>> {
        self.backend().ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_ServerNotRunning))
            .with_append_description(format_args!("Server {} is not served anymore", self.name())))
    }

    /// Register a new class in this server, from its definition.
    pub fn add_class<M: Model>(&self, def: ClassDef<M>) -> Result<Ref<'_, Class<M>>> {

        let def = Arc::new(def);
        let mut classes = self.classes.lock().unwrap();
        let name = ArcName::from(Arc::from(&*def.name));

        let Entry::Vacant(entry) = classes.entry(name) else {
            return Err(Error::new(ErrorKind::ClassAlreadyDefined)
                .with_append_description(format_args!("Class {:?}", def.name)));
        };

        // A server is only ever reached through a reference to it, so it is alive.
        let index = self.class_count.fetch_add(1, Ordering::Relaxed);
        let class = Class::new(self.this.upgrade().unwrap(), def, index);
        entry.insert(Arc::clone(&class) as Arc<dyn AnyClass>);
        Ok(Ref::new(class))

    }

    /// Get a class of this server from its name.
    pub fn get_class(&self, name: &str) -> Option<Ref<'_, dyn AnyClass>> {
        self.classes.lock().unwrap().get(Name::new(name)).map(|class| Ref::new(Arc::clone(class)))
    }

    /// Add a device to this server, as an instance of the class of that name, which is how
    /// the devices listed by the database are added, the database giving the class of each
    /// by its name. See [`Class::add_dev`] for the typed form.
    /// 
    /// The device is only registered: it is initialized, served and exported by the next
    /// [`Self::sync`].
    pub fn add_dev(&self, class: &str, name: &str) -> Result<Ref<'_, dyn AnyDev>> {

        let class = self.classes.lock().unwrap().get(Name::new(class))
            .map(Arc::clone)
            .ok_or_else(|| Error::new(ErrorKind::ClassNotFound)
                .with_append_description(format_args!("Class {class:?}")))?;

        class.add_dev(name).map(Ref::new)

    }

    /// Remove the device of that name from this server, giving back whether it had one.
    /// 
    /// Its name is free again right away, but the device itself is only unexported, no
    /// longer served and deleted by the next [`Self::sync`]. A device removed before any
    /// synchronization reached it is simply dropped.
    pub fn remove_dev(&self, name: &str) -> bool {

        let Some(dev) = self.devs.lock().unwrap().remove(Name::new(name)) else {
            return false;
        };

        // If the device was still pending to be added, then we just remove it, otherwise
        // we also remove its unexported state and then add it to the devices to be 
        // removed.
        let mut pending = self.pending.lock().unwrap();
        if let Some(index) = pending.added.iter().position(|added| Arc::ptr_eq(added, &dev)) {
            pending.added.remove(index);
        } else {
            pending.unexported.retain(|(unexported, _)| !Arc::ptr_eq(unexported, &dev));
            pending.removed.push(dev);
        }

        true

    }

    /// Apply what was done to the devices of this server since the last synchronization:
    /// the devices removed are unexported, no longer served and deleted, then the devices
    /// added are initialized, served and exported, which is what makes them reachable.
    /// 
    /// The devices added are initialized one after the other, class by class in the order
    /// the classes were made, and in the order they were added within a class. A device
    /// whose initialization adds devices has them initialized by this same
    /// synchronization, which is how the admin device boots a server. Each device is then
    /// served and exported on its own, which is what a Tango server always tells the
    /// database.
    /// 
    /// A device whose initialization failed is still served and exported, in FAULT, so
    /// that a client can ask it what went wrong. A device that could not be served or
    /// exported is retried by the next synchronization, and the first such error is given
    /// back once every other device is done.
    /// 
    /// Synchronizations never interleave: one waits for the one running to be done, so
    /// that everything done before calling it is applied once it returns. This is also why
    /// an init function must never call it, since it would wait for itself.
    pub async fn sync(&self) -> Result<()> {

        let _sync = self.sync_lock.lock().await;
        if self.stopped.load(Ordering::Acquire) {
            return Ok(());
        }

        let mut first_err = None;

        // Begin with pending removal of devices, they are first unexported and then we
        // call dev.remove to make it remove itself from the backend.
        let removed = mem::take(&mut self.pending.lock().unwrap().removed);
        for dev in removed {

            if let Err(err) = self.database.unexport_dev_async(dev.name()).await {
                warn!(dev = %dev.name(), %err, "Failed to unexport the removed device");
                first_err.get_or_insert(err);
            }

            if let Err(err) = dev.withdraw().await {
                warn!(dev = %dev.name(), %err, "Failed to delete the removed device");
                first_err.get_or_insert(err);
            }

            // A device added again under the same name, which is what a restart is, keeps
            // the subscriptions of the one it replaces.
            if !self.devs.lock().unwrap().contains_key(Name::new(dev.name())) {
                self.end_dev_events(dev.name());
            }

            debug!(dev = %dev.name(), "Removed the device");

        }

        // Then we loop through added devices, we init their model and then we add them to
        // the unexported list, which tell that they should be exported next!
        loop {

            let mut added = mem::take(&mut self.pending.lock().unwrap().added);
            if added.is_empty() {
                break;
            }

            added.sort_by_key(|dev| dev.class().index());
            for dev in added {
                // Only a panic of its init function ends here, a failed initialization
                // being kept by the device itself: it is exported all the same.
                if let Err(err) = dev.init().await {
                    warn!(dev = %dev.name(), %err, "Failed to complete the initialization of the device");
                }
                self.pending.lock().unwrap().unexported.push((dev, None));
            }

        }

        // Finally we process all the unexported device by exporting them to the database.
        let unexported = mem::take(&mut self.pending.lock().unwrap().unexported);
        let mut failed = Vec::new();
        for (dev, itr) in unexported {

            let served = match itr {
                Some(itr) => Ok(itr),
                None => dev.serve(),
            };

            let itr = match served {
                Ok(itr) => itr,
                Err(err) => {
                    warn!(dev = %dev.name(), %err, "Failed to serve the device");
                    first_err.get_or_insert(err);
                    failed.push((dev, None));
                    continue;
                }
            };

            if let Err(err) = self.export_dev(&*dev, &itr).await {
                warn!(dev = %dev.name(), %err, "Failed to export the device");
                first_err.get_or_insert(err);
                failed.push((dev, Some(itr)));
                continue;
            }

            debug!(dev = %dev.name(), "Exported the device");

        }

        self.pending.lock().unwrap().unexported.extend(failed);
        first_err.map_or(Ok(()), Err)

    }

    /// The blocking form of [`Self::sync`], which must therefore never run on the executor
    /// of a backend, in an asynchronous handler in particular.
    pub fn sync_blocking(&self) -> Result<()> {
        block_on(self.sync())
    }

    /// Export a device served at that reference in the database, which is what a client
    /// looks it up in.
    async fn export_dev(&self, dev: &dyn AnyDev, itr: &str) -> Result<()> {
        let backend = self.served_backend()?;
        self.database.export_dev_async(ExportDevInfo {
            name: Cow::Borrowed(dev.name()),
            itr: Cow::Borrowed(itr),
            host: Cow::Borrowed(&backend.host()),
            version: Cow::Owned(backend.version().to_string()),
            pid: std::process::id() as i32,
        }).await
    }

    /// Every class of this server, in the order they were made, the admin one first, see
    /// [`AnyClass::index`].
    pub fn classes(&self) -> Vec<Ref<'_, dyn AnyClass>> {
        let mut classes = self.classes.lock().unwrap().values().map(Arc::clone).collect::<Vec<_>>();
        classes.sort_by_key(|class| class.index());
        classes.into_iter().map(Ref::new).collect()
    }

    /// Every device of this server, class by class in the order of [`Self::classes`], and
    /// by name within a class.
    pub fn devs(&self) -> Vec<Ref<'_, dyn AnyDev>> {
        let mut devs = self.devs.lock().unwrap().values().map(Arc::clone).collect::<Vec<_>>();
        devs.sort_by(|a, b| a.class().index().cmp(&b.class().index()).then_with(|| a.name().cmp(b.name())));
        devs.into_iter().map(Ref::new).collect()
    }

    /// Get a device of this server from its name.
    pub fn get_dev(&self, name: &str) -> Option<Ref<'_, dyn AnyDev>> {
        self.devs.lock().unwrap().get(Name::new(name)).map(|dev| Ref::new(Arc::clone(dev)))
    }

    /// Stop this server, which is only ever done once: it is unexported as a whole, so
    /// that clients stop finding it, then its backend is stopped, the requests in flight
    /// still being answered. The thread waiting on the server then deletes the models, see
    /// [`ServerHandle::wait`]. No synchronization runs after this, the devices still
    /// waiting for one never being served.
    /// 
    /// This does not wait for the backend, and can therefore be called from anywhere, a
    /// request handler included, which is what the `Kill` command of the admin device does.
    /// The backend is stopped even if the server could not be unexported, which is the
    /// error then given back.
    pub async fn stop(&self) -> Result<()> {

        let _sync = self.sync_lock.lock().await;
        if self.stopped.swap(true, Ordering::AcqRel) {
            return Ok(());
        }

        info!(server = %self.name(), "Stopping the server");

        let unexported = self.database.unexport_server_async(self.name()).await;
        if let Err(err) = &unexported {
            warn!(server = %self.name(), %err, "Failed to unexport the server");
        }

        if let Some(backend) = self.backend() {
            backend.stop();
        }

        self.end_events();

        unexported

    }

    /// Delete the model of every device, the classes in the reverse of their order, the
    /// admin device last, once the backend holds nothing of them anymore and therefore no
    /// request runs on any of them. The devices are taken out, so this is only ever done
    /// once.
    async fn delete_devs(&self) -> Result<()> {

        let mut devs = mem::take(&mut *self.devs.lock().unwrap())
            .into_values()
            .collect::<Vec<_>>();
        
        devs.sort_by_key(|dev| std::cmp::Reverse(dev.class().index()));
        *self.pending.lock().unwrap() = Pending::default();

        let mut first_err = None;
        for dev in devs {
            if let Err(err) = dev.withdraw().await {
                warn!(dev = %dev.name(), %err, "Failed to delete the device");
                first_err.get_or_insert(err);
            }
        }

        first_err.map_or(Ok(()), Err)

    }

    /// Let go of every device and every class, which is what breaks the ownership cycles.
    /// Nothing is deleted here: the shutdown deletes every model before this.
    fn clear(&self) {
        let devs = std::mem::take(&mut *self.devs.lock().unwrap());
        drop(devs);
        let classes = std::mem::take(&mut *self.classes.lock().unwrap());
        drop(classes);
        // A server dropped without stopping lets go of its subscribers all the same.
        self.end_events();
    }

    /// Subscribe to the readings of an attribute of a device, pushed for that event. Nothing
    /// checks here that the device has the attribute, which is the device's to check.
    pub(super) fn subscribe_attr_value(&self, dev: &str, attr: &str, event: AttrValueEvent) -> BoxStream<'static, Result<AttrRead<Value>>> {
        self.subscribe_event(dev, attr, |events| events.values.entry(event).or_default()).boxed()
    }

    /// Subscribe to an attribute of a device being ready to be read.
    pub(super) fn subscribe_attr_ready(&self, dev: &str, attr: &str) -> BoxStream<'static, Result<()>> {
        self.subscribe_event(dev, attr, |events| &mut events.ready).boxed()
    }

    /// Subscribe to the configuration of an attribute of a device.
    pub(super) fn subscribe_attr_conf(&self, dev: &str, attr: &str) -> BoxStream<'static, Result<AttrInfo>> {
        self.subscribe_event(dev, attr, |events| &mut events.conf).boxed()
    }

    /// Push a reading of an attribute of a device, for that event.
    pub(super) fn push_attr_value(&self, dev: &str, attr: &str, event: AttrValueEvent, read: Result<AttrRead<Value>>) {
        self.push_event(dev, attr, |events| events.values.entry(event).or_default(), read);
    }

    /// Push an attribute of a device being ready to be read.
    pub(super) fn push_attr_ready(&self, dev: &str, attr: &str, ready: Result<()>) {
        self.push_event(dev, attr, |events| &mut events.ready, ready);
    }

    /// Push the configuration of an attribute of a device.
    pub(super) fn push_attr_conf(&self, dev: &str, attr: &str, info: Result<AttrInfo>) {
        self.push_event(dev, attr, |events| &mut events.conf, info);
    }

    fn subscribe_event<T>(&self, dev: &str, attr: &str, select: impl FnOnce(&mut AttrEvents) -> &mut Subscribers<T>) -> EventStream<T> {

        let slot = Arc::new(Slot::new());

        let mut devs = self.events.lock().unwrap();
        let attrs = devs.entry(ArcName::from(Arc::<str>::from(dev))).or_default();
        let events = attrs.entry(ArcName::from(Arc::<str>::from(attr))).or_default();
        let subscribers = select(events);
        subscribers.retain(|slot| slot.strong_count() != 0);
        subscribers.push(Arc::downgrade(&slot));

        EventStream {
            slot,
            ended: false,
        }

    }

    fn push_event<T: Clone>(&self, dev: &str, attr: &str, select: impl FnOnce(&mut AttrEvents) -> &mut Subscribers<T>, event: Result<T>) {

        let mut devs = self.events.lock().unwrap();
        let Some(attrs) = devs.get_mut(Name::new(dev)) else { return };
        let Some(events) = attrs.get_mut(Name::new(attr)) else { return };

        // Pushed to under the lock, a slot only waking its subscriber.
        let subscribers = select(events);
        subscribers.retain(|slot| slot.strong_count() != 0);

        // Each subscriber has its own copy of the event, the last one taking it as it is.
        if let Some((last, others)) = subscribers.split_last() {
            
            for slot in others.iter().filter_map(Weak::upgrade) {
                slot.push(match &event {
                    Ok(value) => Ok(value.clone()),
                    Err(err) => Err(err.duplicate()),
                });
            }

            if let Some(slot) = last.upgrade() {
                slot.push(event);
            }

        }

        // What is left with no subscriber at all is removed, all the way up.
        events.values.retain(|_, subscribers| !subscribers.is_empty());
        if events.values.is_empty() && events.ready.is_empty() && events.conf.is_empty() {
            attrs.remove(Name::new(attr));
            if attrs.is_empty() {
                devs.remove(Name::new(dev));
            }
        }

    }

    /// End the subscriptions to the events of that device, which is gone for good.
    fn end_dev_events(&self, dev: &str) {
        let attrs = self.events.lock().unwrap().remove(Name::new(dev));
        for events in attrs.into_iter().flat_map(HashMap::into_values) {
            events.end();
        }
    }

    /// End every subscription, the server having stopped.
    fn end_events(&self) {
        let devs = mem::take(&mut *self.events.lock().unwrap());
        for events in devs.into_values().flat_map(HashMap::into_values) {
            events.end();
        }
    }

}

impl AttrEvents {

    /// End the subscriptions of every subscriber still there.
    fn end(self) {
        for subscribers in self.values.into_values() {
            end_subscribers(subscribers);
        }
        end_subscribers(self.ready);
        end_subscribers(self.conf);
    }

}

fn end_subscribers<T>(subscribers: Subscribers<T>) {
    for slot in subscribers.iter().filter_map(Weak::upgrade) {
        slot.end();
    }
}

/// The events of one subscriber, which unsubscribes by being dropped, see [`Subscribers`].
struct EventStream<T> {
    slot: Arc<Slot<Result<T>>>,
    /// Set once the end of the subscription was given.
    ended: bool,
}

impl<T> Stream for EventStream<T> {

    type Item = Result<T>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {

        let this = self.get_mut();
        if this.ended {
            return Poll::Ready(None);
        }

        // The end of the slot is given as the end of the subscription, once.
        this.slot.poll_take(cx).map(|event| event.or_else(|| {
            this.ended = true;
            Some(Err(Error::new(ErrorKind::DevNotConnected)
                .with_append_description("The device is no longer served")))
        }))

    }

}

/// What a server is built with, and served by, see [`Self::serve`].
pub struct ServerBuilder {
    exe: String,
    inst: String,
    backend: Option<Arc<dyn Backend>>,
    database: Option<DatabaseProxy>,
    /// How the backend serves, see [`Self::host`] and [`Self::port`].
    config: BackendServerConfig,
    /// See [`Self::local`].
    local: bool,
    /// See [`Self::stop_on_signal`].
    stop_on_signal: bool,
    /// The database for testing the server is registered in, which the server owns for
    /// as long as it runs, see [`ServerHandle`].
    pub(super) test_context: Option<TestContext>,
    /// The definitions of the classes of that server, each made a class once the server
    /// exists, which is only once it is served: a class owns the server it is in. Kept in
    /// the order they were defined, which is the order their devices are initialized in.
    classes: Vec<(ArcName, Arc<dyn AnyClassDef>)>,
}

/// A class definition of any model, which is how a [`ServerBuilder`] holds them.
trait AnyClassDef: Send + Sync {
    fn new_class(self: Arc<Self>, server: Arc<Server>, index: usize) -> Arc<dyn AnyClass>;
}

impl<M: Model> AnyClassDef for ClassDef<M> {
    fn new_class(self: Arc<Self>, server: Arc<Server>, index: usize) -> Arc<dyn AnyClass> {
        Class::new(server, self, index)
    }
}

impl ServerBuilder {

    /// Build a server named `exe/inst`, after the executable running it and the name of
    /// the instance of that executable, which is how it is registered in the database.
    pub fn new(exe: &str, inst: &str) -> Self {
        Self {
            exe: exe.to_string(),
            inst: inst.to_string(),
            backend: None,
            database: None,
            config: BackendServerConfig::default(),
            local: true,
            stop_on_signal: true,
            test_context: None,
            classes: Vec::new(),
        }
    }

    /// The backend to serve with, the default one if none is given, see
    /// [`default_backend`].
    pub fn backend(mut self, backend: Arc<dyn Backend>) -> Self {
        self.backend = Some(backend);
        self
    }

    /// The database to register in, the one of the environment if none is given.
    pub fn database(mut self, database: DatabaseProxy) -> Self {
        self.database = Some(database);
        self
    }

    /// The host to listen at, see [`BackendServerConfig::host`].
    pub fn host(mut self, host: &str) -> Self {
        self.config.host = Some(host.to_string());
        self
    }

    /// The port to listen at, see [`BackendServerConfig::port`].
    pub fn port(mut self, port: u16) -> Self {
        self.config.port = Some(port);
        self
    }

    /// Enable or disable registration of the this server's devices into the local 
    /// registry, which allow [`DevProxy`] created in this same process to access the
    /// the device directly without going through the backend's wire protocol.
    pub fn local(mut self, local: bool) -> Self {
        self.local = local;
        self
    }

    /// Define a class in that server, whose devices the database can then list.
    pub fn class<M: Model>(mut self, def: ClassDef<M>) -> Result<Self> {
        
        if def.name.eq_ignore_ascii_case(admin::CLASS_NAME) {
            return Err(Error::new(ErrorKind::ClassAlreadyDefined)
                .with_append_description(format_args!("Class {:?} is reserved to the admin device", def.name)));
        }
        
        if self.classes.iter().any(|(name, _)| **name == *Name::new(&def.name)) {
            return Err(Error::new(ErrorKind::ClassAlreadyDefined)
                .with_append_description(format_args!("Class {:?}", def.name)));
        }

        self.classes.push((ArcName::from(Arc::from(&*def.name)), Arc::new(def)));
        Ok(self)

    }

    /// Stop the server on the signals the process is asked to terminate with, which is the
    /// default: Ctrl-C, and SIGTERM and SIGHUP on Unix, the console being closed on
    /// Windows. The server is then stopped as the `Kill` command of its admin device does,
    /// see [`Server::stop`], and the thread waiting on it deletes its devices, see
    /// [`ServerHandle::wait`].
    ///
    /// There is a single handler of these signals per process, which serving fails to
    /// install if the application installed one of its own: this is then disabled, and
    /// the application stops the server itself.
    pub fn stop_on_signal(mut self, stop_on_signal: bool) -> Self {
        self.stop_on_signal = stop_on_signal;
        self
    }

    /// Start serving: the backend is acquired, every class is made from its definition,
    /// and the admin device is added and synchronized, which boots the whole server.
    /// 
    /// This fails if the admin device fails to initialize, which is what a server that
    /// cannot reach its database does: it would have nothing to serve.
    ///
    /// This blocks, see [`Self::serve_async`] for what does not.
    pub fn serve(self) -> Result<ServerHandle> {
        block_on(self.serve_async())
    }

    /// Serve until the server is stopped, by a signal or the `Kill` command of its admin
    /// device, then delete its devices: this is [`Self::serve`] then [`ServerHandle::wait`],
    /// which is all the main thread of most servers does.
    pub fn run(self) -> Result<()> {
        block_on(self.run_async())
    }

    /// Serve until the server is stopped, see [`Self::run`].
    pub async fn run_async(self) -> Result<()> {
        self.serve_async().await?.wait_async().await
    }

    /// Start serving, see [`Self::serve`], resolving once the whole server is booted.
    pub async fn serve_async(self) -> Result<ServerHandle> {

        // Before anything is served, which a handler that cannot be installed then spares.
        if self.stop_on_signal {
            signal::install()?;
        }

        let backend = self.backend.unwrap_or_else(default_backend);
        let database = self.database
            .unwrap_or_else(|| DatabaseProxy::new_from_env_with_backend(Arc::clone(&backend)));

        // The backend names the events of the server after the server and its database,
        // which is therefore reached before anything is served.
        let admin_dev_name = Arc::<str>::from(format!("{ADMIN_DEV_PREFIX}{}/{}", self.exe, self.inst));
        let (tango_host, tango_port) = database.host_and_port_async().await?;
        let config = self.config.clone()
            .with_admin_name(Arc::clone(&admin_dev_name))
            .with_tango_host(tango_host, tango_port);

        let backend = backend.serve(&config).await?;

        let server = Arc::new_cyclic(|this| Server {
            this: this.clone(),
            admin_dev_name,
            database,
            backend: Arc::downgrade(&backend),
            classes: Mutex::new(HashMap::new()),
            class_count: AtomicUsize::new(1 + self.classes.len()),
            devs: Mutex::new(HashMap::new()),
            pending: Mutex::new(Pending::default()),
            sync_lock: async_lock::Mutex::new(()),
            stopped: AtomicBool::new(false),
            events: Mutex::new(HashMap::new()),
            local: self.local,
        });

        // The admin class is made first, its device being the one booting the others, and
        // is kept typed, which is what its device is added through.
        let admin = Class::new(Arc::clone(&server), Arc::new(Admin::def()), 0);
        let mut classes = HashMap::new();
        classes.insert(ArcName::from(Arc::from(admin::CLASS_NAME)), Arc::clone(&admin) as Arc<dyn AnyClass>);
        for (index, (name, def)) in self.classes.into_iter().enumerate() {
            classes.insert(name, def.new_class(Arc::clone(&server), 1 + index));
        }
        *server.classes.lock().unwrap() = classes;

        // From here the server is in a cycle, which the handle breaks if anything fails.
        let handle = ServerHandle { server, backend, _test_context: self.test_context };
        let admin_dev = admin.add_dev(handle.server.admin_dev_name())?;

        let booted = match handle.server.sync().await {
            Ok(()) => admin_dev.check_model().await,
            Err(err) => Err(err),
        };

        if let Err(err) = booted {
            // The admin device may have been exported in FAULT, which is undone here at
            // best: a server failing to boot usually has no database to tell anything.
            let _ = handle.server.stop().await;
            let _ = handle.wait_async().await;
            return Err(err);
        }

        drop(admin_dev);

        if self.stop_on_signal {
            signal::register(&handle.server);
        }

        info!(server = %handle.server.name(), host = %handle.backend.host(), port = handle.backend.port(),
            devs = handle.server.devs.lock().unwrap().len() - 1, "Serving");
        Ok(handle)

    }

}

/// The handle to a server being served, which is what owns it on behalf of its caller.
///
/// A server ends by being stopped, see [`Server::stop`], by its owner or by a client
/// through the `Kill` command of its admin device, and by its handle being waited for,
/// see [`Self::wait`], which is what deletes its devices. A handle is always waited for:
/// dropping one drops the server as it stands, with no side effect, see [`Drop`].
#[derive(Debug)]
pub struct ServerHandle {
    server: Arc<Server>,
    /// The backend serving the server, which only the handle owns, see [`Server::backend`].
    /// Note that this backend server is co-owning the [`Dev`] instances and so it also
    /// own the server in the field above.
    backend: Arc<dyn BackendServer>,
    /// The database for testing the server is registered in, if any, which is dropped
    /// after it, the server being unregistered from it once it is waited for.
    _test_context: Option<TestContext>,
}

impl ServerHandle {

    #[inline]
    pub fn server(&self) -> &Server {
        &self.server
    }

    /// The server, shared, which does not keep it from being stopped and cleared with this
    /// handle, see [`Drop`].
    pub(crate) fn server_arc(&self) -> Arc<Server> {
        Arc::clone(&self.server)
    }

    /// Wait for the server to be stopped, see [`Server::stop`], and its devices deleted,
    /// which is what the thread running a server spends its life in. The first error met
    /// deleting a model is given back.
    ///
    /// This blocks, see [`Self::wait_async`] for what does not, and [`Self::wait_timeout`]
    /// for a wait that gives up.
    pub fn wait(&self) -> Result<()> {
        block_on(self.wait_async())
    }

    /// Wait for the server to be stopped and its devices deleted, see [`Self::wait`], giving
    /// up with [`ErrorKind::TimedOut`] if the backend is still answering requests once the
    /// timeout has run out, which a request that never ends makes it.
    ///
    /// The timeout only bounds the backend: once it holds nothing of the devices anymore,
    /// their models are all deleted, however long that takes. A wait that gave up deletes
    /// nothing, a request still running being one that may still use its model, and the
    /// handle is then waited for again, or dropped, which ends everything at once, see
    /// [`Drop`].
    pub fn wait_timeout(&self, timeout: Duration) -> Result<()> {
        block_on_timeout(self.backend.wait(), timeout)
            .map_err(|err| err.with_append_description(format_args!("Server {} is still answering requests", self.server.name())))?;
        block_on(self.server.delete_devs())
    }

    /// Wait for the server to be stopped and its devices deleted, see [`Self::wait`].
    ///
    /// The backend is waited for first, after which no request is in flight and it holds
    /// nothing of the devices anymore, see [`BackendServer::wait`], then the model of every
    /// device is deleted, in the reverse order of their classes, the admin device last.
    /// This never runs on the executor of the backend, which it would wait for.
    pub async fn wait_async(&self) -> Result<()> {
        self.backend.wait().await;
        let deleted = self.server.delete_devs().await;
        info!(server = %self.server.name(), "Stopped the server");
        deleted
    }

}

impl Drop for ServerHandle {

    /// Break the ownership cycles of the server, see [`Server::clear`], and drop its
    /// backend, with no side effect: a server that was not waited for, see
    /// [`Self::wait`], is not unexported, the requests in flight are not answered, and the
    /// models are not deleted, see [`BackendServer`].
    fn drop(&mut self) {
        self.server.clear();
    }

}

#[cfg(test)]
mod tests {

    use futures::executor::block_on;
    use futures::FutureExt;

    use crate::dev::AttrQuality;
    use crate::testing::TestContext;

    use super::super::def::ClassDef;
    use super::*;


    /// A server with a device `a/b/c` of a class `A`, served and synchronized.
    fn serve() -> (TestContext, Arc<Server>) {

        let ctx = TestContext::new().unwrap();
        ctx.database().add_server("Test/test", &[("a/b/c", "A")]).unwrap();

        let server = ctx.serve(ServerBuilder::new("Test", "test")
            .class(ClassDef::<()>::new_default("A")).unwrap())
            .unwrap();

        (ctx, server)

    }

    fn read(value: f64) -> AttrRead<Value> {
        AttrRead {
            value: Some(Value::from(value)),
            w_value: None,
            quality: AttrQuality::Valid,
            time: std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1),
        }
    }

    /// The next item of a stream, if there is one already.
    fn now<T>(stream: &mut BoxStream<'static, T>) -> Option<Option<T>> {
        stream.next().now_or_never()
    }

    #[test]
    fn latest_only() {

        let (_ctx, server) = serve();
        let mut sub = server.subscribe_attr_value("a/b/c", "Attr", AttrValueEvent::Change);

        assert!(now(&mut sub).is_none());

        server.push_attr_value("a/b/c", "attr", AttrValueEvent::Change, Ok(read(1.0)));
        server.push_attr_value("A/B/C", "ATTR", AttrValueEvent::Change, Ok(read(2.0)));

        // Another event, or another attribute, reaches nothing.
        server.push_attr_value("a/b/c", "attr", AttrValueEvent::Archive, Ok(read(3.0)));
        server.push_attr_value("a/b/c", "other", AttrValueEvent::Change, Ok(read(4.0)));

        assert_eq!(now(&mut sub).unwrap().unwrap().unwrap(), read(2.0));
        assert!(now(&mut sub).is_none());

    }

    #[test]
    fn every_subscriber() {

        let (_ctx, server) = serve();
        let mut first = server.subscribe_attr_value("a/b/c", "attr", AttrValueEvent::Change);
        let mut second = server.subscribe_attr_value("a/b/c", "attr", AttrValueEvent::Change);

        server.push_attr_value("a/b/c", "attr", AttrValueEvent::Change, Err(Error::new(ErrorKind::Reason(reason::API_AttrNotAllowed))));

        for sub in [&mut first, &mut second] {
            let err = now(sub).unwrap().unwrap().unwrap_err();
            assert_eq!(err.kind(), ErrorKind::Reason(reason::API_AttrNotAllowed));
        }

    }

    #[test]
    fn unsubscribe() {

        let (_ctx, server) = serve();
        let sub = server.subscribe_attr_ready("a/b/c", "attr");
        drop(sub);

        // The push finds nothing to hand the event to, and forgets the attribute.
        server.push_attr_ready("a/b/c", "attr", Ok(()));
        assert!(server.events.lock().unwrap().is_empty());

    }

    #[test]
    fn restart() {

        let (_ctx, server) = serve();
        let mut sub = server.subscribe_attr_conf("a/b/c", "attr");

        // Removed and added again under the same name, as a restart does.
        assert!(server.remove_dev("a/b/c"));
        server.add_dev("A", "a/b/c").unwrap();
        server.sync_blocking().unwrap();
        assert!(now(&mut sub).is_none());

        // Removed for good.
        assert!(server.remove_dev("a/b/c"));
        server.sync_blocking().unwrap();
        assert_eq!(now(&mut sub).unwrap().unwrap().unwrap_err().kind(), ErrorKind::DevNotConnected);
        assert!(now(&mut sub).unwrap().is_none());

    }

    #[test]
    fn stop() {

        let (_ctx, server) = serve();
        let mut sub = server.subscribe_attr_ready("a/b/c", "attr");

        // The event pushed before the end is taken first.
        server.push_attr_ready("a/b/c", "attr", Ok(()));
        block_on(server.stop()).unwrap();

        assert!(now(&mut sub).unwrap().unwrap().is_ok());
        assert_eq!(now(&mut sub).unwrap().unwrap().unwrap_err().kind(), ErrorKind::DevNotConnected);
        assert!(now(&mut sub).unwrap().is_none());

    }

    #[test]
    fn weak() {

        let (_ctx, server) = serve();
        let weak = Ref::downgrade(&server.get_dev("a/b/c").unwrap());
        assert_eq!(weak.upgrade().unwrap().name(), "a/b/c");

        // Restarted under the same name, the weak reference is to the device that was.
        assert!(server.remove_dev("a/b/c"));
        server.add_dev("A", "a/b/c").unwrap();
        server.sync_blocking().unwrap();
        assert!(weak.upgrade().is_none());

    }

    #[test]
    fn dropped() {

        let (ctx, server) = serve();
        let mut sub = server.subscribe_attr_ready("a/b/c", "attr");
        drop(ctx);

        assert_eq!(block_on(sub.next()).unwrap().unwrap_err().kind(), ErrorKind::DevNotConnected);

    }

}