tango/server/

server.rs

//! The server itself: what owns the classes, the devices and the backend serving them.
//!
//! Ownership is a tree rooted at the [`ServerHandle`], which alone owns the `Core` of its
//! server: its classes, its devices and its backend. The [`Server`] the classes and the
//! devices hold is only the context they share, which reaches its core weakly, so that a
//! handler reaches its class and its server as plain field access, without any cycle.

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::ops::{Deref, RangeBounds};
use std::pin::{pin, Pin};
use std::time::Duration;
use std::borrow::Cow;
use std::mem;

use futures::stream::{BoxStream, Stream, StreamExt};
use futures::future::{self, Either};
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, AttrReady, AttrValueEvent};
use crate::error::{Error, ErrorKind, Result, reason};
use crate::name::{self, ArcName, Name};
use crate::future::block_on_timeout;
use crate::testing::TestContext;
use crate::value::Value;
use crate::future::Slot;

use super::class::{AnyClass, Class};
use super::admin::{self, Admin};
use super::def::ClassDef;
use super::Model;
use super::dev::{AnyDev, Dev};
use super::polling::Poller;
use super::terminate;

#[cfg(feature = "cli")]
use super::cli::{self, ServerArgs};


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

/// How long a server is given to boot by default, see [`ServerBuilder::boot_timeout`].
const DEFAULT_BOOT_TIMEOUT: Duration = Duration::from_secs(30);

/// The context shared by the classes and the devices of a server, which every class of it
/// owns, and which reaches what owns them, its `Core`, weakly.
///
/// # Ownership
///
/// ```text
///                  ┌───────────────────────────┐                                     ┌─────────────────┐
///                  │        ServerHandle       │                                     │ DevProxy        │
///                  └─────────────┬─────────────┘                                     └─────┬───────────┘
///                                │                                                         │
///                                │core                                                     │backend: BackendDev
///                                │1                                                        │0..1
///          ┌─────────────────────┴─────────────────────┐      1┌───────────────────┐       │
///          │ Core                                      ├───────┤ dyn BackendServer │       │
///          └─┬──────┬──┬───┬──────┬────┬─────┬─────────┘backend└─────────┬──────┬──┘       │
///            │ name │  │   │ name │    │     ╎1                          │ name │          │
///            ╰──┬───╯  │   ╰──┬───╯    │     ╎core                       ╰──┬───╯          │
///               │      │      │  server│     ╎                              │              │
///               │      │      │       1│     ╎                              │              │
///               │      │      │    ┌───┴─────┴───────┐                      │              │
///               │      │      │    │ Server          │                      │              │
///               │      │      │    └───────────────┬─┘                      │              │
///               │      │      │classes: AnyClass   │1                       │              │
///               │      │      │0..1                │server                  │              │
///               │      │ ┌────┴────────────────────┴───────┐                │              │
///               │      │ │ Class<M>                        │                │              │
///               │      │ └─────────────────────────┬───────┘                │              │
///               │      │                           │1                       │              │
///               │      │                           │class                   │              │
///   devs: AnyDev│      │pending: AnyDev            │        devs: BackendDev│              │
///           0..1│      │*                          │                    0..1│              │
///         ┌─────┴──────┴───────────────────────────┴────────────────────────┴──────────────┴─────────┐
///         │ Dev<M>                                                                                   │
///         └─┬──────┬─┬──────┬──┬────────────┬─────────────┬─────────────────────┬───────────┬────────┘
///           │ name │ │ name │  │            │             │                 0..1╎           ╎0..1
///           ╰──┬───╯ ╰─┬────╯  │            │             │      dev: BackendDev╎           ╎dev: BackendDev
///              │       │       │            │             │                     ╎           ╎
///      commands│       │attrs  │state_attr  │status_attr  │                     ╎           ╎
///          0..1│       │0..1   │1           │1            │                     ╎           ╎
///  ┌───────────┴┐  ┌───┴───────┴────────────┴────────┐    │                     ╎           ╎
///  │ Command<M> │  │ Attr<M>                         │    │                     ╎           ╎
///  └────────────┘  └─────────────────────────────────┘    │local_handle     ╭───┴───╮ ╭─────┴──────╮
///                                                         │0..1             │  itr  │ │ port, name │
///                                                 ┌───────┴────────┐     1┌─┴───────┴─┴────────────┴─┐
///                                                 │ LocalDevHandle ├╌╌╌╌╌╌┤      local registry      │
///                                                 └────────────────┘  keys└──────────────────────────┘
/// ```
///
/// A solid line is an [`Arc`], a dashed one a [`Weak`]. An end of a line has a
/// multiplicity, and the name of the field holding it, only when the class at the other end
/// reaches it: an end without one is not reached through that line. A field holding the
/// other end as a trait object is written `field: Trait`. A rounded box is the key of a map
/// of the class it hangs from, which holds at most one of the other end per key.
/// `LocalDevHandle` holds the keys of the entries of its device, which it takes out of the
/// registry when dropped, and a weak reference to the device, not drawn, which only tells
/// its entries from those of a device registered since.
///
/// The backend is given each device as a `dyn BackendDev`, which is the `DevImpl<M>`
/// wrapper of a [`Dev<M>`]. A [`DevProxy`](crate::dev::DevProxy) holds the backend it
/// reaches its device through as a `dyn BackendDev` too, which is the device itself when it
/// is served by this process and found in the registry, and a client of the protocol of the
/// backend otherwise.
///
/// The commands and attributes of a device, kept in its `mutable` field, each hold their
/// definition, not drawn, which is shared with the [`ClassDef`] of its class.
///
/// Every solid line points down from the handle: the only links going back up are the weak
/// `core` field and those of the local registry. Dropping the handle stops the backend, and
/// frees everything once nothing else holds the core, which a request running a
/// [`Self::sync_async`] or a [`Self::stop_async`] does until it ends, see [`ServerHandle`].
///
/// The local registry only ever holds devices weakly, so it never keeps one alive, while a
/// client of this process holds the device it found for as long as it is connected to it,
/// which refuses its requests once withdrawn, see [`Dev::is_withdrawn`]. A device is taken
/// out of the registry, by dropping its `LocalDevHandle`, before it is removed from the
/// backend, so that local clients stop finding it first.
#[derive(Debug)]
pub struct Server {
    /// What owns the classes, the devices and the backend of that server, which is only
    /// there for as long as it is served, see [`Self::core`].
    core: Weak<Core>,
    /// 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 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 an [`ErrorKind::DevNotConnected`], once the device
    /// is gone for good, see [`Self::sync_async`], or the server stops, see
    /// [`Self::stop_async`]: 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 owner of the classes, the devices and the backend of a server, which only its
/// [`ServerHandle`] holds and which is never handed out: the [`Server`] reaches it weakly,
/// see [`Server::core`], which is what keeps the ownership of a server a tree.
///
/// The methods of the server reaching what it owns hold it for the time of their call,
/// across the whole of it for [`Server::sync_async`], [`Server::stop_async`] and
/// [`Server::wait_async`], so that a call started is finished before the devices go.
#[derive(Debug)]
pub(super) struct Core {
    /// The server, which the handle derefs to.
    server: Arc<Server>,
    /// The backend serving the devices of that server.
    backend: Arc<dyn BackendServer>,
    /// The classes of that server.
    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, 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 [`Server::sync_async`] applies, which is never locked with any other lock.
    pub(super) pending: Mutex<Pending>,
    /// Held for the whole of a [`Server::sync_async`], so that two of them never interleave, and
    /// for the whole of a [`Server::stop_async`], which [`Server::wait`] waits for through it.
    sync_lock: async_lock::Mutex<()>,
    /// Whether that server was stopped, which is only ever done once.
    stopped: AtomicBool,
    /// The polling of the devices of that server, see [`Poller`], which ends with it.
    poller: Poller,
}

impl Drop for Core {

    /// Stop the backend, which the handle did already unless this outlived it: a backend
    /// stops for good once nothing holds its server anymore, whatever it was serving.
    ///
    /// Every device left is marked withdrawn and taken out of the local registry, which a
    /// stopped server has none of, see [`Server::stop_async`]: a server that ended without
    /// being stopped never withdraws them, while its devices may still be held, by a thread
    /// of their own or a client of this process in particular.
    fn drop(&mut self) {
        self.backend.stop();
        self.poller.shutdown();
        let removed = &self.pending.get_mut().unwrap().removed;
        for dev in self.devs.get_mut().unwrap().values().chain(removed) {
            dev.withdraw_now();
        }
    }

}

/// The subscribers of the events of an attribute, see [`Server::events`].
#[derive(Debug, Default)]
struct AttrEvents {
    values: HashMap<AttrValueEvent, Subscribers<AttrRead<Value>>>,
    ready: Subscribers<AttrReady>,
    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_async`].
#[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()..]
    }

    /// The name of the admin device of that server, `dserver/exe/inst`.
    #[inline]
    pub fn admin_dev_name(&self) -> &str {
        &self.admin_dev_name
    }

    /// The database that server is registered in.
    #[inline]
    pub fn database(&self) -> &DatabaseProxy {
        &self.database
    }

    /// What owns the classes, the devices and the backend of that server, failing once it
    /// is gone, which is once its handle is dropped and nothing else holds it, see
    /// [`ServerHandle`].
    pub(super) fn core_checked(&self) -> Result<Arc<Core>> {
        self.core.upgrade().ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_ServerNotRunning))
            .with_description(format!("Server {:?} is not served anymore", self.name())))
    }

    /// The polling of the devices of that server, absent once it is not served anymore,
    /// see [`Self::core`].
    pub(super) fn poller(&self) -> Option<Poller> {
        self.core.upgrade().map(|core| core.poller.clone())
    }

    /// The backend serving the devices of that server, failing once it is not served
    /// anymore, see [`Self::core`].
    pub(super) fn backend_checked(&self) -> Result<Arc<dyn BackendServer>> {
        Ok(Arc::clone(&self.core_checked()?.backend))
    }

    /// The backend serving the devices of that server, [`None`] once it is not served
    /// anymore, see [`ServerHandle`].
    #[inline]
    pub fn backend(&self) -> Option<Arc<dyn BackendServer>> {
        self.core.upgrade().map(|core| Arc::clone(&core.backend))
    }

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

        if !name::check_tango_class_name(&def.name) {
            return Err(Error::new(ErrorKind::ClassNameInvalid)
                .with_description(format!("Class {:?} has an invalid name, it should be made of \
                    letters, digits, '_' and '-'", def.name)));
        }

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

        let Entry::Vacant(entry) = classes.entry(name) else {
            return Err(Error::new(ErrorKind::ClassAlreadyDefined)
                .with_description(format!("Class {:?} is already defined on the server", def.name)));
        };

        let index = core.class_count.fetch_add(1, Ordering::Relaxed);
        let class = Class::new(Arc::clone(&core.server), def, index);
        entry.insert(Arc::clone(&class) as Arc<dyn AnyClass>);
        Ok(class)

    }

    /// Get a class of this server from its name, if it is one of that model.
    pub fn get_class<M: Model>(&self, name: &str) -> Option<Arc<Class<M>>> {
        self.any_class(name)?.into_any().downcast().ok()
    }

    /// Get a class of this server from its name, whatever its model.
    pub(super) fn any_class(&self, name: &str) -> Option<Arc<dyn AnyClass>> {
        let core = self.core.upgrade()?;
        core.classes.lock().unwrap().get(Name::new(name)).map(Arc::clone)
    }

    /// 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_async`].
    pub fn add_dev(&self, class: &str, name: &str) -> Result<()> {

        let class = self.core_checked()?.classes.lock().unwrap()
            .get(Name::new(class))
            .map(Arc::clone)
            .ok_or_else(|| Error::new(ErrorKind::ClassNotFound)
                .with_description(format!("Class {class:?} is not defined on the server")))?;

        class.add_dev(name).map(drop)

    }

    /// 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_async`]. A device removed before any
    /// synchronization reached it is simply dropped.
    pub fn remove_dev(&self, name: &str) -> bool {

        let Some(core) = self.core.upgrade() else {
            return false;
        };

        let Some(dev) = core.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 = core.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_async(&self) -> Result<()> {

        let Some(core) = self.core.upgrade() else {
            return Ok(());
        };

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

        let mut first_err = None;

        // Begin with pending removal of devices, they are first unexported and then
        // withdrawn from the backend.
        let removed = mem::take(&mut core.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 !core.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 tells that they should be exported next!
        loop {

            let mut added = mem::take(&mut core.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, its init function panicked");
                }
                core.pending.lock().unwrap().unexported.push((dev, None));
            }

        }

        // Finally we process all the unexported devices by exporting them to the database.
        let unexported = mem::take(&mut core.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(&*core.backend, &*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");

        }

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

    }

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

    /// Export a device served at that reference in the database, which is what a client
    /// looks it up in.
    async fn export_dev(&self, backend: &dyn BackendServer, dev: &dyn AnyDev, itr: &str) -> Result<()> {
        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
    }

    /// The names of every class of this server, in the order they were made, the admin
    /// one first.
    pub fn class_names(&self) -> Vec<Arc<str>> {
        self.any_classes().iter().map(|class| Arc::clone(class.name())).collect()
    }

    /// Every class of this server, whatever its model, see [`Self::class_names`].
    pub(super) fn any_classes(&self) -> Vec<Arc<dyn AnyClass>> {
        let Some(core) = self.core.upgrade() else {
            return Vec::new();
        };
        let mut classes = core.classes.lock().unwrap().values().map(Arc::clone).collect::<Vec<_>>();
        classes.sort_by_key(|class| class.index());
        classes
    }

    /// The names of every device of this server, class by class in the order of
    /// [`Self::class_names`], and by name within a class.
    pub fn dev_names(&self) -> Vec<Arc<str>> {
        self.any_devs().iter().map(|dev| Arc::clone(dev.name())).collect()
    }

    /// Every device of this server, whatever its model, see [`Self::dev_names`].
    pub(super) fn any_devs(&self) -> Vec<Arc<dyn AnyDev>> {
        let Some(core) = self.core.upgrade() else {
            return Vec::new();
        };
        let mut devs = core.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
    }

    /// Whether this server has a device of that name.
    pub fn has_dev(&self, name: &str) -> bool {
        self.core.upgrade().is_some_and(|core| core.devs.lock().unwrap().contains_key(Name::new(name)))
    }

    /// Get a device of this server from its name, if its class is one of that model.
    pub fn get_dev<M: Model>(&self, name: &str) -> Option<Arc<Dev<M>>> {
        self.any_dev(name)?.into_any().downcast().ok()
    }

    /// Get a device of this server from its name, whatever its model.
    pub(super) fn any_dev(&self, name: &str) -> Option<Arc<dyn AnyDev>> {
        let core = self.core.upgrade()?;
        core.devs.lock().unwrap().get(Name::new(name)).map(Arc::clone)
    }

    /// 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, and its devices are deleted, see [`Self::wait`] for what waits
    /// for the backend to be through. No synchronization runs after this, the devices
    /// still waiting for one never being served.
    ///
    /// The devices are deleted one after the other, the classes in the reverse of their
    /// order, the admin device last, each once the requests still running on it are
    /// through, see [`Dev::is_withdrawn`]: a request that never ends therefore keeps this
    /// from ever returning. This is also why a request handler must never call it, since
    /// it would wait for itself; the `Kill` command of the admin device stops its server
    /// from the pool instead.
    ///
    /// The backend is stopped and the devices deleted even if the server could not be
    /// unexported, and the first error met is given back.
    pub async fn stop_async(&self) -> Result<()> {

        let Some(core) = self.core.upgrade() else {
            return Ok(());
        };

        let _sync = core.sync_lock.lock().await;
        if core.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");
        }

        core.backend.stop();
        core.poller.shutdown();
        self.end_events();

        // The devices removed and not synchronized yet first, as the next synchronization
        // would have, then the others, the classes in the reverse of their order, the admin
        // device last.
        let removed = mem::take(&mut *core.pending.lock().unwrap()).removed;
        let mut devs = mem::take(&mut *core.devs.lock().unwrap())
            .into_values()
            .collect::<Vec<_>>();

        devs.sort_by_key(|dev| std::cmp::Reverse(dev.class().index()));

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

        info!(server = %self.name(), "Stopped the server");
        first_err.map_or(Ok(()), Err)

    }

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

    /// Wait for this server to be stopped, see [`Self::stop_async`], which is what the
    /// thread running a server spends its life in: for its backend to be through with
    /// every request, and for its devices to be deleted. A server that is not served
    /// anymore, see [`ServerHandle`], is not waited for.
    ///
    /// The backend is waited for first, which is only through once stopped, see
    /// [`BackendServer::wait`], then the stop itself, which deletes the devices under the
    /// lock of the synchronizations. This waits for every request in flight, and must
    /// therefore never be called from a request handler, which would wait for itself.
    pub async fn wait_async(&self) {
        let Some(core) = self.core.upgrade() else {
            return;
        };
        core.backend.wait().await;
        drop(core.sync_lock.lock().await);
    }

    /// The blocking form of [`Self::wait_async`], which must therefore never run on the
    /// executor of a backend, in an asynchronous handler in particular. See
    /// [`Self::wait_timeout`] for a wait that gives up.
    pub fn wait(&self) {
        block_on(self.wait_async())
    }

    /// Wait for this server to be stopped, see [`Self::wait_async`], 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, or a server never stopped.
    ///
    /// The timeout only bounds the backend: once it is through, the deletion of the
    /// devices is waited for, however long that takes.
    pub fn wait_timeout(&self, timeout: Duration) -> Result<()> {
        let Some(core) = self.core.upgrade() else {
            return Ok(());
        };
        block_on_timeout(core.backend.wait(), timeout)
            .map_err(|err| err.with_description(format!("Server {:?} is still answering requests after {timeout:?}", self.name())))?;
        block_on(core.sync_lock.lock());
        Ok(())
    }

    /// 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<AttrReady>> {
        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_config(&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<AttrReady>) {
        self.push_event(dev, attr, |events| &mut events.ready, ready);
    }

    /// Push the configuration of an attribute of a device.
    pub(super) fn push_attr_config(&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 dev = Arc::<str>::from(dev);
        let mut devs = self.events.lock().unwrap();
        let attrs = devs.entry(ArcName::from(Arc::clone(&dev))).or_default();
        let events = attrs.entry(ArcName::from(attr)).or_default();
        let subscribers = select(events);
        subscribers.retain(|slot| slot.strong_count() != 0);
        subscribers.push(Arc::downgrade(&slot));

        EventStream {
            dev,
            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(event.clone());
            }

            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> {
    /// The name of the device the events are of, which the end of the subscription names.
    dev: Arc<str>,
    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_description(format!("Device {:?} is no longer served", this.dev))))
        }))

    }

}

/// 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_terminate`].
    stop_on_terminate: bool,
    /// See [`Self::terminate_on_signals`].
    #[cfg(feature = "cli")]
    terminate_on_signals: bool,
    /// See [`Self::boot_timeout`].
    boot_timeout: Duration,
    /// The database for testing the server is registered in, which the server owns for
    /// as long as it runs, see [`ServerHandle`].
    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_terminate: false,
            #[cfg(feature = "cli")]
            terminate_on_signals: false,
            boot_timeout: DEFAULT_BOOT_TIMEOUT,
            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 = self.config.host(host);
        self
    }

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

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

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

    /// Enable or disable registration of this server's devices into the local registry,
    /// which allows a [`DevProxy`](crate::dev::DevProxy) created in this same process to
    /// access the device directly without going through the backend's wire protocol.
    /// Enabled by default.
    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.
    ///
    /// # Panics
    ///
    /// If the name of the class is invalid, see [`name::check_tango_class_name`], if it is
    /// the one of the admin class, or if the server already defines a class of that name.
    #[track_caller]
    pub fn class<M: Model>(mut self, def: ClassDef<M>) -> Self {

        if !name::check_tango_class_name(&def.name) {
            panic!("the class name {:?} is invalid", def.name);
        }

        if def.name.eq_ignore_ascii_case(admin::CLASS_NAME) {
            panic!("the class name {:?} is reserved to the admin device", def.name);
        }

        if self.classes.iter().any(|(name, _)| **name == *Name::new(&def.name)) {
            panic!("the class {:?} is already defined", def.name);
        }

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

    }

    /// Stop the server when the process terminates, see [`terminate`], which is not the
    /// default unless built from the command line, see [`Self::terminate_on_signals`]. The
    /// server is then stopped as the `Kill` command of its admin device does, which deletes
    /// its devices, see [`Server::stop_async`], and the thread waiting on it returns, see
    /// [`Server::wait`]. A server still booting gives up its boot, see [`Self::serve`].
    ///
    /// [`terminate`]: super::terminate()
    pub fn stop_on_terminate(mut self, stop_on_terminate: bool) -> Self {
        self.stop_on_terminate = stop_on_terminate;
        self
    }

    /// How long the server is given to boot when served blocking, 30 seconds by default,
    /// see [`Self::serve`]: the database being reached, every device initialized and every
    /// one of them exported. An asynchronous caller bounds [`Self::serve_async`] itself.
    pub fn boot_timeout(mut self, boot_timeout: Duration) -> Self {
        self.boot_timeout = boot_timeout;
        self
    }

    /// Boot the server, see [`Self::serve_async`].
    async fn boot(self) -> Result<ServerHandle> {

        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()
            .admin_name(Arc::clone(&admin_dev_name))
            .tango_host(tango_host, tango_port);

        let backend = backend.serve(&config).await?;
        let poller = Poller::spawn()?;

        let core = Arc::new_cyclic(|core| Core {
            server: Arc::new(Server {
                core: core.clone(),
                admin_dev_name,
                database,
                events: Mutex::new(HashMap::new()),
                local: self.local,
            }),
            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),
            poller,
        });

        // 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(&core.server), Arc::new(Admin::def()), 0);
        let mut classes = HashMap::new();
        classes.insert(ArcName::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(&core.server), 1 + index));
        }
        *core.classes.lock().unwrap() = classes;

        // From here the handle owns the server, which it stops if anything fails.
        let handle = ServerHandle { core, _test_context: self.test_context };
        let admin_dev = admin.add_dev(handle.admin_dev_name())?;

        let booted = match handle.sync_async().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.stop_async().await;
            handle.wait_async().await;
            return Err(err);
        }

        drop(admin_dev);

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

        Ok(handle)

    }

    /// 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, and fails once the server
    /// did not boot within its timeout, see [`Self::boot_timeout`], or once the process
    /// terminates while it boots, see [`Self::stop_on_terminate`]. The boot is then given
    /// up wherever it was, so devices it had exported already stay exported in the
    /// database until the server is served again.
    pub fn serve(self) -> Result<ServerHandle> {
        let boot_timeout = self.boot_timeout;
        let server_name = format!("{}/{}", self.exe, self.inst);
        block_on_timeout(self.serve_async(), boot_timeout)
            .map_err(|err| err.with_description(format!("Server {server_name:?} did not boot within {boot_timeout:?}")))?
    }

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

        let server_name = format!("{}/{}", self.exe, self.inst);
        if !name::check_tango_server_part_name(&self.exe) 
        || !name::check_tango_server_part_name(&self.inst) {
            return Err(Error::new(ErrorKind::ServerNameInvalid)
                .with_description(format!("Server {server_name:?} has an invalid name, it should be \
                    exe/inst, each part made of letters, digits, '_' and '-'")));
        }

        // Before anything is served, which a handler that cannot be installed then spares.
        #[cfg(feature = "cli")]
        if self.terminate_on_signals {
            cli::install_signal_handler()?;
        }

        let stop_on_terminate = self.stop_on_terminate;
        let boot = pin!(self.boot());
        if !stop_on_terminate {
            return boot.await;
        }

        let handle = match future::select(boot, pin!(terminate::terminating())).await {
            Either::Left((booted, _)) => booted?,
            Either::Right(((), _)) => return Err(Error::new(ErrorKind::Aborted)
                .with_description(format!("Server {server_name:?} did not finish booting, the process is terminating"))),
        };

        // The process may have terminated right after the boot, the server being missed.
        if terminate::register(&handle.core.server) {
            let _ = handle.stop_async().await;
        }

        Ok(handle)

    }

    /// Serve until the server is stopped, by [`terminate`](super::terminate()) (on a signal
    /// when [`Self::terminate_on_signals`] is set) or by the `Kill` command of its admin
    /// device, which deletes its devices: this is [`Self::serve`] then [`Server::wait`],
    /// which is all the main thread of most servers does.
    pub fn run(self) -> Result<()> {
        self.serve()?.wait();
        Ok(())
    }

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

}

#[cfg(feature = "cli")]
impl ServerBuilder {

    /// Terminate the process on the signals it is asked to terminate with, see
    /// [`terminate::terminate`], which a server built from the command line does: Ctrl-C, and SIGTERM
    /// and SIGHUP on Unix, the console being closed on Windows. Handling them also stops
    /// the server on termination, see [`Self::stop_on_terminate`].
    ///
    /// 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 calls [`terminate::terminate`] from its own handler.
    pub fn terminate_on_signals(mut self, terminate_on_signals: bool) -> Self {
        self.terminate_on_signals = terminate_on_signals;
        if terminate_on_signals {
            self = self.stop_on_terminate(true);
        }
        self
    }

    /// Build the server from the command line of this process, see [`ServerArgs`], which
    /// is what launching a server usually takes. The process exits with its usage when
    /// the command line is invalid or asks for help.
    ///
    /// See [`Self::from_raw_args`] for what the server is named after.
    pub fn from_args() -> Self {
        Self::from_raw_args(None, None)
    }

    /// Build the server named after that executable and the instance of that command
    /// line, see [`ServerArgs`], which is also where it listens. The logs of the process
    /// are printed at the level it asks for.
    ///
    /// With no executable given, the server is named after the one of this process, as
    /// cppTango names it: the name the process was run under, without its directory nor
    /// the suffix of the platform's executables, which is what the Starter of a host runs
    /// a server by. Giving one keeps the name of the server whatever its file is called.
    ///
    /// With no command line given, the one of this process is parsed, as
    /// [`Self::from_args`] does; a server with options of its own parses it itself and
    /// gives what it flattened, see [`ServerArgs`].
    pub fn from_raw_args(exec: Option<&str>, args: Option<&ServerArgs>) -> Self {

        use std::process;

        use clap::{Args, Command, FromArgMatches};
        use tracing_subscriber::filter::LevelFilter;
        use tracing_subscriber::EnvFilter;

        let exec = exec.map(Cow::Borrowed).unwrap_or_else(|| Cow::Owned(cli::exe()));

        let args = match args {
            Some(args) => Cow::Borrowed(args),
            None => {
                let matches = ServerArgs::augment_args(Command::new(exec.to_string())).get_matches();
                let args = ServerArgs::from_arg_matches(&matches).unwrap_or_else(|err| err.exit());
                Cow::Owned(args)
            }
        };

        // The subscriber printing the logs of the process, at the level asked for unless
        // `RUST_LOG` says otherwise. An application that installed one of its own keeps it.
        let level = match args.verbose {
            0 => LevelFilter::INFO,
            1 => LevelFilter::DEBUG,
            _ => LevelFilter::TRACE,
        };

        let _ = tracing_subscriber::fmt()
            .with_env_filter(EnvFilter::builder()
                .with_default_directive(level.into())
                .from_env_lossy())
            .try_init();

        let mut builder = Self::new(&exec, &args.inst)
            .terminate_on_signals(true);

        if let Some(host) = &args.host {
            builder = builder.host(host);
        }

        if let Some((start, end)) = args.port {
            builder = builder.port(start..=end);
        }

        if let Some((start, end)) = args.zmq_heartbeat_port {
            builder = builder.zmq_heartbeat_port(start..=end);
        }

        if let Some((start, end)) = args.zmq_event_port {
            builder = builder.zmq_event_port(start..=end);
        }

        // The database for testing, served from now on, which the server is registered in
        // and owns: failing to serve it is failing to launch the server at all.
        if !args.devs.is_empty() {
            let ctx = cli::test_context(&exec, &args).unwrap_or_else(|err| {
                eprintln!("Error: {err:#}");
                process::exit(1)
            });
            builder = builder.database(ctx.database().clone());
            builder.test_context = Some(ctx);
        }

        builder

    }

}

/// The handle to a server being served, which is what owns it on behalf of its caller, and
/// derefs to it.
///
/// A server ends by being stopped, see [`Server::stop_async`], which deletes its devices, by its
/// owner or by a client through the `Kill` command of its admin device, and is waited for
/// by its owner, see [`Server::wait`]. A server is always stopped: dropping the handle of a
/// server that was not only stops its backend, with no other side effect, see [`Drop`].
#[derive(Debug)]
pub struct ServerHandle {
    /// What owns the server, which only the handle holds for good.
    core: Arc<Core>,
    /// 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 {

    /// The server this handle serves.
    #[inline]
    pub fn server(&self) -> &Server {
        &self.core.server
    }

    /// The server, shared, which does not keep it served: once the handle is dropped, see
    /// [`Drop`], it is not served anymore.
    pub(crate) fn server_arc(&self) -> Arc<Server> {
        Arc::clone(&self.core.server)
    }

}

impl Deref for ServerHandle {

    type Target = Server;

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

}

impl Drop for ServerHandle {

    /// Stop the backend and end the subscriptions, with no other side effect: a server
    /// that was not stopped, see [`Server::stop_async`], is not unexported, and its models are
    /// not deleted. The requests in flight are not waited for, and everything of the server
    /// is let go of once they end, see [`BackendServer`], a request still holding it for
    /// the time of a [`Server::sync_async`] or a [`Server::stop_async`] it runs.
    fn drop(&mut self) {
        self.core.backend.stop();
        self.core.server.end_events();
    }

}

#[cfg(test)]
mod tests {

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

    use crate::dev::{AttrQuality, DevState};
    use crate::value::AttrDataKind;
    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();

        (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(AttrReady { data_kind: AttrDataKind::Long, counter: 1 }));
        assert!(server.events.lock().unwrap().is_empty());

    }

    #[test]
    fn restart() {

        let (_ctx, server) = serve();
        let mut sub = server.subscribe_attr_config("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().unwrap();
        assert!(now(&mut sub).is_none());

        // Removed for good.
        assert!(server.remove_dev("a/b/c"));
        server.sync().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(AttrReady { data_kind: AttrDataKind::Long, counter: 1 }));
        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 = Arc::downgrade(&server.any_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().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);

    }

    #[test]
    fn freed() {

        let (ctx, server) = serve();
        server.add_dev("A", "a/b/d").unwrap();
        let served = Arc::downgrade(&server.any_dev("a/b/c").unwrap());
        let pending = Arc::downgrade(&server.any_dev("a/b/d").unwrap());
        drop(ctx);

        // Every device is let go of, those still waiting for a synchronization included,
        // while the server itself, still held, is not served anymore.
        assert!(served.upgrade().is_none());
        assert!(pending.upgrade().is_none());
        assert!(!server.has_dev("a/b/c"));
        assert_eq!(server.add_dev("A", "a/b/e").unwrap_err().kind(), ErrorKind::Reason(reason::API_ServerNotRunning));
        server.wait();

    }

    #[test]
    fn withdrawn() {

        let (ctx, server) = serve();
        let dev = server.get_dev::<()>("a/b/c").unwrap();
        assert!(!dev.is_withdrawn());
        dev.push_user_event("state", DevState::On).unwrap();

        // Restarted under the same name, the device held is not the one served anymore.
        assert!(server.remove_dev("a/b/c"));
        server.add_dev("A", "a/b/c").unwrap();
        server.sync().unwrap();
        assert!(dev.is_withdrawn());
        assert_eq!(dev.push_user_event("state", DevState::On).unwrap_err().kind(), ErrorKind::DevNotConnected);

        // Nor is any once the server is not served anymore.
        let restarted = server.get_dev::<()>("a/b/c").unwrap();
        assert!(!restarted.is_withdrawn());
        drop(ctx);
        assert!(restarted.is_withdrawn());

    }

    #[test]
    fn withdrawn_local() {

        let (ctx, server) = serve();
        let client = ctx.dev("a/b/c");
        client.ping().unwrap();
        assert!(ctx.is_local("a/b/c").unwrap());

        // The proxy reaches the device itself, which is kept alive here once removed: it
        // is refused all the same, rather than answered as an uninitialized device, which
        // makes the proxy look it up again on its next request, and find it unexported.
        let dev = server.get_dev::<()>("a/b/c").unwrap();
        assert!(server.remove_dev("a/b/c"));
        server.sync().unwrap();
        assert!(dev.is_withdrawn());
        assert_eq!(client.ping().unwrap_err().kind(), ErrorKind::DevNotConnected);
        assert_eq!(client.state().unwrap_err().kind(), ErrorKind::DevNotExported);

    }

    #[test]
    fn wait_twice() {

        let (_ctx, server) = serve();
        server.stop().unwrap();
        server.wait();
        server.wait();
        assert!(server.dev_names().is_empty());

    }

}