rstango 0.3.0

use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};
use std::collections::{HashMap, hash_map};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use std::sync::{Arc, Mutex, Weak};
use std::{io, mem, slice, task};
use std::net::SocketAddr;
use std::future::Future;
use std::borrow::Cow;
use std::pin::Pin;

use mio::{Events, Interest, Poll, Registry, Token, Waker};
use tracing::{debug, error, trace, warn};
use mio::net::TcpStream;

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

use super::cdr::{CdrReader, CdrWrite, CdrWriter, CdrSender};
use super::ior::{Ior, ProfileKind, ComponentKind};
use super::{MAX_MESSAGE_LEN, EVENTS_CAPACITY};
use super::io::{ReadExt as _, WriteExt as _};
use super::codeset::{self, Codesets};
use super::endian::Endian;
use super::giop::{
    CancelRequest, Message, MessageLen, ReplyStatus, Request,
    MessageWhole, ServiceContext, ServiceContextKind, SystemException, Target, Version};


/// The latest GIOP version this client speaks, and the one every request goes out in unless
/// the reference it was made out of published an earlier one.
const VERSION: Version = Version::V12;

/// How often the polling thread goes through the connections for the ones nothing is using.
const SWEEP_INTERVAL: Duration = Duration::from_secs(30);

/// The token the waker of the polling thread comes back under, which is the client having
/// been dropped and nothing else.
const WAKER_TOKEN: Token = Token(0);

/// The token of the first socket registered with the poll, the ones before it being taken.
const FIRST_CONNECTION_TOKEN: usize = 1;
/// The room a request is started with, which is enough that the usual ones are written
/// into the one allocation they were made with: a GIOP header, an object key and an
/// operation name is what most of the requests of this protocol are. It is the same
/// length, for the same reason, as what a read is offered room for, see
/// [`ReadExt::read_nb_to_cdr`].
const REQUEST_MIN_LEN: usize = 512;


/// This represent a single corba client.
#[derive(Debug)]
pub struct Client {
    shared: Arc<Shared>,
    /// Takes the polling thread out of its poll when this client is dropped, see
    /// [`Self::drop`].
    waker: Waker,
    /// The polling thread, which is only ever taken out of here to be joined when this
    /// client is dropped, and so is only ever `None` while that is happening.
    thread: Option<JoinHandle<()>>,
}

impl Client {

    /// Start a client, whose polling thread is running by the time this comes back and runs
    /// until this client is dropped.
    ///
    /// A client is what connections are shared through: every object reached on one server,
    /// with the same codesets and version, goes through one socket, see [`Shared::connect`].
    /// So a program wants one of these and not one per object.
    pub fn new() -> Result<Self> {

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

        // The poll itself goes to the thread, which is the only one ever to block in it, and
        // what is left here is what a socket is registered with, by whichever thread opens
        // the connection, see [`Shared::connect`].
        let poll_registry = poll.registry().try_clone()
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to clone the poll registry of a CORBA client")
                .with_source(Box::new(err)))?;

        // Made here and not when it is first needed: a waker is tied to the poll it wakes,
        // and by the time this client is dropped the thread is blocked in that very poll.
        let waker = Waker::new(poll.registry(), WAKER_TOKEN)
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to create the waker of a CORBA client")
                .with_source(Box::new(err)))?;

        let shared = Arc::new(Shared {
            connections: Mutex::new(Connections::default()),
            poll_registry,
            next_token: AtomicUsize::new(FIRST_CONNECTION_TOKEN),
            running: AtomicBool::new(true),
        });

        let thread = Thread {
            shared: Arc::clone(&shared),
            connections: HashMap::new(),
            next_sweep: Instant::now() + SWEEP_INTERVAL,
            poll,
            events: Events::with_capacity(EVENTS_CAPACITY),
        };

        let handle = thread::Builder::new()
            .name("corba-client".to_string())
            .spawn(move || thread.main())
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to spawn the polling thread of a CORBA client")
                .with_source(Box::new(err)))?;

        Ok(Self {
            shared,
            waker,
            thread: Some(handle),
        })

    }

    /// Connect to a specific CORBA object using an IOR, the underlying implementation 
    /// currently only supports IIOP and so TPC connection.
    pub async fn connect_object_ior(&self, ior: &str, type_id_filter: &[&str]) -> Result<ClientObject> {

        let profiles = parse_ior_profiles(ior, type_id_filter)
            .map_err(|err| Error::new(ErrorKind::Corba)
                .with_append_description("Cannot parse IOR profiles to connect to CORBA object")
                .with_parent(err))?;

        let mut last_err = None::<Error>;
        
        for profile in profiles {
            for (host, port) in profile.addrs {
                
                let err = match self.connect_object_iiop(&host, port, profile.codesets, Some(profile.version), Arc::clone(&profile.object)).await {
                    Ok(object) => return Ok(object),
                    Err(err) => err,
                };

                debug!(host, port, %err, "Failed to connect to the CORBA object, trying the next profile of the IOR");
                last_err = Some(err);

            }
        }

        match last_err {
            Some(err) => Err(Error::new(ErrorKind::Corba)
                .with_append_description(format_args!("No profile of IOR accepted a connection"))
                .with_parent(err)),
            None => Err(Error::new(ErrorKind::Corba)
                .with_append_description(format_args!("The IOR resolves to no profile at all"))),
        }
        
    }

    /// Connect to a specific CORBA object using TCP host/port and object id.
    pub async fn connect_object_iiop(&self, host: &str, port: u16, codesets: Option<Codesets>, version: Option<Version>, object: Arc<[u8]>) -> Result<ClientObject> {
        let connection = self.connect_iiop(host, port, codesets, version).await?;
        Ok(ClientObject {
            connection,
            object,
        })
    }

    /// Get the connection to a server, which is the one this client is already on, or one
    /// opened here and waited for.
    ///
    /// The candidates are tried in the order the resolver gave them in, and what the last
    /// one failed with is what the whole of this failed with, the earlier ones being traced
    /// and nothing more: a host resolving to several addresses is one that answers on one
    /// of them.
    /// 
    /// The GIOP version might be specified and will be used when initiating the 
    /// connection, and if the server is not compatible, it might degrade after.
    /// When not
    async fn connect_iiop(&self, host: &str, port: u16, codesets: Option<Codesets>, version: Option<Version>) -> Result<Arc<Connection>> {

        let candidates = resolve(host, port).await?;
        let mut last_err = None::<(SocketAddr, Error)>;

        for addr in candidates {

            let err = match self.shared.connect(addr, codesets, version) {
                Ok(connection) => {
                    let init = connection.wait_init().await;
                    match init {
                        Ok(()) => return Ok(connection),
                        Err(err) => err,
                    }
                }
                Err(err) => err,
            };

            debug!(host, port, %addr, %err, "Failed to connect to the CORBA server, trying the next address");
            last_err = Some((addr, err));

        }

        match last_err {
            Some((addr, err)) => Err(Error::new(ErrorKind::Corba)
                .with_append_description(format_args!("No address of {host}:{port} accepted a connection, last tried {addr}"))
                .with_parent(err)),
            None => Err(Error::new(ErrorKind::Corba)
                .with_append_description(format_args!("The host {host}:{port} resolves to no address at all"))),
        }

    }

}

impl Drop for Client {

    /// Stop the polling thread and wait for it, which closes every connection this client
    /// was left holding: an object made out of this client and outliving it is one whose
    /// requests never go anywhere.
    fn drop(&mut self) {

        self.shared.running.store(false, Ordering::Relaxed);

        // The thread spends nearly all of its time blocked in the poll, which nothing but
        // the sweep would take it out of on its own, half a minute away: waking it is what
        // makes the drop of a client cost nothing. A wake that fails is only that wait being
        // that long again, the thread still coming out of the poll for its sweep.
        if let Err(err) = self.waker.wake() {
            warn!(%err, "Failed to wake the polling thread of the CORBA client being dropped");
        }

        // Nothing here ever runs on the polling thread, which holds the shared state and
        // nothing of the client, so this is never a thread joining itself.
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }

    }

}

/// This represent an client object connection.
#[derive(Debug)]
pub struct ClientObject {
    connection: Arc<Connection>,
    object: Arc<[u8]>,
}

impl ClientObject {

    #[inline]
    pub fn object(&self) -> &[u8] {
        &self.object
    }

    /// Send a request without blocking, and get back a [`ClientRequest`] which can be
    /// awaited for its reply.
    ///
    /// This never fails on its own: a request that could not be laid down at all still
    /// gives back a [`ClientRequest`], whose failure only turns up once it is awaited, see
    /// [`Connection::request`].
    #[inline]
    pub fn request(&self, operation: &[u8], data: &dyn CdrWrite) -> ClientRequest {
        self.connection.request(self.object(), operation, data)
    }

}

/// The state of the polling thread, which reads the sockets and finishes the connects, and
/// owns nothing: a connection is opened by whoever wants it, see [`Shared::connect`].
struct Thread {
    shared: Arc<Shared>,
    /// Every connection this thread has seen an event on, under the token of its socket.
    connections: HashMap<Token, RecvConnection>,
    /// When the connections are next to be gone through, see [`SWEEP_INTERVAL`].
    next_sweep: Instant,
    poll: Poll,
    events: Events,
}

impl Thread {

    /// The loop of the polling thread, which goes on until the client it belongs to is
    /// dropped, see [`Client::drop`].
    ///
    /// Whatever connections are left when this comes back go with it, which closes their
    /// sockets, everything that was waiting on one of them having been let go of.
    fn main(mut self) {

        while self.shared.running.load(Ordering::Relaxed) {
            match self.run() {
                Ok(()) => {}
                // A poll cut short by a signal is a poll that brought nothing back, and the
                // next one is simply made.
                Err(err) if err.kind() == io::ErrorKind::Interrupted => {}
                // The poll is what every connection of this client is read through, so one
                // that fails is the whole of them being over, and there is nothing this
                // thread can do about it but give them back.
                Err(err) => {
                    error!(%err, "Failed to poll the connections of the CORBA client, giving up on them");
                    break;
                }
            }
        }

        for (_, held) in self.connections.drain() {
            held.connection.fail();
        }

        // And everything the cache still knows of, which is not the same set: a connection
        // whose socket never came up in the poll is in no table of this thread at all, and
        // the caller waiting on its connect would otherwise wait for a thread that is gone.
        let mut connections = self.shared.connections.lock().unwrap();
        for (_, connection) in connections.by_token.drain() {
            if let Some(connection) = connection.upgrade() {
                connection.fail();
            }
        }
        connections.by_identity.clear();

    }

    /// One turn of the polling thread.
    fn run(&mut self) -> io::Result<()> {

        // The poll comes back for the sweep even when nothing has happened on any socket,
        // so that a client which is only letting go of connections, and not opening any,
        // still lets go of what is left of them.
        let timeout = self.next_sweep.saturating_duration_since(Instant::now());
        self.poll.poll(&mut self.events, Some(timeout))?;

        self.handle_events();

        if Instant::now() >= self.next_sweep {
            self.sweep();
            self.next_sweep = Instant::now() + SWEEP_INTERVAL;
        }

        Ok(())

    }

    /// Let go of a connection that is over, which closes its socket, the system then taking
    /// it out of the poll on its own, and take it out of the cache so that the next caller
    /// reaching that server opens one of its own.
    fn forget(&mut self, token: Token, identity: Identity) {

        self.connections.remove(&token);

        let mut connections = self.shared.connections.lock().unwrap();

        // The token of a connection is its own, never reused, so this only ever takes out
        // the one that is over, see [`Shared::next_token`].
        let Some(gone) = connections.by_token.remove(&token) else {
            return;
        };

        // Its identity is not its own: another caller may have opened a connection to the
        // very same server, with the very same codesets and version, between this one
        // failing and this call, and what is held under that identity is then that other
        // connection and not this one, see [`Shared::connect`].
        if connections.by_identity.get(&identity).is_some_and(|held| Weak::ptr_eq(held, &gone)) {
            connections.by_identity.remove(&identity);
        }

    }

    /// Let go of the connections nothing is using, closing them.
    fn sweep(&mut self) {

        // Held across the whole sweep, and taken before the counts are read: a caller
        // reaching a server takes this very lock to look a connection up and raises its
        // count by upgrading out of it, see [`Shared::connect`]. Reading the counts under
        // it is what keeps a connection from being let go of here at the very instant it is
        // handed to a caller, which would leave that caller holding a connection this
        // thread no longer reads, and lose whatever of a message had arrived on it.
        let mut connections = self.shared.connections.lock().unwrap();

        // A connection this table is the only holder of is one nothing is reaching a server
        // through any more: letting go of it here is what closes its socket, which the
        // system then takes out of the poll on its own.
        self.connections.retain(|_, held| Arc::strong_count(&held.connection) > 1);

        // And what the cache is left holding of the ones let go of is a reference to nothing
        // at all.
        connections.by_token.retain(|_, connection| connection.strong_count() > 0);
        connections.by_identity.retain(|_, connection| connection.strong_count() > 0);

    }

    /// Hand the events the poll has just given out to the connections 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 table 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 {

            // The waker only ever comes back as the client having been dropped, which the
            // loop finds out on its own: there is nothing to do with the event itself.
            if event.token() == WAKER_TOKEN {
                continue;
            }

            // The first event of a connection is the one that takes it out of the cache and
            // leaves it here, and the only one that costs anything at all.
            let recv = match self.connections.entry(event.token()) {
                hash_map::Entry::Occupied(entry) => entry.into_mut(),
                hash_map::Entry::Vacant(entry) => {

                    let connection = self.shared.connections.lock().unwrap()
                        .by_token.get(&event.token())
                        .and_then(Weak::upgrade);

                    let Some(connection) = connection else {
                        warn!(?event, "Received a poll event for no known connection, ignoring");
                        continue;
                    };

                    entry.insert(RecvConnection {
                        connection,
                        recv: RecvState {
                            buf: Buf::new(),
                            fragments: HashMap::new(),
                            adjacent: None,
                        },
                    })

                }
            };

            // An event on a socket whose connect has not resolved is that connect having
            // resolved, whichever readiness it comes as, and nothing else is to happen on
            // the connection until it has.
            if recv.connection.is_init() {

                let failed = recv.connection.complete_init();
                let identity = recv.connection.identity;

                if failed {
                    self.forget(event.token(), identity);
                }

                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 request that the
            // socket could not take whole finishes going out.
            let mut failure = None;

            if event.is_writable() {
                let mut send_state = recv.connection.send_state.lock().unwrap();
                failure = recv.connection.flush(&mut send_state.writer).err();
            }

            if failure.is_none() && event.is_readable() {
                failure = recv.connection.receive(&mut recv.recv).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 server closed the connection"));
            }

            // Either side of a connection failing is the whole of it being over: everything
            // waiting on it is let go of, and the connection with it.
            if let Some(err) = failure {

                let identity = recv.connection.identity;
                debug!(addr = %identity.addr, %err, "Lost the connection to the CORBA server");
                recv.connection.fail();

                self.forget(event.token(), identity);

            }

        }

        self.events = events;

    }
    
}

/// This is shared between the client, the thread, and the connections.
#[derive(Debug)]
struct Shared {
    /// Every connection this client has, under the two ways they are looked up.
    connections: Mutex<Connections>,
    /// Where the socket of a connection is registered, by whichever thread opens it.
    poll_registry: Registry,
    /// Hands out the token of the next socket registered with the poll. Tokens are never
    /// reused: the one of a connection that is gone simply never comes up again.
    next_token: AtomicUsize,
    /// Whether the polling thread is to go on, which is cleared by the client being dropped
    /// and by nothing else, see [`Client::drop`].
    running: AtomicBool,
}

impl Shared {

    /// Get the connection to the given address, with the given codesets and version, which
    /// is what a connection is told apart by, see [`Identity`]. Its socket is opened here if
    /// this client is not on that server yet.
    ///
    /// A connection opened here is not connected when it comes back, only started, and
    /// [`Connection::wait_init`] is what waits for it.
    fn connect(&self, addr: SocketAddr, codesets: Option<Codesets>, version: Option<Version>) -> Result<Arc<Connection>> {

        let identity = Identity {
            addr,
            codesets,
            version,
        };

        // The lock is held across the opening of the socket, which is a few system calls
        // that do not block, `connect(2)` on a non-blocking socket starting the connect and
        // answering `EINPROGRESS` rather than waiting for it. Two callers reaching the same
        // server at the same instant are what that is for: they end up on one connection,
        // one of them having opened it and both waiting for it.
        let mut connections = self.connections.lock().unwrap();

        if let Some(connection) = connections.by_identity.get(&identity).and_then(Weak::upgrade) {
            trace!(%addr, "Reusing the connection to the CORBA server");
            return Ok(connection);
        }

        trace!(%addr, "Connecting to the CORBA server");

        let mut stream = TcpStream::connect(addr)
            .map_err(|err| Error::new(ErrorKind::Corba)
                .with_append_description(format_args!("Could not open a socket to {}", addr))
                .with_source(Box::new(err)))?;

        // A request is a small message this side is about to wait for the reply of, so it
        // goes out as soon as it is written and not once the system has enough of them. It
        // matters the more here in that several callers share the connection: a request of
        // one of them waiting on the acknowledgement of another's is what this avoids.
        if let Err(err) = stream.set_nodelay(true) {
            debug!(%addr, %err, "Failed to disable Nagle's algorithm on the stream");
        }

        // Registering from here, while the polling thread sits in its poll, is what a
        // registry is for, and the readiness of a connect that resolved before this call
        // is not lost by it: registering samples the socket as it is, hands the event over
        // right away if it is ready already, and wakes a poll that is blocked on it. So
        // there is no thread to wake, and no window to register too late in.
        let token = Token(self.next_token.fetch_add(1, Ordering::Relaxed));
        self.poll_registry.register(&mut stream, token, Interest::WRITABLE | Interest::READABLE)
            .map_err(|err| Error::new(ErrorKind::Internal)
                .with_description("Failed to register the socket of a CORBA connection with the poll")
                .with_source(Box::new(err)))?;

        let connection = Arc::new(Connection {
            stream,
            identity,
            _token: token,
            init: Init {
                done: AtomicBool::new(false),
                state: Mutex::new(InitState::Connecting(Vec::new())),
            },
            next_request_id: AtomicU32::new(0),
            request_state: Mutex::new(RequestState {
                slots: HashMap::new(),
                lost: false,
            }),
            send_state: Mutex::new(SendState {
                writer: CdrWriter::new().with_codesets(identity.codesets).into_sender(),
                version: identity.version.unwrap_or(VERSION).min(VERSION),
                version_settled: false,
                codesets_sent: false,
            }),
        });

        connections.by_token.insert(token, Arc::downgrade(&connection));
        connections.by_identity.insert(identity, Arc::downgrade(&connection));

        Ok(connection)

    }

}

/// The connections of a client, under the token of their socket, which is what an event
/// carries, and under their identity, which is what a caller reaches a server through.
///
/// These watch the connections rather than hold them: what holds them is the table of the
/// polling thread, where every one of them ends up, and what closes one is that thread
/// letting go of it, see [`Thread::connections`]. So an entry here can upgrade to nothing,
/// which is what a connection that is over looks like from here, and those are swept when a
/// connection is opened, the one moment these ever grow, and every so often by the polling
/// thread, see [`Shared::connect`] and [`Thread::sweep`].
#[derive(Default, Debug)]
struct Connections {
    /// Every connection this client has, under the token of its socket.
    by_token: HashMap<Token, Weak<Connection>>,
    /// The very same connections, under their identity.
    by_identity: HashMap<Identity, Weak<Connection>>,
}

/// Identity of a connection, uniquely identify a connection that can be reused.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct Identity {
    /// The fully resolved socket address this is connected to.
    addr: SocketAddr,
    /// The codesets that the client wants to use with the server.
    codesets: Option<Codesets>,
    /// The explicit version requested for this connection, this might be downgraded.
    version: Option<Version>,
}

/// A connection tracking structure, this is shared within a `Arc`.
#[derive(Debug)]
struct Connection {
    /// The underlying MIO-capable TCP stream.
    stream: TcpStream,
    /// Unique identity of this connection, connections with the same identity can be 
    /// cached and then reused.
    identity: Identity,
    /// The token this connection's stream is registered with on the poll.
    _token: Token,
    /// How far the connect of this connection has got, see [`Connection::wait_init`].
    init: Init,
    /// Hands out the id of the next request on this connection, which wraps around at large
    /// values, which is perfectly fine: what a request id has to be unlike is the ones of the
    /// requests in flight beside it, and not every one that ever went out.
    next_request_id: AtomicU32,
    /// The requests waiting for their reply.
    request_state: Mutex<RequestState>,
    /// What has been written on this connection and has not gone out yet.
    ///
    /// This is taken before [`Connection::request_state`] wherever the two are held at
    /// once, which is what keeps them from ever being taken the other way around.
    send_state: Mutex<SendState>,
}

impl Connection {

    /// Wait for the connect of this connection to be over, which is no wait at all for one
    /// that is established already: what comes back is this connection being usable, or
    /// what its connect failed with.
    fn wait_init(&self) -> WaitInit<'_> {
        WaitInit(self)
    }

    /// Whether the connect of this connection has yet to resolve.
    fn is_init(&self) -> bool {
        !self.init.is_done()
    }

    /// Answer an event on the socket of a connection that is still connecting, and wake
    /// everyone waiting on it if that connect has resolved.
    ///
    /// Tells whether it resolved into a failure. An event is not the connect being over: a
    /// socket can come up ready with it still under way, and there is then another event to
    /// wait for.
    fn complete_init(&self) -> bool {

        // Steps 4 and 5 of the procedure `mio::net::TcpStream::connect` documents: the
        // error left on the socket, and then the peer it ended up connected to, are what
        // say which way the connect resolved.
        let err = match self.stream.take_error() {
            Ok(Some(err)) | Err(err) => err,
            Ok(None) => match self.stream.peer_addr() {
                Ok(_) => {
                    debug!(addr = %self.identity.addr, "Connected to the CORBA server");
                    self.init.set(InitState::Established);
                    return false;
                }
                // The connect is still under way, which is another event to wait for. This
                // is `ENOTCONN`, and `EINPROGRESS` on the systems that answer that instead,
                // which has no `ErrorKind` of its own.
                Err(err) if err.kind() == io::ErrorKind::NotConnected => return false,
                Err(err) => err,
            }
        };

        debug!(addr = %self.identity.addr, %err, "Failed to connect to the CORBA server");
        self.init.set(InitState::Failed(Some(err)));

        true

    }

    /// Send a request to an object of this connection and register what is waiting for its
    /// reply, which is what comes back: the request is out by the time this comes back, and
    /// what it gives back is only the reply of it.
    ///
    /// This never fails on its own: whatever went wrong is given back through the future
    /// instead, once it is awaited, see [`ClientRequestState::Failed`]. Nothing here is
    /// awaited before the request is either fully out or has failed to be, so a caller
    /// making several of these before awaiting any of them still sends every one of them
    /// right away.
    fn request(self: &Arc<Self>, object: &[u8], operation: &[u8], data: &dyn CdrWrite) -> ClientRequest {
        ClientRequest {
            state: match self.try_request(object, operation, data) {
                Ok(state) => state,
                Err(err) => ClientRequestState::Failed(Some(err)),
            },
        }
    }

    /// The fallible part of [`Self::request`], kept apart so that every way of failing goes
    /// through the same `?` rather than a state of its own to build by hand.
    fn try_request(self: &Arc<Self>, object: &[u8], operation: &[u8], data: &dyn CdrWrite) -> Result<ClientRequestState> {

        let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed);

        // Lock the send state to avoid requests to be interlieved.
        let mut send_state = self.send_state.lock().unwrap();
        let version = send_state.version;

        // The codesets travel with the first request that goes out and never again, so this
        // is the one that carries them, whichever it turns out to be. The request only
        // borrows it, so it is held here for as long as the request is.
        let announce = match self.identity.codesets {
            Some(codesets) if !send_state.codesets_sent => Some(ServiceContextKind::Codesets(codesets).build()?),
            _ => None,
        };

        // If the version is not settled yet, we have to encode the request for every
        // version before the one we are using to send the request. If a downgrade is
        // requested then we can easily resend the request. This should be relatively
        // free since the first request is usually a '_is_a' operation!
        let mut alternates = Vec::new();
        if !send_state.version_settled {
            for minor in 0..version.minor() {
                let alternate = Version::from_minor(minor).expect("every minor under a version is one");
                let mut writer = CdrWriter::with_capacity(REQUEST_MIN_LEN).with_codesets(self.identity.codesets);
                self.write_request(&mut writer, alternate, request_id, object, operation, data, announce.as_ref())?;
                alternates.push((alternate, writer.into_buf().freeze()));
            }
        }

        // The state goes in before anything goes out: the polling thread can be through with
        // the whole exchange before this even comes back, and a reply arriving before there
        // is anywhere to put it is a reply that is dropped.
        //
        // A connection that is over stays with whoever holds it, so this is where they are
        // told, see [`Self::fail`].
        {
            let mut request_state = self.request_state.lock().unwrap();
            if request_state.lost {
                return Err(Error::new(ErrorKind::Corba)
                    .with_description("The connection to the server is over"));
            }
            request_state.slots.insert(request_id, RequestSlot {
                wakers: Vec::new(),
                reply: None,
                alternates,
            });
        }

        trace!(request_id, %version, "Sending a request");
        self.write_request(&mut send_state.writer, version, request_id, object, operation, data, announce.as_ref())?;
        self.flush(&mut send_state.writer)?;

        // We just sent the codesets...
        send_state.codesets_sent = true;

        Ok(ClientRequestState::Sent {
            connection: Arc::clone(self),
            request_id,
        })

    }

    /// Tell the server that nothing is waiting for the reply of a request any more.
    ///
    /// It is a courtesy and not an order, see [`CancelRequest`]: a server is free to answer
    /// a request it was asked to drop, and the reply of one nothing is waiting for is
    /// dropped when it turns up, see [`dispatch`].
    fn cancel(&self, request_id: u32) -> Result<()> {

        // Lock the send state to avoid requests to be interlieved.
        let mut send_state = self.send_state.lock().unwrap();

        // The version this connection settled on and not the one the reference published: a
        // server that refused the latter would refuse this message in it just the same, see
        // [`Self::downgrade`].
        let version = send_state.version;

        trace!(request_id, %version, "Cancelling a request");
        let message = Message::CancelRequest(CancelRequest::new(version, request_id));
        self.write_message(&mut send_state.writer, &message, None)?;
        self.flush(&mut send_state.writer)

    }

    /// Lay a message down into the octets of a buffer, behind whatever they already hold.
    fn write_message(&self, writer: &mut CdrWriter, message: &Message, body: Option<&dyn CdrWrite>) -> Result<()> {

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

        // Now that everything is reverted we might return the error!
        res

    }

    /// Lay a request down into the octets of a buffer, behind whatever they already hold, in
    /// one of the versions it may go out in.
    fn write_request(&self, writer: &mut CdrWriter, version: Version, request_id: u32, object: &[u8], operation: &[u8], data: &dyn CdrWrite, announce: Option<&ServiceContext>) -> Result<()> {

        let mut request = Request::new(version, request_id, Target::new_key(object), operation)
            .with_response_expected(true);

        if let Some(announce) = announce {
            request = request.with_service_contexts(slice::from_ref(announce));
        }

        self.write_message(writer, &Message::Request(request), Some(data))

    }

    /// 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(&self, sender: &mut CdrSender) -> Result<()> {

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

    }

    /// Read everything the socket has to give into the buffer, and hand every whole message
    /// that makes to whoever is waiting for 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(&self, recv: &mut RecvState) -> Result<()> {

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

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

        // Every whole message that made out of what arrived, and not only the first. A
        // server that closed right behind its last reply is one whose reply came in the
        // very read that found the end, so what arrived is taken before the connection is
        // given up on below.
        while self.dispatch(recv)? { }

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

        Ok(())

    }

    /// Take the message at the front of what has arrived and put it where its request is
    /// waiting, 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.
    fn dispatch(&self, recv: &mut RecvState) -> 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 mut reader = CdrReader::new(&recv.buf);
        let message = reader.read::<Message>()
            .map_err(|err| Error::new(ErrorKind::Corba)
                .with_description("The server sent a GIOP message that could not be read")
                .with_parent(err))?;

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

        let (request_id, status, body_at, endian, message) = match message {

            // The length is the one thing the server 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::Corba)
                        .with_description(format!("The server 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`].
                let buf = recv.buf.make_mut();
                buf.try_reserve(total_len.saturating_sub(buf.len()))?;

                return Ok(false);

            }

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

                // A whole message in the middle of a fragmented one is only a server
                // 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::Corba)
                        .with_description("The server sent a whole GIOP message in the middle of one it was sending in fragments"));
                }

                let (request_id, status, body_at, endian) =
                    (reply.request_id(), reply.status(), body_at, reply.endian());

                (request_id, status, body_at, endian, recv.buf.split_to(message_len))

            }

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

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

                // Read out of the reply before its octets leave the buffer, the version
                // being what says how its fragments are followed.
                let request_id = reply.request_id();
                let version = reply.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::Corba)
                        .with_description("The server 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::Corba)
                        .with_description(format!("The server 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::Corba)
                        .with_description(format!("The server 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);
                let part = writer.into_buf().freeze();

                res.map_err(|err| Error::new(ErrorKind::Corba)
                    .with_description("A GIOP message could not be put back together")
                    .with_parent(err))?;

                // Nothing of the message could be read while it was in parts, its stream
                // stopping in the middle, so this is the first it is read at all. It is
                // read right here and not set aside for another turn the way the server
                // does with one of its own, see [`super::server::RecvState`]: everything
                // wanted of a reply is four numbers of its header, where a request has to
                // go back through the whole of a dispatch to be answered.
                let (request_id, status, body_at, endian) = {

                    let message = CdrReader::new(&part).read::<Message<'_>>()
                        .map_err(|err| Error::new(ErrorKind::Corba)
                            .with_description("The server sent a GIOP message that could not be read")
                            .with_parent(err))?;

                    let Message::Reply(reply) = message else {
                        return Err(Error::new(ErrorKind::Corba)
                            .with_description("The server sent the fragments of a GIOP message that is not a reply"));
                    };

                    (reply.request_id(), reply.status(), part.len() - reply.body_len(), reply.endian())

                };

                (request_id, status, body_at, endian, part)

            }

            // The server made nothing of what it was sent, which is the version it
            // went out in that it makes nothing of, and the request is sent again in
            // the one it answers in.
            Message::MessageError(refused) => {
                let _ = recv.buf.split_to(message_len);
                self.downgrade(refused)?;
                return Ok(true);
            }

            _ => {
                return Err(Error::new(ErrorKind::Corba)
                    .with_description("The server sent a GIOP message that is not a reply"));
            }

        };

        // Very short lock to define the version as settled, since this is a reply!
        self.send_state.lock().unwrap().version_settled = true;

        // Lock the request in order to wake the all waiters on this request!
        let mut request_state = self.request_state.lock().unwrap();

        // A reply of a request nothing is waiting for any more is one whose caller gave
        // up, and there is nothing to be done with it.
        let Some(state) = request_state.slots.get_mut(&request_id) else {
            trace!(request_id, "Dropping the reply of a request nothing is waiting for");
            return Ok(true);
        };

        if state.reply.is_some() {
            trace!(request_id, "Dropping a second reply for a request");
            return Ok(true);
        }

        state.reply = Some(ClientReply {
            body: message,
            codesets: self.identity.codesets,
            status,
            body_at,
            endian,
        });

        let wakers = mem::take(&mut state.wakers);
        drop(request_state);

        for waker in wakers {
            waker.wake();
        }


        Ok(true)

    }

    /// Come back with the version a server answers in, and send again everything that went
    /// out in the one it refused.
    ///
    /// A server that makes nothing of the version it was sent answers with a message error
    /// carrying the highest version it does speak, which is therefore an earlier one than
    /// the one it was sent, see section 9.4.1 of the specification: that is the one to come
    /// back with, and there is no point trying anything in between.
    fn downgrade(&self, refused: Version) -> Result<()> {

        let mut send_state = self.send_state.lock().unwrap();

        // Anything else is not a version to come back with: either it answers a message this
        // side already dealt with, several requests having been in flight when the version
        // was refused, or it is a server making nothing of octets in a version it does
        // speak, which an earlier one would not help with either.
        if refused >= send_state.version {
            debug!(%refused, "Received a GIOP message error for a version there is no falling back from, ignoring");
            return Ok(());
        }

        debug!(%refused, "The CORBA server refused the GIOP version, falling back to the one it answered in");
        send_state.version = refused;

        // The request that announced the codesets went out in the refused version as well,
        // so the server was never told, and whichever request tells it now is the one that
        // carries them.
        send_state.codesets_sent = false;

        let mut request_state = self.request_state.lock().unwrap();
        let mut void = Vec::new();

        // We go through all inflight requests and resend them with a newer version...
        for (request_id, state) in request_state.slots.iter_mut() {

            let alternate = state.alternates.iter()
                .position(|(version, _)| *version == refused)
                .map(|at| state.alternates.swap_remove(at).1);

            state.alternates.retain(|(version, _)| *version < refused);

            let Some(alternate) = alternate else {
                void.push(*request_id);
                continue;
            };

            trace!(request_id, version = %refused, "Sending a request again");

            send_state.writer.interleave(alternate);

        }

        // A request that was laid down in no version to come back with is one that will
        // never be answered, and the caller waiting on it is woken to be told so.
        for request_id in void {
            let Some(state) = request_state.slots.remove(&request_id) else {
                continue;
            };
            for waker in state.wakers {
                waker.wake();
            }
        }
        drop(request_state);

        self.flush(&mut send_state.writer)

    }

    /// Let go of everything waiting on this connection, which is what it being over comes
    /// to: a request whose slot is gone is one that will never be answered, and the caller
    /// waiting on it is woken to be told exactly that.
    ///
    /// A connect that has not resolved is settled here as well: a connection given up on
    /// before it was ever established is one nobody will ever hear from, and the callers
    /// waiting on it would otherwise wait forever, see [`Self::wait_init`]. One that is
    /// established already keeps what it came to, this not being a connect that failed.
    fn fail(&self) {

        self.init.set(InitState::Failed(Some(io::Error::new(io::ErrorKind::ConnectionAborted,
            "the connection was given up on before its connect resolved"))));

        // A request whose reply arrived is through, and keeps it for whoever waits on it:
        // a server closing right behind its last reply is one whose reply arrives in the
        // very read that finds the connection over. Only the ones still waiting fail.
        let mut wakers = Vec::new();
        
        {   let mut request_state = self.request_state.lock().unwrap();
            request_state.lost = true;
            request_state.slots.retain(|_, slot| {
                if slot.reply.is_some() {
                    return true;
                }
                wakers.append(&mut slot.wakers);
                false
            });
        }

        for waker in wakers {
            waker.wake();
        }

    }

}

/// How far the connect of a connection has got, and what it came to.
///
/// The flag is the whole of what the polling thread reads, which it does on every event, and
/// it is only ever set under the lock that settles the state, so what it says never runs
/// ahead of what is behind it. Everything a caller waiting on the connect does is behind
/// that lock instead, which it does once per connect and not once per event.
#[derive(Debug)]
struct Init {
    /// Whether the connect has resolved, either way.
    done: AtomicBool,
    /// Who is waiting for it to resolve until it has, and what it came to afterward.
    state: Mutex<InitState>,
}

impl Init {

    /// Whether the connect has resolved, which is one atomic load.
    fn is_done(&self) -> bool {
        self.done.load(Ordering::Acquire)
    }

    /// Settle what the connect came to, and wake everyone it was keeping waiting.
    ///
    /// A connect that has resolved already keeps what it came to and nothing is woken: the
    /// first answer is the one that holds, and a connection given up on long afterward is
    /// not one that failed to connect, see [`Connection::fail`].
    fn set(&self, state: InitState) {

        let mut guard = self.state.lock().unwrap();

        let InitState::Connecting(wakers) = &mut *guard else {
            return;
        };

        let wakers = mem::take(wakers);
        *guard = state;

        // Release ordering in order to ensure coherency with is_done(), this ensures 
        // that, if is_done() returns true, the state above takes effect on remote thread.
        self.done.store(true, Ordering::Release);
        drop(guard);

        for waker in wakers {
            waker.wake();
        }

    }

}

/// What the connect of a connection has come to, if it has.
#[derive(Debug)]
enum InitState {
    /// The connect has not resolved yet, and these are waiting for it to.
    Connecting(Vec<task::Waker>),
    /// Done and the connection is usable.
    Established,
    /// Done but with an error.
    Failed(Option<io::Error>),
}

/// The future of [`Connection::wait_init`].
struct WaitInit<'a>(&'a Connection);

impl Future for WaitInit<'_> {

    type Output = Result<()>;

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {

        let addr = self.0.identity.addr;
        let mut state = self.0.init.state.lock().unwrap();

        let wakers = match &mut *state {
            InitState::Connecting(wakers) => wakers,
            InitState::Established => return task::Poll::Ready(Ok(())),
            InitState::Failed(err) => return task::Poll::Ready(Err(match err.take() {
                Some(err) => Error::new(ErrorKind::Corba)
                    .with_append_description(format_args!("Could not connect to {addr}"))
                    .with_source(Box::new(err)),
                None => Error::new(ErrorKind::Corba)
                    .with_append_description(format_args!("Could not connect to {addr}, what it \
                        failed with having gone to the caller that asked first")),
            })),
        };

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

        task::Poll::Pending

    }

}

/// What the polling thread holds of a connection: the connection itself, and what it has
/// read off its socket and not framed into a whole message yet.
struct RecvConnection {
    connection: Arc<Connection>,
    recv: RecvState,
}

/// What the polling thread has of a connection that is not 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 answers.
    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>,
}

/// The requests of a connection waiting for their reply, see [`Connection::request_state`].
#[derive(Debug)]
struct RequestState {
    /// The slot of every request waiting for its reply, under the id of that request. A
    /// slot that is gone is a request that will never be answered, which is what a
    /// connection that is over leaves behind, see [`Connection::fail`].
    slots: HashMap<u32, RequestSlot>,
    /// Set once the connection is over, see [`Connection::fail`]: the polling thread no
    /// longer watches it, so a request sent on it would never be answered.
    lost: bool,
}

/// Such object represent an inflight request.
#[derive(Debug)]
struct RequestSlot {
    /// Who is waiting for the reply of that request.
    wakers: Vec<task::Waker>,
    /// The octets the reply arrived as, once one has: what waits here has to be owned, a
    /// reply borrowing the message it is read out of.
    reply: Option<ClientReply>,
    /// The very same request laid down in each version this side could still come back with,
    /// which are none once the version is settled.
    ///
    /// The arguments of a request are gone by the time a server refuses the version it went
    /// out in, so an encoding not done then cannot be done later: it is laid down once per
    /// version while there is still one to come back with, which costs a few small buffers
    /// on the first request of a connection and nothing at all after, see
    /// [`Connection::downgrade`].
    alternates: Vec<(Version, Buf<u8, 1>)>,
}

/// The sending side of a connection, which is one thing at a time: what goes out goes out
/// whole, and the octets of the requests 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,
    /// The GIOP version every request of this connection goes out in, which starts at what
    /// the reference published and only ever goes down: a server that makes nothing of a
    /// version answers with a message error, and this side comes back with the one it
    /// answered in, see [`Connection::downgrade`].
    version: Version,
    /// Set to true when the GIOP version above have been settlte and we know that the
    /// server agrees on using it for the rest of the communication!
    version_settled: bool,
    /// Whether a request carrying the codesets of this connection has gone out, which is
    /// what makes them hold for everything behind it: the server is told once.
    codesets_sent: bool,
}

/// The reply of a request that has gone out, see [`Connection::request`].
///
/// Dropping this is giving up on that reply, and takes the slot of the request with it: one
/// that turns up afterward is a reply nothing is waiting for, and is dropped.
pub struct ClientRequest {
    state: ClientRequestState,
}

/// What a [`ClientRequest`] is waiting on, or the failure it already came to before
/// anything went out at all, see [`Connection::request`].
enum ClientRequestState {
    /// The request went out, and this is what its reply is being waited for on.
    Sent {
        connection: Arc<Connection>,
        request_id: u32,
    },
    /// The request could not be laid down at all, so nothing went out and there is
    /// nothing to cancel: the failure itself is given back the first time this is
    /// polled, [`None`] on every one after that, see [`ClientRequest::poll`].
    Failed(Option<Error>),
}

impl Future for ClientRequest {

    type Output = Result<ClientReply>;

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {

        let this = self.get_mut();

        let (connection, request_id) = match &mut this.state {
            ClientRequestState::Sent { connection, request_id } => (connection, *request_id),
            ClientRequestState::Failed(err) => {
                // NOTE: The error should never trigger since calling poll a second time 
                // is unspecified, we can panic if we want!
                let err = err.take().unwrap_or_else(|| Error::new(ErrorKind::Corba)
                    .with_description("This request's failure was already given to a previous poll"));
                return task::Poll::Ready(Err(err));
            }
        };

        let mut request_state = connection.request_state.lock().unwrap();

        let Some(slot) = request_state.slots.get_mut(&request_id) else {
            return task::Poll::Ready(Err(Error::new(ErrorKind::Corba)
                .with_description("The connection to the server failed while its reply was being waited for")));
        };

        let Some(reply) = slot.reply.take() else {

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

            return task::Poll::Pending;

        };

        let _ = request_state.slots.remove(&request_id);
        task::Poll::Ready(Ok(reply))

    }

}

impl Drop for ClientRequest {

    fn drop(&mut self) {

        let ClientRequestState::Sent { connection, request_id } = &self.state else {
            // Nothing was ever sent, so there is nothing to cancel.
            return;
        };

        let Some(state) = connection.request_state.lock().unwrap().slots.remove(request_id) else {
            return;
        };

        if state.reply.is_some() {
            return;
        }

        if let Err(err) = connection.cancel(*request_id) {
            debug!(request_id, %err, "Failed to cancel the request");
        }

    }

}

/// The reply of a request, as the octets it arrived as: what waits in a request state has
/// to be owned.
///
/// [`giop::Reply`]: super::giop::Reply
#[derive(Debug)]
pub struct ClientReply {
    body: 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 is, which the header said on the way in.
    status: ReplyStatus,
    /// Where the body of the reply starts 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 peer are in, which is not this side's to choose.
    endian: Endian,
}

impl ClientReply {

    /// Read this reply's octets from an arbitrary position rather than the very start of
    /// the body, for rewinding a reader back to where an earlier one of several entries
    /// decoded out of the very same reply left off, instead of decoding the status (and,
    /// for a `UserException`, the name) all over again on every one of them, see
    /// [`Self::read`].
    pub fn read_at(&self, position: usize) -> CdrReader<'_> {
        // The header was read on the way in and nothing of it is read again: where the
        // body starts, or any position past it, is what it left behind, see
        // [`Connection::dispatch`].
        let mut reader = CdrReader::new(&self.body).with_endian(self.endian).with_codesets(self.codesets);
        reader.set_position(position);
        reader
    }

    /// Read the reply back out of the octets it arrived as: its body, in place in those very
    /// octets, and the id of the user exception it carries if it carries one.
    pub fn read(&self) -> Result<(CdrReader<'_>, Option<Cow<'_, str>>)> {

        let mut reader = self.read_at(self.body_at);

        match self.status {
            ReplyStatus::NoException => Ok((reader, None)),
            ReplyStatus::UserException => {
                let exception_id = reader.read_string()?;
                Ok((reader, Some(exception_id)))
            }
            // Which names itself, decodes its own minor code, and says how far the operation
            // got, see its [`Display`].
            ReplyStatus::SystemException => {
                let exception = reader.read::<SystemException<'_>>()?;
                Err(Error::new(ErrorKind::Corba)
                    .with_description(exception.to_string()))
            }
            // The object is somewhere else, or wants to be addressed some other way. Both
            // are the server asking for the request to be sent again differently, which this
            // implementation does not do: a Tango device is where its reference says it is.
            status => Err(Error::new(ErrorKind::NotHandled)
                .with_description(format!("The server answered with a {status:?}, which this implementation does not follow"))),
        }

    }

}

/// Internal type describing IOR profile that this client understands for connecting to
/// an object with an IOR.
#[derive(Debug)]
struct IorProfile {
    /// The object key of the profile, which is what a request addresses.
    object: Arc<[u8]>,
    /// The codesets negociated out of the components of the profile, if any.
    codesets: Option<Codesets>,
    /// The GIOP version the profile publishes, see [`profile_version`].
    version: Version,
    /// The addresses to try, the one of the profile itself first and the alternate ones
    /// behind it, an address the IOR declares twice being left out.
    addrs: Vec<(String, u16)>,
}

/// Internal function to parse an IOR into an iterator that returns candidate TCP 
/// addresses
fn parse_ior_profiles(ior: &str, type_id_filter: &[&str]) -> Result<Vec<IorProfile>> {

    let raw_ior = ior.strip_prefix("IOR:").unwrap_or(ior);
    let buf = BufMut::from_hex(raw_ior)
        .ok_or_else(|| Error::new(ErrorKind::InvalidData)
            .with_description("The IOR of the device is not valid hexadecimal")
            .with_append_description(raw_ior))?
        .freeze();
    
    let mut reader = CdrReader::new(&buf);
    reader.read_endian()?;

    let ior = reader.read::<Ior>()?;

    if !type_id_filter.contains(&&*ior.type_id) {
        return Err(Error::new(ErrorKind::InvalidData)
            .with_description(format!("Invalid type id of the IOR, got {}, expected one of {type_id_filter:?}", ior.type_id))); 
    }

    let mut profiles = Vec::with_capacity(ior.profiles.len());

    for profile in &*ior.profiles {
        match profile.parse()? {
            Some(ProfileKind::Iiop(iiop_service)) => {

                let mut codesets = None;

                // The candidates are tried in the order they are declared in, which is
                // the address of the profile itself first, and then the alternate ones,
                // leaving out an address that the IOR declares twice, which would
                // otherwise be tried twice. There are only ever a handful of them, so
                // looking each one up in the candidates is cheaper than hashing it.
                let mut addrs = Vec::with_capacity(1);
                addrs.push((iiop_service.host.to_string(), iiop_service.port));

                for component in &*iiop_service.components {
                    match component.parse()? {
                        Some(ComponentKind::Codesets(codesets_comp)) => {

                            if let Some(native_codeset) = codesets_comp.parse_native_codeset()
                            && let Some(native_wide_codeset) = codesets_comp.parse_native_wide_codeset() {

                                let conversion_codesets = codesets_comp.parse_conversion_codesets().collect::<Vec<_>>();
                                let conversion_wide_codesets = codesets_comp.parse_conversion_wide_codesets().collect::<Vec<_>>();

                                if let Some(codeset) = codeset::negociate_with_server(native_codeset, &conversion_codesets)
                                && let Some(wide_codeset) = codeset::negociate_with_server(native_wide_codeset, &conversion_wide_codesets) {

                                    trace!(?codeset, ?wide_codeset, "Negotiated the codesets of the IOR");
                                    codesets = Some(Codesets {
                                        codeset,
                                        wide_codeset,
                                    });

                                } else {
                                    debug!(?native_codeset, ?native_wide_codeset, "Failed to negotiate the codesets of the IOR, using the defaults");
                                }

                            } else {
                                debug!("Failed to parse the codesets of the IOR, using the defaults");
                            }
                        }
                        Some(ComponentKind::AltIiopAddr(alt_addr_comp)) => {
                            let host_and_port = (alt_addr_comp.host.to_string(), alt_addr_comp.port);
                            if !addrs.contains(&host_and_port) {
                                addrs.push(host_and_port);
                            }
                        }
                        // Components this implementation has no use for, which are many.
                        Some(_) | None => {}
                    }
                }

                profiles.push(IorProfile {
                    object: iiop_service.object_key.to_vec().into(),
                    codesets,
                    version: iiop_service.version,
                    addrs,
                });

            }
            // Profiles of other protocols than IIOP, which are of no use either.
            Some(_) | None => {}
        }
    }

    if profiles.is_empty() {
        return Err(Error::new(ErrorKind::InvalidData)
            .with_description("Could not find an IIOP profile to connect to in the given IOR"));
    }

    Ok(profiles)
    
}

#[cfg(test)]
mod tests {

    use std::net::TcpListener;

    use crate::future::block_on_timeout;

    use super::*;

    const TIMEOUT: Duration = Duration::from_secs(5);

    /// A connection the server closes fails the request waiting on it, and every request
    /// made on it afterward, rather than leaving them waiting for a reply that never comes.
    #[test]
    fn connection_lost() {

        let client = Client::new().unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();

        let object = block_on_timeout(client.connect_object_iiop("127.0.0.1", port, None, None, Arc::from(&b"object"[..])), TIMEOUT)
            .unwrap()
            .unwrap();
        drop(listener.accept().unwrap());

        let err = block_on_timeout(object.request(b"_non_existent", &()), TIMEOUT).unwrap().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Corba);

        let err = block_on_timeout(object.request(b"_non_existent", &()), TIMEOUT).unwrap().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::Corba);

    }

}