rstango 0.4.0

//! Partial implementation of the server side of CORBA, which mirrors the client side of
//! the very same protocol, see [`super::client`].
//!
//! One polling thread does everything: it accepts the connections, receives the requests,
//! and drives the futures the objects answer them with, see [`Thread::main`]. There is no worker
//! pool and no thread per request: an object that only forwards calls to lower level ones
//! costs nothing but the memory of its future while it waits, so a high level object may
//! fan out as many calls as it likes without holding a thread for any of them.
//!
//! What that asks of an object is that it never blocks the loop, see
//! [`ServerObject::request`]: one that has to hands the blocking part to a thread of its own
//! and awaits its answer.

use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, ToSocketAddrs, UdpSocket};
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::ops::{ControlFlow, Deref, DerefMut, Range};
use std::{env, fs, mem, task, thread};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::borrow::Cow;
use std::pin::Pin;
use std::io;

use futures::stream::{FuturesUnordered, StreamExt};
use futures::future::poll_fn;
use mio::{Events, Interest, Poll, Token, Waker};
use mio::net::{TcpListener, TcpStream};
use pin_project_lite::pin_project;
use tracing::{debug, error, trace, warn};
use futures::task::ArcWake;

use crate::error::{Error, ErrorKind, Result};
use crate::buf::{Buf, BufMut};

use super::ior::{AltIiopAddr, CodesetsComponent, ComponentKind, IiopProfile, Ior, ProfileKind};
use super::codeset::{AnyCodeset, Codeset, Codesets, WideCodeset};
use super::{MAX_MESSAGE_LEN, EVENTS_CAPACITY, OBJECT_REPO_ID};
use super::cdr::{CdrReader, CdrWrite, CdrWriter, CdrSender};
use super::io::{WriteExt as _, ReadExt as _};
use super::endian::Endian;
use super::giop::{
    LocateReply, LocateStatus, Message, MessageLen, MessageWhole, Reply, ReplyStatus, Target,
    SystemException, CompletionStatus,
    Version, exception};


/// The token of the listener, which is the only source that accepts.
const LISTENER_TOKEN: Token = Token(0);
/// The token of the waker, which is how a task that became ready on another thread brings
/// the polling loop back to it, see [`PollWaker`].
const WAKER_TOKEN: Token = Token(1);
/// The token the connections start at, the two sources that are not connections taking the
/// ones below it.
const FIRST_CONNECTION_TOKEN: usize = 2;

/// The server is serving, see [`Shared::state`], which is where it starts.
const RUN: u8 = 0;
/// The server was stopped and its polling thread drains it, see [`Server::stop`].
const STOP: u8 = 1;
/// The handle to the server was let go of, and its polling thread drops everything of it
/// as it stands, see [`Drop`].
const DROP: u8 = 2;
/// The polling thread is over, and nothing of any object is held anymore, see
/// [`Server::wait`].
const DONE: u8 = 3;

/// The latest GIOP version this server speaks, which is what it names to a client whose
/// message it could make nothing of, see section 9.4.8 of the specification.
const VERSION: Version = Version::V12;
/// The room a reply is started with, which is enough that the usual ones are written into
/// the one allocation they were made with: a GIOP header and a short string is what most
/// of the replies of this protocol are, and so is the answer of a single-attribute
/// operation. It is the same length, for the same reason, as what a read is offered room
/// for, see [`ReadExt::read_nb_to_cdr`](super::io::ReadExt::read_nb_to_cdr).
const REPLY_MIN_LEN: usize = 512;


/// A partial CORBA server implementation for Tango needs.
///
/// Binding one starts the thread that serves it, see [`Thread::main`], so an object answers
/// from the moment it is registered and nothing has to be run for it to. This is the handle
/// to that thread: stopping it is done through it, see [`Self::stop`], and letting go of it
/// drops everything at once, see [`Drop`].
#[derive(Debug)]
pub struct Server<O: ServerObject> {
    /// What the polling thread and the callers registering objects both reach.
    shared: Arc<Shared<O>>,
}

impl<O: ServerObject> Server<O> {

    /// Bind a server to the given address and start the thread that serves it, which is
    /// listening and answering by the time this comes back, see [`Thread::main`].
    ///
    /// A port of zero is the system choosing one, which is why the address is read back here
    /// rather than kept as it was given, see [`Self::local_addr`].
    pub fn bind(ip: IpAddr, ports: (u16, u16)) -> Result<Self> {
        
        let mut listener = 
            if ports.0 == 0 {

                TcpListener::bind(SocketAddr::new(ip, 0))
                    .map_err(|err| Error::new(ErrorKind::Corba)
                        .with_description(format!("Failed to bind the CORBA server to {ip}"))
                        .with_source(Box::new(err)))?
                
            } else {

                let res = super::super::try_port_range(ports, |port| {
                    match TcpListener::bind(SocketAddr::new(ip, port)) {
                        Ok(res) => ControlFlow::Break(Ok(res)),
                        Err(err) if let io::ErrorKind::AddrInUse = err.kind() => ControlFlow::Continue(()),
                        Err(err) => ControlFlow::Break(Err(Error::new(ErrorKind::Corba)
                            .with_description(format!("Failed to bind the CORBA server to {ip}:{port}"))
                            .with_source(Box::new(err)))),
                    }
                });

                match res {
                    ControlFlow::Break(Ok(ret)) => ret,
                    ControlFlow::Break(Err(err)) => return Err(err),
                    ControlFlow::Continue(()) => return Err(Error::new(ErrorKind::Corba)
                        .with_description(format!("Failed to bind the CORBA server to {ip}, every port being in use"))
                        .with_append_description(format_args!("Tried: ports {}..={}", ports.0, ports.1))),
                }

            };

        let local_addr = listener.local_addr()
            .map_err(|err| Error::new(ErrorKind::Corba)
                .with_description(format!("Failed to read back the address of the listener bound to {ip}"))
                .with_source(Box::new(err)))?;

        let poll = Poll::new()
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to create the poll of a CORBA server")
                .with_source(Box::new(err)))?;

        poll.registry().register(&mut listener, LISTENER_TOKEN, Interest::READABLE)
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to register the listener of a CORBA server with the poll")
                .with_source(Box::new(err)))?;

        // Made here and not on the first task that needs one: a waker is tied to the poll it
        // brings back, and by the time a task is woken the loop is blocked in that poll.
        let waker = Arc::new(Waker::new(poll.registry(), WAKER_TOKEN)
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to create the waker of a CORBA server")
                .with_source(Box::new(err)))?);

        let addrs = local_addrs(local_addr);
        let host = Arc::from(addrs[0].to_string());

        let shared = Arc::new(Shared {
            objects: Mutex::new(HashMap::new()),
            local_addr,
            addrs,
            host,
            state: AtomicU8::new(RUN),
            waits: Mutex::new(Vec::new()),
            waker: Arc::clone(&waker),
        });

        let thread = Thread {
            shared: ThreadShared {
                global: Arc::clone(&shared),
                tasks: FuturesUnordered::new(),
            },
            listener,
            connections: HashMap::new(),
            task_waker: futures::task::waker(Arc::new(PollWaker {
                waker,
                failed: AtomicBool::new(false),
            })),
            next_token: FIRST_CONNECTION_TOKEN,
            poll,
            events: Events::with_capacity(EVENTS_CAPACITY),
        };

        // Never joined: the thread says it is over through the state of the server, once
        // everything of it is dropped, panicking included, see [`Done`].
        let done = Done(Arc::clone(&shared));
        thread::Builder::new()
            .name("corba-server".to_string())
            .spawn(move || {
                let _done = done;
                thread.main();
            })
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to start the polling thread of a CORBA server")
                .with_source(Box::new(err)))?;

        Ok(Self { shared })

    }

    /// The address this server is listening on, whose port is the one the system chose for a
    /// server that asked for port zero.
    pub fn local_addr(&self) -> SocketAddr {
        self.shared.local_addr
    }

    /// The addresses a client is told to reach this server at, in the order it should try
    /// them, the first being the one most likely to be reachable from elsewhere. Never
    /// empty, see [`local_addrs`].
    #[inline]
    pub fn addrs(&self) -> &[IpAddr] {
        &self.shared.addrs
    }

    /// The first of those addresses as a string, which is the one to tell a caller about
    /// when it is told one address and not a whole reference, see [`Self::build_ior`].
    #[inline]
    pub fn host(&self) -> Arc<str> {
        Arc::clone(&self.shared.host)
    }

    /// Serve an object, which is reachable from the moment this comes back and until
    /// [`Self::deregister`] takes it out, under the key it names itself with, see
    /// [`ServerObject::id`].
    ///
    /// An object registered under the key another one is served at replaces it, which is
    /// what a device restarted in place comes to.
    ///
    /// What comes back is the stringified reference a client reaches the object at, see
    /// [`Self::build_ior`].
    pub fn register(&self, object: Arc<O>) -> Result<String> {

        let ior = self.build_ior(&object)?;
        let key = Box::<[u8]>::from(object.id());

        trace!(?key, "Serving an object");

        if self.shared.objects.lock().unwrap().insert(key, object).is_some() {
            debug!("Replaced the object served under the same key");
        }

        Ok(ior)

    }

    /// Stop serving the object of the given key, giving back whether one of that key was
    /// being served at all.
    ///
    /// The object is not given back, and is not necessarily dropped here either: a request
    /// of it that is already being answered holds it until it has answered. What this does
    /// is keep any further request from ever reaching it.
    pub fn deregister(&self, id: &[u8]) -> bool {
        self.shared.objects.lock().unwrap().remove(id).is_some()
    }

    /// Whether an object of the given key is being served, see [`Self::register`].
    pub fn is_registered(&self, id: &[u8]) -> bool {
        self.shared.objects.lock().unwrap().contains_key(id)
    }

    /// The object served under the given key, if any, see [`Self::register`].
    pub fn get(&self, id: &[u8]) -> Option<Arc<O>> {
        self.shared.objects.lock().unwrap().get(id).cloned()
    }

    /// The stringified IOR of an object of this server, which is what a client is given to
    /// connect back here with, see [`Self::register`].
    ///
    /// The reference announces the largest GIOP version this server speaks, see [`VERSION`],
    /// which is well past the first one whose profile has room for components. Both of the
    /// things a client is told beyond the address are components:
    ///
    /// - the codesets this server is able to encode what it sends in, which is what a
    ///   client picks one from and then announces in the service context of its first
    ///   request. Without this a client has nothing to pick from and the narrow strings
    ///   stay in the one the specification mandates;
    /// - the other addresses this machine is reachable at, which a client falls back to
    ///   when the one of the profile itself does not answer, see [`local_addrs`].
    ///
    /// So a single profile, and not one per address the way a reference of IIOP 1.0 has to
    /// carry them for want of anywhere else to put them.
    fn build_ior(&self, object: &O) -> Result<String> {

        let port = self.shared.local_addr.port();

        // Everything this server is able to encode the strings it sends in, its own being
        // the one it would rather use, which is what a client picks from.
        fn conversions<C: AnyCodeset>(all: &[C]) -> Cow<'static, [u32]> {
            Cow::Owned(all.iter()
                .copied()
                .filter(|codeset| *codeset != C::NATIVE)
                .map(AnyCodeset::id)
                .collect())
        }

        let mut components = vec![
            ComponentKind::Codesets(CodesetsComponent {
                native_codeset: Codeset::NATIVE.id(),
                conversion_codesets: conversions(Codeset::ALL),
                native_wide_codeset: WideCodeset::NATIVE.id(),
                conversion_wide_codesets: conversions(WideCodeset::ALL),
            }).build()?,
        ];

        // The first address is the one of the profile itself, so the alternates start at
        // the second one, and a server reachable at a single address has none at all.
        for addr in self.shared.addrs.iter().skip(1) {
            components.push(ComponentKind::AltIiopAddr(AltIiopAddr {
                host: Cow::Owned(addr.to_string()),
                port,
            }).build()?);
        }

        let profile = ProfileKind::Iiop(IiopProfile {
            version: VERSION,
            host: Cow::Owned(self.shared.addrs[0].to_string()),
            port,
            object_key: Cow::Borrowed(object.id()),
            components: Cow::Owned(components),
        }).build()?;

        // The type of a reference is the most derived interface of what it points at, which
        // is the first of the ones the object declares, and `CORBA::Object` for one that
        // declares none, that being the only interface it is then known to have.
        let type_id = object.repo_ids().first().copied()
            .unwrap_or(OBJECT_REPO_ID);

        // A stringified IOR is an encapsulation of its own, so its octets start with the
        // byte order flag and the CDR alignment restarts from them.
        let mut writer = CdrWriter::new();
        writer.write_endian()?;
        writer.write(&Ior {
            type_id: Cow::Borrowed(type_id),
            profiles: Cow::Owned(vec![profile]),
        })?;

        Ok(format!("IOR:{:X}", writer.into_buf()))

    }

    /// Stop serving: the polling thread is told so and woken, so that it finds out now
    /// rather than on the next client that turns up. No client is accepted and no request
    /// taken from then on, but the requests being answered still are, as if every object
    /// was deregistered at once, see [`Self::deregister`], after which the objects are let
    /// go of and the thread is over, see [`Self::wait`].
    ///
    /// Nothing is waited for here, so this is called from anywhere, a request of an object
    /// of this very server included, and nothing of it can be undone: a server that was
    /// stopped is one nothing reaches again, whatever is registered with it afterward.
    pub fn stop(&self) {

        self.shared.state.fetch_max(STOP, Ordering::AcqRel);

        if let Err(err) = self.shared.waker.wake() {
            warn!(%err, "Failed to wake the polling thread of the CORBA server up to stop it, ignoring");
        }

    }

    /// Resolve once this server was stopped, see [`Self::stop`], and its polling thread is
    /// through, after which nothing of any object is held anymore: neither by a request
    /// that was being answered, nor by the objects registered, which the thread lets go of
    /// before it ends. A thread that panicked is through as well.
    ///
    /// This must not be awaited from a request of an object of this server, which would
    /// wait for itself.
    pub fn wait(&self) -> impl Future<Output = ()> + Send + '_ {
        poll_fn(|cx| self.shared.poll_done(cx))
    }

}

impl<O: ServerObject> Drop for Server<O> {

    /// Letting go of a server drops everything of it, with nothing sent and nothing waited
    /// for: the polling thread is woken to find the server dropped, see [`DROP`], and
    /// drops the requests being answered without answering them, and the connections.
    ///
    /// The objects are let go of here and not left to the thread, which does so as well on
    /// its way out, see [`Thread::main`]: an object that is not answering anything is then
    /// gone by the time this returns, rather than at some point after.
    fn drop(&mut self) {
        self.shared.state.fetch_max(DROP, Ordering::AcqRel);
        self.shared.objects.lock().unwrap().clear();
        if let Err(err) = self.shared.waker.wake() {
            warn!(%err, "Failed to wake the polling thread of the CORBA server up to drop it, ignoring");
        }
    }

}

/// What the polling loop and the callers registering objects both reach.
#[derive(Debug)]
struct Shared<O: ServerObject> {
    /// The objects being served, under the key a request addresses them by.
    objects: Mutex<HashMap<Box<[u8]>, Arc<O>>>,
    /// The address the listener is bound to, resolved once because a server bound to port
    /// zero only learns its own port from the system.
    local_addr: SocketAddr,
    /// The addresses a client is told to reach this server at, in the order it should try
    /// them, which is what every reference this server hands out carries, see
    /// [`local_addrs`]. Never empty.
    addrs: Box<[IpAddr]>,
    /// The first of those as a string, which is what a caller is told when it is told one
    /// address and not a whole reference, see [`Server::host`].
    host: Arc<str>,
    /// Where the server is in its life, [`RUN`], [`STOP`], [`DROP`] then [`DONE`], only ever
    /// going forward: what the polling thread does is read off this, see [`Thread::main`].
    state: AtomicU8,
    /// The tasks waiting for the server to be [`DONE`], see [`Server::wait`].
    waits: Mutex<Vec<task::Waker>>,
    /// How a thread that is not the polling one brings it back out of its poll, which is
    /// what says there is something for it to look at, see [`PollWaker`].
    waker: Arc<Waker>,
}

impl<O: ServerObject> Shared<O> {

    /// The object a request is addressed to, [`None`] for one this server does not serve,
    /// which a request addressing it any other way than by its key is as well.
    ///
    /// It is held for the length of the call and no longer: an object that stops being
    /// served while it is answering still answers that one request, see
    /// [`Server::deregister`].
    fn get_object(&self, target: &Target<'_>) -> Option<Arc<O>> {
        let key = target.key()?;
        self.objects.lock().unwrap().get(key).map(Arc::clone)
    }

    fn has_object(&self, target: &Target<'_>) -> bool {
        let Some(key) = target.key() else { return false };
        self.objects.lock().unwrap().contains_key(key)
    }

    #[inline]
    fn state(&self) -> u8 {
        self.state.load(Ordering::Acquire)
    }

    /// Whether the server is [`DONE`], the task being woken once it is if not, see
    /// [`Server::wait`].
    fn poll_done(&self, cx: &mut task::Context<'_>) -> task::Poll<()> {

        // Read again under the lock, which the thread being done is set under, so that it
        // is either seen here or sees this task, see [`Done`].
        let mut waits = self.waits.lock().unwrap();
        if self.state() == DONE {
            return task::Poll::Ready(());
        }

        if !waits.iter().any(|waker| waker.will_wake(cx.waker())) {
            waits.push(cx.waker().clone());
        }

        task::Poll::Pending

    }

}

/// Held by the polling thread for the whole of its life and dropped once everything else
/// of it is, panicking included: the server is then [`DONE`] and what waits for it woken.
struct Done<O: ServerObject>(Arc<Shared<O>>);

impl<O: ServerObject> Drop for Done<O> {
    fn drop(&mut self) {
        let shared = &self.0;
        shared.objects.lock().unwrap().clear();
        let waits = {
            let mut waits = shared.waits.lock().unwrap();
            shared.state.store(DONE, Ordering::Release);
            mem::take(&mut *waits)
        };
        for waker in waits {
            waker.wake();
        }
    }
}

/// The polling thread and everything it owns: it accepts the connections, receives the
/// requests, and drives the futures the objects answer them with.
///
/// Nothing here is reachable from anywhere else — what a caller reaches is [`Shared`] — so
/// none of it is locked, the thread being the only one that ever touches it.
#[derive(Debug)]
struct Thread<O: ServerObject> {
    /// The thread-specific shared data, containing the global shared and tasks.
    shared: ThreadShared<O>,
    /// The listener the clients connect to, only ever accepted from by the loop.
    listener: TcpListener,
    /// Every connection this server has, under the token of its socket.
    connections: HashMap<Token, Connection>,
    /// What every task is woken through, which brings the loop back out of its poll.
    task_waker: task::Waker,
    /// The token of the next connection accepted. Tokens are never reused: the one of a
    /// connection that is gone simply never comes up again.
    next_token: usize,
    poll: Poll,
    events: Events,
}

#[derive(Debug)]
struct ThreadShared<O: ServerObject> {
    global: Arc<Shared<O>>,
    /// The requests being answered, each of which writes its own reply once the object is
    /// through with it, so there is nothing here to look one up by.
    tasks: FuturesUnordered<RequestTask<O>>,
}

impl<O: ServerObject> Thread<O> {

    /// The whole of the polling thread: turn after turn until the server this serves is
    /// stopped, see [`STOP`], and drained then, see [`Self::drain`], or dropped, see
    /// [`DROP`], or until a turn fails, which only something that leaves the thread unable
    /// to poll at all does.
    ///
    /// Everything of the server that is left when this comes back is let go of here, the
    /// requests still being answered, the listener and the connections included, which is
    /// what closes them, and the objects right after, see [`Done`].
    fn main(mut self) {

        while self.shared.global.state() == RUN {
            if let Err(err) = self.run() {
                error!(%err, "Failed to poll the sources of the CORBA server, stopping it");
                break;
            }
        }

        if self.shared.global.state() != DROP {
            self.drain();
        }

        debug!(addr = %self.shared.global.local_addr, "Stopped the polling thread of the CORBA server");

    }

    /// One turn: wait for something to happen and deal with everything that did.
    ///
    /// Nothing bounds the wait, a server having nothing to do when nothing is happening: what
    /// brings this back is a client, or a task of an object that became ready, see
    /// [`PollWaker`].
    fn run(&mut self) -> Result<()> {

        match self.poll.poll(&mut self.events, None) {
            Ok(()) => {}
            // A signal is not the poll failing: nothing happened, which is what a turn that
            // timed out is as well, and the turn after this one waits again.
            Err(err) if err.kind() == io::ErrorKind::Interrupted => return Ok(()),
            Err(err) => return Err(Error::new(ErrorKind::Internal)
                .with_description("Failed to poll the sources of a CORBA server")
                .with_source(Box::new(err))),
        }

        self.handle_events();

        self.poll_tasks();

        Ok(())

    }

    /// Finish what a server that was stopped had started: no client is accepted anymore and
    /// no request taken, but the ones being answered are, and their replies sent, the
    /// request that stopped the server above all, which is how a client asking a server to
    /// stop is told it did, see [`Server::stop`].
    ///
    /// This lasts until no request is being answered and every reply went out, or the
    /// connection it was for is gone, however long that takes, unless the server is
    /// dropped in the meantime, see [`DROP`].
    fn drain(&mut self) {

        if let Err(err) = self.poll.registry().deregister(&mut self.listener) {
            debug!(%err, "Failed to take the listener of the stopped CORBA server out of the poll");
        }

        loop {

            // What woke the server to stop is a turn of its own, so the tasks are driven
            // first: the request that stopped it may well be through already.
            self.poll_tasks();

            if self.shared.tasks.is_empty() && self.connections.values().all(Connection::is_flushed) {
                return;
            }

            if self.shared.global.state() == DROP {
                debug!(tasks = self.shared.tasks.len(), "Dropped the stopped CORBA server before it answered every request");
                return;
            }

            match self.poll.poll(&mut self.events, None) {
                Ok(()) => {}
                Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
                Err(err) => {
                    warn!(%err, "Failed to poll the sources of the stopped CORBA server, giving up on the requests still being answered");
                    return;
                }
            }

            // Only what goes out, and connections closing, still matter: what a client sends
            // from now on is not read, the server answering nothing new.
            for event in &self.events {

                let token = event.token();
                let Some(connection) = self.connections.get_mut(&token) else {
                    continue;
                };

                let failure = match event.is_writable() {
                    true => connection.flush().err(),
                    false => None,
                };

                if failure.is_some() || event.is_read_closed() || event.is_write_closed() {
                    self.connections.remove(&token);
                }
                
            }

        }

    }

    /// Hand the events the poll has just given out to whatever they happened on.
    fn handle_events(&mut self) {

        // The events are moved out of the thread for the time they are dispatched: handling
        // one takes the tables they would be borrowed alongside. What is left behind holds
        // nothing, so the buffer they arrived in is the one that goes back.
        let events = mem::replace(&mut self.events, Events::with_capacity(0));

        for event in &events {

            let token = event.token();

            if token == LISTENER_TOKEN {
                self.accept();
                continue;
            }

            // A task that was woken from another thread, which is the whole of what this
            // event says: the tasks are driven once the events are through.
            if token == WAKER_TOKEN {
                continue;
            }

            // Held for the turn of this event rather than borrowed out of the table, the
            // tables of the loop being what answering a request on it takes.
            let Some(connection) = self.connections.get_mut(&token) else {
                warn!(?event, "Received a poll event for no known connection, ignoring");
                continue;
            };

            // The socket takes what is waiting to go out only once it has room for it again,
            // which is what a writable event says: this is where a reply the socket could
            // not take whole finishes going out.
            let mut failure = None;

            if event.is_writable() {
                failure = connection.flush().err();
            }

            if failure.is_none() && event.is_readable() {
                failure = connection.receive(&self.shared).err();
            }

            // The peer is through sending, which the read above finds only by reading to
            // zero: an event that carries the close without carrying anything readable is
            // caught here instead. The socket stays readable from then on, so there is no
            // second event to wait for.
            if failure.is_none() && event.is_read_closed() {
                failure = Some(Error::new(ErrorKind::Corba)
                    .with_description("The client closed the connection"));
            }

            // Either side of a connection failing is the whole of it being over. Letting go
            // of it is what closes its socket, which the system then takes out of the poll on
            // its own, and a task still answering a request of it holds it until it has,
            // whatever it answers then going nowhere.
            if let Some(err) = failure {
                debug!(addr = %connection.addr, %err, "Lost the connection to the CORBA client");
                self.connections.remove(&token);
            }

        }

        self.events = events;

    }

    /// Take every client that is waiting to be accepted, which is what a readable event on
    /// the listener says there is at least one of.
    ///
    /// The listener is watched edge-triggered, so it is accepted from until it has nothing
    /// left to give: a client not taken now is not announced again.
    fn accept(&mut self) {

        loop {

            let (mut stream, addr) = match self.listener.accept() {
                Ok(accepted) => accepted,
                Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
                Err(err) if err.kind() == io::ErrorKind::WouldBlock => break,
                Err(err) => {
                    warn!(%err, "Failed to accept a CORBA client");
                    break;
                }
            };

            // A reply is a small message a client is waiting for, so it goes out as soon as
            // it is written and not once the system has enough of them to be worth sending.
            if let Err(err) = stream.set_nodelay(true) {
                debug!(%addr, %err, "Failed to disable Nagle's algorithm on the stream");
            }

            let token = Token(self.next_token);
            self.next_token += 1;

            // Nothing is kept of a socket that could not be registered, and letting go of it
            // is what closes it: a client this server would never hear from again is better
            // told right away.
            if let Err(err) = self.poll.registry().register(&mut stream, token, Interest::READABLE | Interest::WRITABLE) {
                warn!(%addr, %err, "Failed to register the socket of the CORBA client with the poll, dropping it");
                continue;
            }

            debug!(%addr, "Accepted a CORBA client");

            self.connections.insert(token, Connection {
                token,
                stream,
                addr,
                recv_state: RecvState {
                    buf: Buf::new(),
                    fragments: HashMap::new(),
                    adjacent: None,
                    fragment_buf: None,
                },
                send_state: SendState {
                    writer: CdrWriter::new().into_sender(),
                },
            });

        }

    }

    /// Drive every task that has something to do, which is where the objects answer the
    /// requests and where the replies they came to are written.
    ///
    /// A task that is waiting on something is not polled here: [`FuturesUnordered`] only goes
    /// through the ones that were woken, and the ones that were just pushed, so this costs
    /// what there is to do and not what there is in flight.
    fn poll_tasks(&mut self) {

        // The tasks and the connections are held apart: driving a task writes its answer
        // into the connection it arrived on, so both have to be borrowed at once, which is
        // what the tasks living in a field of their own is for, see [`ThreadShared`].
        let Thread { shared, connections, task_waker, .. } = self;

        let mut cx = task::Context::from_waker(task_waker);

        while let task::Poll::Ready(Some((answer, reply))) = shared.tasks.poll_next_unpin(&mut cx) {

            // A request nothing is waiting for the answer of was run all the same, which
            // is what the client asked for, and what it answered goes nowhere.
            if !answer.response_expected {
                continue;
            }

            // And the connection it arrived on may be gone by the time the object is
            // through with it, a client that hung up being what that comes to: the answer
            // then goes nowhere either, which is exactly where it would have gone had the
            // connection lasted a moment longer.
            let Some(connection) = connections.get_mut(&answer.token) else {
                trace!(request_id = answer.request_id, "Dropping the answer to a request whose connection is gone");
                continue;
            };

            // Either side of a connection failing is the whole of it being over, the same
            // as it is for an event, see [`Thread::handle_events`].
            if let Err(err) = connection.send_answer(&answer, reply) {
                debug!(addr = %connection.addr, %err, "Lost the connection to the CORBA client");
                connections.remove(&answer.token);
            }

        }

    }

}

/// One connection of a client, shared between the polling loop, which receives the requests
/// on it, and the tasks answering them, which write their replies into it.
#[derive(Debug)]
struct Connection {
    /// The token for the poll registration.
    token: Token,
    /// The socket itself, which never blocks. Only the loop reads it, and what is written to
    /// it is written under [`Self::send_state`].
    stream: TcpStream,
    /// Where the client is, which is only ever traced.
    addr: SocketAddr,
    /// What has arrived and is not a whole message yet.
    ///
    /// Only ever locked by the thread running the loop, so it is never contended: a lock
    /// rather than a field of [`Thread`] itself because answering a request takes the tables
    /// of the loop, which this could not be borrowed alongside. This is where the server
    /// differs from the client, whose receiving state is owned by its thread.
    recv_state: RecvState,
    /// What has been written on this connection and has not gone out yet.
    send_state: SendState,
}

impl Connection {

    /// Read everything the socket of a connection has to give, and take every whole message
    /// that makes out of it.
    ///
    /// The socket is watched edge-triggered, so it is read until it has nothing left to
    /// give: what is not taken now is not announced again.
    fn receive<O: ServerObject>(&mut self, shared: &ThreadShared<O>) -> Result<()> {

        let mut stream = &self.stream;
        let (len, eof) = stream.read_nb_to_cdr(&mut self.recv_state.buf)
            .map_err(|err| Error::new(ErrorKind::Corba)
                .with_description("The connection to the client failed while it was being read")
                .with_source(Box::new(err)))?;

        trace!(addr = %self.addr, len, eof, "Read from the CORBA client");

        // Every whole message that made out of what arrived, and not only the first. A
        // client that closed right behind its last request is one whose request came in
        // the very read that found the end, so what arrived is taken before the connection
        // is given up on below.
        let failure = loop {
            match self.dispatch(shared) {
                Ok(true) => {}
                Ok(false) => break None,
                Err(err) => break Some(err),
            }
        };

        // Everything the messages above were answered with goes out here, once, and not
        // once per message: a client that sends several requests without waiting for any of
        // them is answered in as few writes as the socket will take them in. It goes out
        // even when one of those messages ended the connection, the ones in front of it
        // having been answered before it ever arrived.
        let flushed = self.flush();

        // What the messages came to is told before what writing them back came to, it
        // being the one of the two that says why this connection is over.
        if let Some(err) = failure {
            return Err(err);
        }

        flushed?;

        if eof {
            return Err(Error::new(ErrorKind::Corba)
                .with_description("The client closed the connection"));
        }

        Ok(())

    }

    /// Take the message at the front of what has arrived on a connection and give it to the
    /// object it is addressed to, leaving what is behind it for the next call.
    ///
    /// Tells whether a message was taken at all, [`false`] for octets that do not hold a
    /// whole one yet, which is what says there is no point in calling again before more of
    /// them have arrived.
    ///
    /// Failing is the connection being over, and nothing is sent to say so: the octets made
    /// no sense, so where the message they were part of stops is not known and neither is
    /// where the next one starts. A message that was framed and only then refused is not a
    /// failure at all — it is answered here and the connection carries on, see
    /// [`Self::refuse_message`] and [`Self::send_error`].
    fn dispatch<O: ServerObject>(&mut self, shared: &ThreadShared<O>) -> Result<bool> {

        // Where the reader stops is how much of the buffer this message is: whatever is
        // behind it is the front of the ones that follow.
        let recv = &mut self.recv_state;

        let buffer = recv.fragment_buf.as_ref()
            .unwrap_or(&recv.buf);

        let mut reader = CdrReader::new(buffer)
            .with_codesets(self.send_state.writer.codesets());

        let message = reader.read::<Message>()
            .map_err(|err| Error::new(ErrorKind::InvalidData)
                .with_description("Failed to read a GIOP message sent by the client")
                .with_parent(err))?;

        let body_at = reader.position();
        let message_len = body_at + message.body_len();

        let (
            response_expected, 
            object, 
            request,
        ) = match message {

            // The length is the one thing the client says that has to be believed before
            // anything of the message can be checked, so it is bounded here rather than
            // trusted.
            Message::Incomplete(total_len) => {

                if total_len > MAX_MESSAGE_LEN {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description(format!("The client announced a message of {total_len} octets, more than the {MAX_MESSAGE_LEN} allowed")));
                }

                // The header said how long the whole of it is, so the room for all of it
                // is made here, once, rather than grown a read at a time. It stays made
                // until the message has arrived, an allocation only ever growing, so the
                // read that follows is offered the whole of what is still to come and
                // takes it in one, see [`ReadExt::read_nb_to_cdr`](super::io::ReadExt::read_nb_to_cdr).
                let buf = recv.buf.make_mut();
                buf.try_reserve(total_len.saturating_sub(buf.len()))?;

                return Ok(false);

            }

            // A message in a version this side does not speak, which was measured and
            // taken off the stream and never read into, see [`Message::InvalidVersion`].
            Message::InvalidVersion(major, minor) => {
                debug!(addr = %self.addr, major, minor, "Refusing a GIOP message in a version this server does not speak");
                return self.refuse_message(message_len);
            }

            // And one of a kind this side has no notion of in the version it came in,
            // which is as far as that one was read either, see [`Message::InvalidKind`].
            Message::InvalidKind(version, kind) => {
                debug!(addr = %self.addr, %version, kind, "Refusing a GIOP message of a kind this server has no notion of");
                return self.refuse_message(message_len);
            }

            // A request that is whole, which is what most of them are.
            Message::Request(request) if !request.fragment_follows() => {

                // A whole message in the middle of a fragmented one is only a client that
                // is not making sense before 1.2, which is when the fragments that follow
                // are left attached to nothing, see [`RecvState::adjacent`].
                if recv.adjacent.is_some() {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description("The client sent a whole GIOP message in the middle of one it was sending in fragments"));
                }

                // Read and not acted on yet: a service context that cannot be read is the
                // client's doing and loses nothing but this one request, the message
                // having been framed whole, so the message is taken off the stream below
                // before anything is said about it.
                let codesets = request.find_codesets();

                let addressed = shared.global.get_object(request.target());
                let response_expected = request.response_expected();

                let server_request = ServerRequest {
                    codesets: self.send_state.writer.codesets(),
                    id: request.request_id(),
                    version: request.version(),
                    // A request that was read always has one, this being the one place a
                    // request is ever made out of one, see [`giop::Request::operation`].
                    operation: request.operation_range()
                        .expect("a GIOP request that was read names the operation it is for"),
                    body_at,
                    endian: request.endian(),
                    buf: recv.fragment_buf.take()
                        .unwrap_or_else(|| recv.buf.split_to(message_len)),
                };

                match codesets {
                    Ok(None) => {}
                    // They travel in a service context of whichever request carries them
                    // and hold for everything behind it, so they are settled before the
                    // reply to that very request is written.
                    Ok(Some(codesets)) => {
                        trace!(addr = %self.addr, ?codesets, "The CORBA client announced its codesets");
                        self.send_state.writer.set_codesets(Some(codesets));
                    }
                    // Which is a marshalling failure of the request and not of the stream:
                    // the client is told so and the connection carries on, see
                    // [`system_exception`].
                    Err(err) => {

                        let err = Error::new(ErrorKind::InvalidData)
                            .with_description("Failed to read the codesets of a GIOP request sent by the client")
                            .with_parent(err);

                        if response_expected {
                            self.send_error(server_request.version, server_request.id, &err)?;
                        }

                        return Ok(true);

                    }
                }

                (response_expected, addressed, server_request)

            }

            // The first part of a request that is being sent in fragments, which is kept
            // until the ones carrying the rest of it have arrived.
            Message::Request(request) => {

                if recv.adjacent.is_some() {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description("The client sent a whole GIOP message in the middle of one it was sending in fragments"));
                }

                let request_id = request.request_id();
                let version = request.version();

                // Only a fragmentation that has to be followed by adjacency is remembered,
                // every fragment of 1.2 and later naming the request it continues itself.
                if version < Version::V12 {
                    recv.adjacent = Some(request_id);
                }

                let part = recv.buf.split_to(message_len).into_mut();
                let _ = recv.fragments.insert(request_id, part);

                return Ok(true);

            }

            Message::Fragment(frag) => {
                
                // 1.2 has a fragment name the request it continues, and the versions before
                // it do not: the message that announced them is then the only thing that can
                // say, see [`RecvState::adjacent`].
                let Some(request_id) = frag.request_id().or(recv.adjacent) else {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description("The client sent a GIOP fragment of no message at all"));
                };

                let fragment_follows = frag.fragment_follows();
                // The octets it carries lie right where reading it left the reader, and
                // there are exactly as many of them as the fragment says, its length
                // having been checked against what arrived before it was ever read.
                let data = reader.read_octet_slice(frag.body_len())
                    .expect("a fragment carries the octets it says it carries");

                let Some(part) = recv.fragments.get_mut(&request_id) else {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description(format!("The client sent a GIOP fragment of the request {request_id}, which no message was announced for")));
                };

                // The octets of a fragment carry on the CDR stream of the message it
                // continues, so they are appended to it as they are and nothing but the
                // header of the fragment itself is left behind.
                if part.len() + data.len() > MAX_MESSAGE_LEN {
                    return Err(Error::new(ErrorKind::InvalidData)
                        .with_description(format!("The client sent a GIOP message of more than {MAX_MESSAGE_LEN} octets in fragments")));
                }

                part.extend_from_slice(&data);

                let _ = recv.buf.split_to(message_len);

                if fragment_follows {
                    return Ok(true);
                }

                // The message is whole, so it leaves the ones being put back together, the
                // others of 1.2 that are still going on staying where they are.
                let part = recv.fragments.remove(&request_id)
                    .expect("the message being put back together was just written into");

                recv.adjacent = None;

                // Its header is patched to say what it now is rather than what its first
                // part was, see [`MessageWhole`]. Nothing but this side could fail there,
                // the length being capped above and the room for it being in the header the
                // first part brought.
                let mut writer = CdrWriter::from_buf(part);
                let res = writer.write(&MessageWhole);

                // We set the fragment buffer and return Ok(true) to handle it next. It is
                // set even when the patching failed, so that the octets it holds go with
                // the connection rather than staying in front of everything behind them.
                recv.fragment_buf = Some(writer.into_buf().freeze());

                res.map_err(|err| Error::new(ErrorKind::InvalidData)
                    .with_description("Failed to put a GIOP message back together")
                    .with_parent(err))?;

                return Ok(true);

            }

            // A client asking whether an object is here before it sends it anything, which
            // this side answers by itself: no object is reached for it.
            Message::LocateRequest(locate) => {

                let request_id = locate.request_id();
                let version = locate.version();

                let status = if shared.global.has_object(locate.target()) {
                    LocateStatus::ObjectHere
                } else {
                    LocateStatus::UnknownObject
                };

                let _ = recv.buf.split_to(message_len);

                trace!(addr = %self.addr, request_id, ?status, "Answering a locate request");
                self.send_message(&Message::LocateReply(LocateReply::new(version, request_id, status)), None)?;

                return Ok(true);

            }

            // A client that has given up on the reply of a request. It is a courtesy and not
            // an order, see the specification: a server is free to answer a request it was
            // asked to drop, and this one does, the client dropping the reply when it turns
            // up.
            //
            // TODO: Drop the task of that request, which asks for the tasks to be looked up
            // by the request they answer, which they are not.
            Message::CancelRequest(cancel) => {
                trace!(addr = %self.addr, request_id = cancel.request_id(), "Ignoring the cancellation of a request");
                let _ = recv.buf.split_to(message_len);
                return Ok(true);
            }

            // Every kind that is left is one this side reads and a client never sends: a
            // reply, a locate reply, a close, or a message error. They are not octets this
            // side made nothing of, they are a peer that is not a client, and there is
            // nothing to carry on with.
            _ => {
                return Err(Error::new(ErrorKind::InvalidData)
                    .with_description("The client sent a GIOP message that is not a request"));
            }

        };

        let Some(object) = object else {
            
            debug!(addr = %self.addr, request_id = request.id, "Refusing a request to an object that is not served");
            if response_expected {
                self.send_exception(request.version, request.id,
                    &SystemException::new(exception::OBJECT_NOT_EXIST))?;
            }
            
            return Ok(true);

        };

        // The operations every CORBA object has, which this side answers out of what the
        // object says about itself and without ever reaching it.
        if self.builtin_operation(&*object, response_expected, &request)? {
            return Ok(true);
        }

        // And everything else is the object's own, which answers it on its own time: what
        // it gives back is driven by the loop, see [`Thread::poll_tasks`].
        //
        // Where the answer goes is read out of the request before it is given away, the
        // object owning it from here on and the answer having to find its way back
        // without it.
        let answer = Answer {
            token: self.token,
            version: request.version,
            request_id: request.id,
            response_expected,
        };

        shared.tasks.push(RequestTask {
            fut: object.request(request),
            answer,
        });

        Ok(true)

    }

    /// Answer one of the operations every CORBA object has, whatever its IDL declares, if
    /// the request names one: they are answered out of what the object says about itself
    /// and none of them reaches the object at all.
    ///
    /// Tells whether the request was answered here, [`false`] for an operation of the
    /// object itself, which is then the only thing left to give it to.
    fn builtin_operation<O: ServerObject>(&mut self, object: &O, response_expected: bool, request: &ServerRequest) -> Result<bool> {

        // Two spellings of `_non_existent` are out there and both are sent.
        if !matches!(request.operation(), b"_is_a" | b"_non_existent" | b"_not_existent" | b"_interface") {
            return Ok(false);
        }

        // One nothing is waiting for the answer of is one there is nothing at all to do
        // about: it was taken off the stream like any other, and that is the whole of it.
        if !response_expected {
            return Ok(true);
        }

        match request.operation() {
            // Whether the object is of the repository ID the client names, which it is for
            // its own and for every one it inherits, see [`ServerObject::repo_ids`].
            b"_is_a" => {

                let mut reader = request.read();

                // The message was framed whole, so the stream still stands where it should
                // and only this request is lost: arguments that make no sense for the
                // operation they were sent to are answered and not given up on, unlike
                // octets that cannot be framed at all.
                let repo_id = match reader.read_string() {
                    Ok(repo_id) => repo_id,
                    Err(err) => {
                        self.send_error(request.version, request.id, &err)?;
                        return Ok(true);
                    }
                };

                let is_a = repo_id == OBJECT_REPO_ID || object.repo_ids().iter().any(|declared| *declared == repo_id);

                let reply = request.make_reply(ReplyStatus::NoException);
                self.send_message(&Message::Reply(reply), Some(&is_a))?;

            }
            // Whether the object is gone, which it never is here: a request only ever
            // reaches one that is being served, see [`Shared::get_object`].
            b"_non_existent" | b"_not_existent" => {
                let reply = request.make_reply(ReplyStatus::NoException);
                self.send_message(&Message::Reply(reply), Some(&false))?;
            }
            // The interface repository entry of the object, which this side keeps none of,
            // and the only operation left of the ones named above.
            _ => {
                self.send_exception(request.version, request.id,
                    &SystemException::new(exception::NO_IMPLEMENT))?;
            }
        }

        Ok(true)

    } 

    /// Whether everything written on this connection went out.
    fn is_flushed(&self) -> bool {
        self.send_state.writer.buffer().is_empty()
    }

    /// Give the socket as much of what is waiting to go out as it will take, leaving the rest
    /// for the writable event that says it will take more.
    fn flush(&mut self) -> Result<()> {

        let mut stream = &self.stream;
        stream.write_nb_from_cdr(&mut self.send_state.writer)
            .map_err(|err| Error::new(ErrorKind::Corba)
                .with_description("The connection to the client failed while it was being written to")
                .with_source(Box::new(err)))

    }

    /// Lay down a message this server answers by itself, and give the socket as much of
    /// what is then waiting to go out as it will take.
    ///
    /// A message that could not be laid down whole leaves nothing of itself behind: the
    /// octets it did write are dropped and the stream goes back to where it stood, so what
    /// waits here is only ever whole messages. The origin goes back either way, it being
    /// where the octets that have yet to go out start, and a message failing halfway must
    /// not take the ones in front of it along.
    fn send_message(&mut self, message: &Message<'_>, body: Option<&dyn CdrWrite>) -> Result<()> {

        let writer = &mut *self.send_state.writer;

        let origin = writer.origin();
        let at = writer.position();
        writer.set_origin(at);

        let res = writer.write(message)
            .and_then(|()| body.map_or(Ok(()), |body| writer.write(body)))
            .and_then(|()| writer.write(&MessageLen));

        writer.set_origin(origin);
        if res.is_err() {
            writer.set_position(at);
        }

        // A message this side could not write is this side's own doing, whatever the writer
        // failed with, and is said to be: a reply that could not be encoded reads nothing
        // like a client whose octets could not be decoded, and the two are not to be
        // confused in a log.
        res.map_err(|err| Error::new(ErrorKind::Corba)
            .with_description("Failed to write a GIOP message this server answers with")
            .with_parent(err))

    }

    /// Take the message at the front off the stream without reading it, and tell the
    /// client that this side made nothing of it.
    ///
    /// This is for a message that was framed and not understood: its length was read out
    /// of a header that says the same thing in every version and for every kind, so what
    /// follows it is still a message and the connection carries on with that one. What the
    /// client is told names the latest version this side speaks, so that one which spoke a
    /// later one comes back in it, see section 9.4.8 of the specification.
    ///
    /// Gives back [`true`], which is what says a message was taken, so that a caller can
    /// answer with this and nothing more.
    fn refuse_message(&mut self, message_len: usize) -> Result<bool> {

        // Before 1.2 a fragment names nothing and continues the message right in front of
        // it, so one skipped here would leave the fragments behind it carrying on a message
        // that never was: that one ends the connection, see [`RecvState::adjacent`].
        if self.recv_state.adjacent.is_some() {
            return Err(Error::new(ErrorKind::InvalidData)
                .with_description("The client sent a GIOP message this server made nothing of in the middle of one it was sending in fragments"));
        }

        let _ = self.recv_state.buf.split_to(message_len);
        self.send_message(&Message::MessageError(VERSION), None)?;

        Ok(true)

    }

    /// Send what an object answered a request with, which is either the reply it wrote or
    /// the failure it came to, see [`Thread::poll_tasks`].
    fn send_answer(&mut self, answer: &Answer, reply: Result<ServerReply>) -> Result<()> {

        match reply {
            // What the object wrote is a whole message of its own already, so it is laid
            // into the stream as it stands rather than copied in behind whatever is
            // waiting there, see [`CdrSender::interleave`].
            Ok(reply) => match reply.finalize() {
                Ok(buffer) => self.send_state.writer.interleave(buffer),
                // The operation ran and this side could not write down what it answered,
                // which is this side's own doing and not the client's. The client is told
                // so, and told that the operation did run, unlike one whose arguments
                // could not be read in the first place, see [`system_exception`].
                Err(err) => {
                    warn!(addr = %self.addr, request_id = answer.request_id, %err,
                        "Failed to write the answer of an object");
                    self.send_exception(answer.version, answer.request_id,
                        &SystemException::new(exception::MARSHAL)
                            .with_completion_status(CompletionStatus::Yes))?;
                }
            },
            // And a failure of the object itself, which goes back as whichever system
            // exception says what it was.
            Err(err) => self.send_error(answer.version, answer.request_id, &err)?,
        }

        self.flush()

    }

    /// Tell a client that its request failed for a reason of this side and not of the
    /// object, which is what a request addressed to an object that is not here comes to.
    fn send_exception(&mut self, version: Version, request_id: u32, exception: &SystemException) -> Result<()> {
        let reply = Reply::new(version, request_id, ReplyStatus::SystemException);
        self.send_message(&Message::Reply(reply), Some(exception))
    }

    /// Answer a request with the CORBA system exception that says what this side made of a
    /// failure, see [`system_exception`].
    ///
    /// This is for what the backend itself could not do, and not for what a device has to
    /// say about a request it could not run: that is a `DevFailed`, a user exception the
    /// IDL of the object declares, which the object writes into its own reply and which
    /// never reaches here, see [`ServerObject::request`].
    fn send_error(&mut self, version: Version, request_id: u32, err: &Error) -> Result<()> {

        let (exception_id, completion_status) = system_exception(err.kind());

        debug!(addr = %self.addr, request_id, exception_id, %completion_status, %err,
            "Answering a request with a system exception");

        self.send_exception(version, request_id, &SystemException::new(exception_id)
            .with_completion_status(completion_status))

    }

}

/// What a connection has received and not framed into a whole message yet.
#[derive(Debug)]
struct RecvState {
    /// The octets that have arrived and are not framed into a whole message yet.
    buf: Buf<u8, 1>,
    /// The messages being put back together out of their fragments, under the id of the
    /// request each of them carries.
    fragments: HashMap<u32, BufMut<u8, 1>>,
    /// The request the fragments that follow carry on the message of, for a version that
    /// does not have them say it themselves: before 1.2 a fragment names nothing, so the
    /// message right in front of it is the only thing that can say what it continues.
    adjacent: Option<u32>,
    /// When a fragment is done, it is set here, and the next dispatch() call will use it
    /// instead of the normal receiving buffer, temporarily.
    fragment_buf: Option<Buf<u8, 1>>,
}

/// The sending side of a connection, which is one thing at a time: a reply goes out whole,
/// and the ones behind it wait their turn here.
#[derive(Debug)]
struct SendState {
    /// What has been written and the socket has not taken yet, which goes out on the writable
    /// event that says it will take more.
    writer: CdrSender,
}

/// An object a server serves, which is what a request is addressed to.
pub trait ServerObject: Send + Sync + 'static {

    /// What answering a request of this object is waiting on.
    ///
    /// It outlives the call that made it, so it borrows nothing of the object: an
    /// implementation that needs the object itself puts an [`Arc`] of it in there, which is
    /// what the server holds it as anyway, see [`Server::register`].
    type RequestFuture: Future<Output = Result<ServerReply>> + Send + 'static;

    /// The object key this object is reachable at, which is what a client puts in every
    /// request it sends it and what the server looks it up by, see [`Server::register`].
    fn id(&self) -> &[u8];

    /// Return all the repository IDs in the inheritance chain of this object's type.
    fn repo_ids(&self) -> &[&str];

    /// Answer one request, of the operation the client named.
    ///
    /// Which operation that is comes off the request itself and is never handed over
    /// beside it: it is read back out of the very octets the request arrived as, so an
    /// object that keeps the request keeps the name with it and copies nothing, see
    /// [`ServerRequest::operation`].
    ///
    /// The future this gives back is driven by the polling loop itself, so everything it does
    /// between two awaits happens on the very thread that receives every other request. An
    /// object that only forwards calls to other objects therefore costs no thread at all
    /// while it waits, however many calls it has out at once, and one that has to block — a
    /// device reaching hardware, a reply large enough that encoding it is real work — hands
    /// that part to a thread of its own and awaits its answer.
    ///
    /// Blocking here is not only unfair to the requests behind this one: an object that
    /// blocks on another object of this very server deadlocks it, the loop that would answer
    /// it being the one that is blocked.
    fn request(self: Arc<Self>, req: ServerRequest) -> Self::RequestFuture;

}

/// Where the answer of a request goes, which a task carries because the request itself is
/// given away to the object that answers it, see [`Thread::poll_tasks`].
#[derive(Debug, Clone, Copy)]
struct Answer {
    /// The connection the request arrived on, which is the one its answer goes out on, if
    /// that connection is still there by then.
    token: Token,
    /// The version the reply goes out in, which is the one the request came in.
    version: Version,
    /// The request the reply answers, which is what the client matches it against.
    request_id: u32,
    /// Whether the client is waiting for an answer at all: one that is not is a request
    /// that is still run, and whose answer simply goes nowhere.
    response_expected: bool,
}

pin_project! {
    /// Internal [`Future`] implementor for polling a task resolution and get a reply back.
    struct RequestTask<O: ServerObject> {
        #[pin]
        fut: O::RequestFuture,
        answer: Answer,
    }
}

impl<O: ServerObject> Future for RequestTask<O> {

    type Output = (Answer, Result<ServerReply>);

    #[inline]
    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
        let this = self.project();
        this.fut.poll(cx).map(|reply| (*this.answer, reply))
    }

}

/// A request an object is to answer, as the octets it arrived as.
///
/// Its header was read on the way in, which is what routed it here, and its arguments are
/// read in place out of these very octets and never copied, see [`Self::read`].
#[derive(Debug)]
pub struct ServerRequest {
    /// The whole of the GIOP message the request arrived as.
    buf: Buf<u8, 1>,
    /// The codesets of the connection it arrived on, which the message itself says nothing
    /// about and which its strings are encoded in.
    codesets: Option<Codesets>,
    /// What the reply to this request answers, which its header carried and which is kept
    /// here so that starting the reply does not read it again.
    id: u32,
    /// The version the request came in, which is the one its reply goes out in, whatever
    /// else this side speaks.
    version: Version,
    /// Where the name of the operation lies in those octets, so that it is read back out
    /// of them rather than copied anywhere, see [`Self::operation`].
    operation: Range<usize>,
    /// Where the arguments start in those octets, which the header said as well, the
    /// alignment staying relative to the first octet of the message.
    body_at: usize,
    /// The order the octets of the client are in, which is not this side's to choose.
    endian: Endian,
}

impl ServerRequest {

    /// The name of the operation the client asked for, as the octets it is on the wire,
    /// its terminating nul stripped.
    ///
    /// They are the very octets the request arrived as and are cut out of them here, so
    /// nothing of the name was copied anywhere on the way, see
    /// [`giop::Request::operation_range`](super::giop::Request::operation_range).
    #[inline]
    pub fn operation(&self) -> &[u8] {
        &self.buf.as_bytes()[self.operation.clone()]
    }

    #[inline]
    fn make_reply(&self, status: ReplyStatus) -> Reply<'_> {
        Reply::new(self.version, self.id, status)
    }

    /// Read the arguments of the request, in place in the very octets it arrived as.
    ///
    /// It reads through the whole message and not through the arguments alone, the CDR
    /// alignment being relative to the first octet of that message: a caller that goes back
    /// over what it read moves the reader by [`CdrReader::set_position`] rather than making
    /// one of its own.
    pub fn read(&self) -> CdrReader<'_> {
        // The header was read on the way in and nothing of it is read again: where the
        // arguments start is what it left behind, see [`Thread::dispatch`].
        let mut reader = CdrReader::new(&self.buf).with_endian(self.endian).with_codesets(self.codesets);
        reader.set_position(self.body_at);
        reader
    }

    /// Start the reply to this request, the writer that comes back being where its body
    /// goes, see [`ServerReply::finalize`].
    ///
    /// The header is laid down here and not once the body is written, so that everything the
    /// object writes is aligned where it really lies in the message: CDR alignment is
    /// relative to the first octet of a message and not to the first octet of a body.
    ///
    /// The reply is a normal one unless the object raises, which it does by starting the
    /// reply with [`ReplyStatus::UserException`] and writing the exception into it: an
    /// exception the IDL of the object declares is not a failure.
    pub fn reply(&self, status: ReplyStatus) -> Result<ServerReply> {
        
        let mut writer = CdrWriter::with_capacity(REPLY_MIN_LEN).with_codesets(self.codesets);
        writer.write(&Message::Reply(self.make_reply(status)))?;

        Ok(ServerReply { 
            writer,
        })
        
    }

    /// The reply to this request carrying a user exception, which is one the IDL of the
    /// object declares and therefore an answer and not a failure.
    ///
    /// What comes back is left right behind the repository ID of the exception, which is
    /// where its own fields go, see [`Self::reply`].
    pub fn reply_user_exception(&self, repo_id: &str) -> Result<ServerReply> {
        let mut reply = self.reply(ReplyStatus::UserException)?;
        reply.write_string(repo_id)?;
        Ok(reply)
    }

    /// The reply to this request carrying a system exception, which is a failure of the
    /// protocol itself and not anything the IDL declares: an object answers one when the
    /// request makes no sense at all, an operation it does not have being the usual
    /// reason, see [`Self::reply`].
    ///
    /// What comes back is whole, the exception being the entire body of such a reply.
    pub fn reply_system_exception(&self, exception: &SystemException<'_>) -> Result<ServerReply> {
        let mut reply = self.reply(ReplyStatus::SystemException)?;
        reply.write(exception)?;
        Ok(reply)
    }

}

/// What an object answered a request with, as the octets of the whole reply message, which
/// are the very ones that go out on the wire.
#[derive(Debug)]
pub struct ServerReply {
    writer: CdrWriter,
}

impl ServerReply {

    fn finalize(mut self) -> Result<Buf<u8, 1>> {
        self.writer.write(&MessageLen)?;
        Ok(self.writer.into_buf().freeze())
    }

}

impl Deref for ServerReply {
    type Target = CdrWriter;
    #[inline]
    fn deref(&self) -> &Self::Target {
        &self.writer
    }
}

impl DerefMut for ServerReply {
    #[inline]
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.writer
    }
}

/// The CORBA system exception a failure of this backend is told to a client as, and how far
/// the operation had got when it failed, which is what says whether the client may send it
/// again, see [`CompletionStatus`].
///
/// Only what this side has no word of its own for comes through here. Everything a Tango
/// device has to say about a request it could not run is a `DevFailed`, a user exception the
/// IDL declares, which the object writes into its own reply: an exception an operation
/// declares is an answer and not a failure, and it never reaches this.
fn system_exception(kind: ErrorKind<'_>) -> (&'static str, CompletionStatus) {
    match kind {
        // Octets that make no sense for what they were sent as, which is the client's doing
        // and says nothing the operation did, it never having been reached.
        ErrorKind::InvalidData => (exception::MARSHAL, CompletionStatus::No),
        // An operation the object does not have, which is a request that should never have
        // been sent rather than one the object could have answered.
        ErrorKind::NotHandled => (exception::BAD_OPERATION, CompletionStatus::No),
        // One it does have and nothing behind it implements.
        ErrorKind::NotImplemented => (exception::NO_IMPLEMENT, CompletionStatus::No),
        // The object was there when the request was routed to it and is not any more, which
        // is a device deregistered while it was being answered, see [`Server::deregister`].
        ErrorKind::DevNotFound | ErrorKind::DevNotExported => (exception::OBJECT_NOT_EXIST, CompletionStatus::No),
        // A wait that was given up on is one the operation may well be running past, which
        // is what a client has to assume of anything it stopped waiting for.
        ErrorKind::TimedOut => (exception::TIMEOUT, CompletionStatus::Maybe),
        // This side broke inside, which says nothing about how far the operation got.
        ErrorKind::Internal => (exception::INTERNAL, CompletionStatus::Maybe),
        // And everything else is a failure CORBA has no word for, which is exactly what
        // this one is: the client is told that the request failed and nothing more.
        _ => (exception::UNKNOWN, CompletionStatus::Maybe),
    }
}

/// Waking a task is bringing the polling loop back out of its poll, which is what then
/// drives that task, see [`Thread::poll_tasks`].
///
/// The wake may come from any thread — an object answering on a thread of its own, or the
/// client connection a forwarding object is waiting on — and it may also come from the loop
/// itself, a task woken while it is being polled, which the poll then comes straight back
/// out of.
#[derive(Debug)]
struct PollWaker {
    waker: Arc<Waker>,
    /// Whether a wake was ever lost, so that a server whose waker is broken says so once and
    /// not on every task it drives.
    failed: AtomicBool,
}

impl ArcWake for PollWaker {

    fn wake_by_ref(arc_self: &Arc<Self>) {

        let Err(err) = arc_self.waker.wake() else {
            return;
        };

        if !arc_self.failed.swap(true, Ordering::Relaxed) {
            warn!(%err, "Failed to wake the polling thread of the CORBA server up for a task that is ready, ignoring");
        }

    }

}
/// The addresses a client is told to reach a server bound to the given address at, in the
/// order it should try them, which is one profile of every reference that server hands out.
///
/// They are addresses and not the name of the host, which spares every client a name
/// resolution, and spares the server from depending on the name it knows itself by being
/// the one that resolves to it from where the clients are.
///
/// A server bound to one interface is only reachable through that one, whatever else this
/// machine answers to, so that is the only address handed out for it, the loopback
/// included: a server bound to the loopback is one that means to be reachable from this
/// machine and nowhere else, and saying anything else of it would be a lie. Only a server
/// bound to every interface, which is the usual one, is looked up below.
///
/// Never empty: a machine that answers nothing below is one that is only reachable through
/// its loopback, which is what is then handed out, and which is exactly right for a client
/// running beside the server.
///
/// TODO: Every address of every interface, which is what a machine bound to all of them is
/// really reachable at, and which needs `getifaddrs` and therefore a binding to it. What is
/// found here is the address of the interface the default route goes through, which is the
/// one a client elsewhere on the network reaches, along with the ones the name of the host
/// resolves to.
fn local_addrs(bound: SocketAddr) -> Box<[IpAddr]> {

    // Bound to one interface and not to every one of them, so there is nothing to look up:
    // that interface is the whole of where this server is.
    if !bound.ip().is_unspecified() {
        return Box::from([bound.ip()]);
    }

    /// The address the system would send a datagram to the given one from, which is asked
    /// of the routing table and sends nothing at all: connecting a datagram socket only
    /// picks the interface, no packet ever leaving.
    ///
    /// The address to ask about is one of the ranges the specification reserves for the
    /// documentation, so it is never one this machine really talks to, and it is only
    /// there to stand for 'somewhere else on the network'.
    fn route_addr(to: (IpAddr, u16), from: IpAddr) -> Option<IpAddr> {
        let socket = UdpSocket::bind((from, 0)).ok()?;
        socket.connect(to).ok()?;
        Some(socket.local_addr().ok()?.ip())
    }

    let mut addrs = Vec::new();

    let mut push = |addr: IpAddr| {
        if !addr.is_unspecified() && !addr.is_loopback() && !addrs.contains(&addr) {
            addrs.push(addr);
        }
    };

    // The interface the default route goes through, which is where a client that is not on
    // this machine reaches it, and therefore the address to try first.
    if let Some(addr) = route_addr((IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), 9), IpAddr::V4(Ipv4Addr::UNSPECIFIED)) {
        push(addr);
    }
    if let Some(addr) = route_addr((IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)), 9), IpAddr::V6(Ipv6Addr::UNSPECIFIED)) {
        push(addr);
    }

    // Whatever the name of this host resolves to, which is what a machine with several
    // interfaces usually has declared, and which the route above only found one of.
    if let Ok(name) = fs::read_to_string("/etc/hostname")
        .map(|name| name.trim().to_string())
        .or_else(|_| env::var("HOSTNAME"))
    && !name.is_empty()
    && let Ok(resolved) = (&*name, 0u16).to_socket_addrs() {
        for addr in resolved {
            push(addr.ip());
        }
    }

    if addrs.is_empty() {
        addrs.push(IpAddr::V4(Ipv4Addr::LOCALHOST));
    }

    addrs.into_boxed_slice()

}