rstango 0.4.0

//! Common Data Representation codec.
//! 
//! Specification: <https://www.omg.org/spec/CORBA/3.4/Interoperability/PDF>
//!                Section 9.2.1

use std::ops::{Deref, DerefMut, Range};
use std::collections::TryReserveError;
use std::fmt::{self, Debug};
use std::{mem, slice};
use std::borrow::Cow;
use std::sync::Arc;

use widestring::{Utf16Str, Utf16String};

use crate::value::{DevEnum, Scalar, Spectrum};
use crate::error::{Error, ErrorKind, Result};
use crate::buf::{Buf, BufMut, PackedElement};

use super::codeset::{AnyCodeset, Codeset, CodesetDef, WideCodeset, WideCodesetDef, Codesets};
use super::endian::Endian;


// ==================== //
//        Reader        //
// ==================== //

/// A structure for reading a CDR buffer, the internal buffer is not necessarily aligned
/// to be properly read nor has the correct endianness. So every reading operation returns
/// a clone-on-write buffer in case a slice referencing the input buffer is aligned and
/// with the correct endianness.
///
/// # Alignment
///
/// CDR aligns every primitive on its own size, relative to the origin of the stream, and
/// this reader tracks that alignment *logically*, from the offset of its cursor and not
/// from the address of the octets. So the buffer can start anywhere, and an encapsulation,
/// which restarts the alignment at its own origin, is simply read through a reader of its
/// own borrowing it in place, see [`Self::read_encapsulation`].
///
/// The address is only ever looked at to decide whether a slice of numbers can be borrowed
/// in place, which additionally requires the endianness to be the native one, and it is
/// copied into an owned and aligned buffer when it cannot, hence the clone-on-write.
///
/// # Sharing
///
/// A reader always reads the octets of a [`Buf`], and the arrays it reads as [`Buf`]s
/// share that buffer rather than borrow it: they outlive the reader, and the buffer, see
/// [`Self::read_buf`]. Anything else it reads still borrows the octets, or copies them.
///
/// # Rewinding
///
/// The octets are never modified, byte swapping happening in the copy when it is needed,
/// so this reader is [`Clone`] and cloning it is how a rewind point is kept: read ahead
/// through the clone, and everything that was read stays valid and untouched.
#[derive(Clone)]
pub struct CdrReader<'a> {
    /// The octets being read, which the CDR alignment is relative to the first one of, and
    /// which are never modified, so that they can be borrowed in place and read again.
    buf: &'a [u8],
    /// The buffer those octets are in, which the arrays read out of them are shared out of.
    owner: &'a Buf<u8, 1>,
    /// Offset in the buffer the next primitive is read at, which the CDR alignment is
    /// derived from, and which is the length of the buffer once it is fully read.
    pos: usize,
    /// Endianness of the octets that are not read yet, which the byte order flag of the
    /// message or of the encapsulation being read gives, see [`Self::read_endian`].
    endian: Endian,
    /// Codesets to use for the (wide) strings, when negotiated.
    codesets: Option<Codesets>,
}

impl<'a> CdrReader<'a> {

    /// A reader over the octets of that buffer, which shares the arrays it reads out of it,
    /// see [`Self::read_buf`].
    pub fn new(buf: &'a Buf<u8, 1>) -> Self {
        Self {
            buf: buf.as_slice(),
            owner: buf,
            pos: 0,
            endian: Endian::NATIVE,
            codesets: None,
        }
    }

    /// A reader over no octets at all.
    pub fn empty() -> CdrReader<'static> {
        static EMPTY: Buf<u8, 1> = Buf::new();
        CdrReader::new(&EMPTY)
    }

    /// A reader over some of the octets of this one, as a stream of its own: its CDR
    /// alignment restarts at the first of them, and it is left on the native byte order and
    /// on no codesets, like any new reader.
    ///
    /// # Panics
    ///
    /// If those octets are not within the ones of this reader.
    pub fn substream(&self, bytes: &'a [u8]) -> Self {
        let range = self.buf.as_ptr_range();
        assert!(range.start <= bytes.as_ptr() && bytes.as_ptr_range().end <= range.end,
            "the octets of a CDR substream must be within its reader");
        Self { buf: bytes, ..Self::new(self.owner) }
    }

    #[inline]
    pub const fn endian(&self) -> Endian {
        self.endian
    }

    /// Set the endianness of the octets that are not read yet, which is what the byte
    /// order flag of the message or of the encapsulation being read gives.
    #[inline]
    pub const fn set_endian(&mut self, endian: Endian) {
        self.endian = endian;
    }

    #[inline]
    pub const fn with_endian(mut self, endian: Endian) -> Self {
        self.endian = endian;
        self
    }

    #[inline]
    pub const fn codesets(&self) -> Option<Codesets> {
        self.codesets
    }

    #[inline]
    pub const fn set_codesets(&mut self, codesets: Option<Codesets>) {
        self.codesets = codesets;
    }

    #[inline]
    pub const fn with_codesets(mut self, codesets: Option<Codesets>) -> Self {
        self.codesets = codesets;
        self
    }

    /// The codeset of the narrow strings, defaulting to the one mandated by the
    /// specification when no codeset has been negotiated.
    #[inline]
    pub fn codeset(&self) -> Codeset {
        self.codesets.map(|codesets| codesets.codeset).unwrap_or(Codeset::Iso8859_1)
    }

    /// The codeset of the wide strings, there is no default one for those.
    #[inline]
    pub fn wide_codeset(&self) -> Option<WideCodeset> {
        self.codesets.map(|codesets| codesets.wide_codeset)
    }

    /// The whole buffer being read, the octets that are already read included.
    #[inline]
    pub const fn buffer(&self) -> &'a [u8] {
        self.buf
    }

    /// Offset of the cursor from the origin of the buffer, which is what the CDR alignment
    /// is relative to.
    #[inline]
    pub const fn position(&self) -> usize {
        self.pos
    }

    #[inline]
    pub fn set_position(&mut self, pos: usize) -> usize {
        assert!(pos <= self.buf.len(), "cannot move the CDR position past the {} octets of the buffer", self.buf.len());
        mem::replace(&mut self.pos, pos)
    }

    /// Number of octets that are not read yet.
    #[inline]
    pub const fn remaining(&self) -> usize {
        self.buf.len() - self.pos
    }

    #[inline]
    pub const fn is_empty(&self) -> bool {
        self.remaining() == 0
    }

    /// Borrow the octets that are not read yet, without reading them.
    #[inline]
    pub const fn remaining_slice(&self) -> &'a [u8] {
        self.buf.split_at(self.pos).1
    }

    /// Try to skip a given region by length.
    #[track_caller]
    pub fn skip(&mut self, len: usize) -> Result<()> {
        let remaining_len = self.remaining();
        if remaining_len < len {
            return Err(invalid_data_err(format!("Failed to skip {len} octets, only {remaining_len} remain")));
        }
        self.pos += len;
        Ok(())
    }

    /// Internal function to read a raw slice of bytes of the given length and with the
    /// first byte aligned on the given alignment. Note that this alignment is the logical
    /// alignment and not necessarily the alignment of the first element's address.
    #[track_caller]
    fn read_raw_slice(&mut self, len: usize, align: usize) -> Result<&'a [u8]> {

        debug_assert!(align.is_power_of_two());
        
        let offset = align_offset(self.pos, align);
        let remaining_len = self.remaining();

        // Checked this way instead of 'offset + len' so that it cannot overflow.
        if offset > remaining_len || remaining_len - offset < len {
            return Err(invalid_data_err(format!("Failed to read {len} octets aligned to {align}, only {remaining_len} remain")));
        }

        self.pos += offset;
        let ret = &self.buf[self.pos..][..len];
        self.pos += len;

        Ok(ret)

    }

    /// Same as [`Self::read_raw_slice`] but the slice is moved into an owned buffer in case
    /// its address is not aligned for the type or its octets are not in the native order,
    /// and is borrowed in place when they are.
    #[track_caller]
    fn read_num_slice<T: CdrNum>(&mut self, len: usize) -> Result<Cow<'a, [T]>> {

        let size_and_align = mem::size_of::<T>();
        let byte_len = len.checked_mul(size_and_align)
            .ok_or_else(|| invalid_data_err(format!("Failed to read {len} elements of {size_and_align} octets, their size overflows")))?;

        let slice = self.read_raw_slice(byte_len, size_and_align)?;

        // A single byte type is always address-aligned and has no order of its own, so it
        // is always borrowed in place, whatever the endianness of the buffer is.
        if slice.as_ptr().addr() & (size_and_align - 1) == 0 && (size_and_align == 1 || self.endian.is_native()) {

            // SAFETY: The slice is properly address-aligned so we can cast it safely
            // for the requested length, because this length is a multiple of type size.
            Ok(Cow::Borrowed(unsafe { slice::from_raw_parts(slice.as_ptr().cast::<T>(), len) }))

        } else {

            // Here we know that it is required to move the read slice into it buffer to
            // ensure its alignment, we might also need to swap bytes for endianness.
            let mut owned = Vec::<T>::with_capacity(len);
            
            // SAFETY: We just reserved at least 'len' elements in the buffer, so the 
            // spare slice of elements should have 'byte_len' bytes of elements in its
            // buffer that we can copy to from the raw slice.
            unsafe { 
                owned.spare_capacity_mut().as_mut_ptr().cast::<u8>().copy_from_nonoverlapping(slice.as_ptr(), byte_len);
                owned.set_len(len);
            }

            // Now that the vector is fully initialized, we swap bytes in place.
            if !self.endian.is_native() {
                for value in &mut owned {
                    *value = T::swap_bytes(*value);
                }
            }

            Ok(Cow::Owned(owned))

        }

    }

    #[track_caller]
    fn read_num<T: CdrNum>(&mut self) -> Result<T> {

        let size_and_align = mem::size_of::<T>();
        let slice = self.read_raw_slice(size_and_align, size_and_align)?;

        // SAFETY: The slice holds exactly the octets of a single value, of which every bit
        // pattern is valid, and the read is only the unaligned one when the address is not
        // aligned for the type, which an architecture is free to make slower.
        let num = unsafe {
            let ptr = slice.as_ptr().cast::<T>();
            if ptr.is_aligned() {
                ptr.read()
            } else {
                ptr.read_unaligned()
            }
        };

        if self.endian.is_native() {
            Ok(num)
        } else {
            Ok(T::swap_bytes(num))
        }

    }

    /// Read a slice of octets, which is always borrowed in place because a single byte
    /// needs neither alignment nor byte swapping.
    #[track_caller]
    pub fn read_octet_slice(&mut self, len: usize) -> Result<Cow<'a, [u8]>> {
        self.read_raw_slice(len, 1).map(Cow::Borrowed)
    }

    #[track_caller]
    pub fn read_octet(&mut self) -> Result<u8> {
        self.read_num()
    }

    /// Read a slice of booleans, which is borrowed in place like the octets it is made of,
    /// see [`Self::read_octet_slice`].
    ///
    /// Only the 0 and 1 bit patterns are valid booleans, so this fails if any of the octets
    /// is something else.
    #[track_caller]
    pub fn read_bool_slice(&mut self, len: usize) -> Result<Cow<'a, [bool]>> {
        let ret = self.read_raw_slice(len, 1)?;
        if ret.iter().all(|&b| b <= 1) {
            // SAFETY: A bool has the same size and alignment as an u8, and we checked all
            // bytes to have a valid bit pattern to be casted to booleans, so the slice can
            // simply be reinterpreted.
            Ok(Cow::Borrowed(unsafe { slice::from_raw_parts(ret.as_ptr().cast::<bool>(), len) }))
        } else {
            Err(invalid_data_err("Failed to read booleans, one of the octets is neither 0 nor 1"))
        }
    }

    #[track_caller]
    pub fn read_bool(&mut self) -> Result<bool> {
        let ret = self.read_octet()?;
        match ret {
            0 => Ok(false),
            1 => Ok(true),
            _ => Err(invalid_data_err("Failed to read a boolean, its octet is neither 0 nor 1"))
        }
    }

    #[track_caller]
    pub fn read_ushort_slice(&mut self, len: usize) -> Result<Cow<'a, [u16]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_ushort(&mut self) -> Result<u16> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_short_slice(&mut self, len: usize) -> Result<Cow<'a, [i16]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_short(&mut self) -> Result<i16> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_ulong_slice(&mut self, len: usize) -> Result<Cow<'a, [u32]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_ulong(&mut self) -> Result<u32> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_long_slice(&mut self, len: usize) -> Result<Cow<'a, [i32]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_long(&mut self) -> Result<i32> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_ulonglong_slice(&mut self, len: usize) -> Result<Cow<'a, [u64]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_ulonglong(&mut self) -> Result<u64> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_longlong_slice(&mut self, len: usize) -> Result<Cow<'a, [i64]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_longlong(&mut self) -> Result<i64> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_float_slice(&mut self, len: usize) -> Result<Cow<'a, [f32]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_float(&mut self) -> Result<f32> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_double_slice(&mut self, len: usize) -> Result<Cow<'a, [f64]>> {
        self.read_num_slice(len)
    }

    #[track_caller]
    pub fn read_double(&mut self) -> Result<f64> {
        self.read_num()
    }

    #[track_caller]
    pub fn read_sequence_len(&mut self) -> Result<usize> {
        usize::try_from(self.read_ulong()?)
            .map_err(|e| invalid_data_err("Failed to read a sequence length, it does not fit into a usize")
                .with_source(Box::new(e)))
    }

    /// Read a string as the raw octets it is on the wire, its terminating nul checked for
    /// and stripped, with no codeset applied to them at all.
    ///
    /// This is for the strings no codeset can honestly be applied to, which are the ones
    /// that come before the negotiation saying which codeset they are in: the operation name
    /// of a GIOP 1.2 request is behind the target address and in front of the service
    /// contexts, one of which is that very negotiation.
    ///
    /// The octets are borrowed in place like any other run of them, see
    /// [`Self::read_octet_slice`].
    #[track_caller]
    pub fn read_raw_string(&mut self) -> Result<&'a [u8]> {

        // The length of a string counts its terminating nul octet in.
        let len = self.read_sequence_len()?;
        if len == 0 {
            return Err(invalid_data_err("Failed to read a string, its length is zero"));
        }

        let raw = self.read_raw_slice(len, 1)?;

        if raw[len - 1] != 0 {
            return Err(invalid_data_err("Failed to read a string, it has no terminating nul octet"));
        }

        Ok(&raw[..len - 1])

    }

    /// Read a string, decoded with this reader's codeset, borrowing the buffer in place
    /// when the codeset allows it. Note that the octets themselves are always borrowed,
    /// only the codeset decides whether the decoded string can be one of them.
    #[track_caller]
    pub fn read_string(&mut self) -> Result<Cow<'a, str>> {

        let codeset = self.codeset();
        let raw = self.read_raw_string()?;

        codeset.decode_to_cow(raw)
            .ok_or_else(|| invalid_data_err(format!("Failed to read a string, decoding it with {codeset:?} failed")))

    }

    /// Read a wide string, decoded with this reader's wide codeset, see
    /// [`Self::read_string`]. Its length is in octets and not in code units, and those are
    /// only borrowed in place when their address is aligned and their order is the native
    /// one, see [`Self::read_ushort_slice`], so the decoded string is owned when they are
    /// not: it cannot borrow the copy they were read into.
    ///
    /// The terminating nul code unit is optional because GIOP 1.1 counts it in and GIOP 1.2
    /// does not, so it is only stripped when it is actually there.
    #[track_caller]
    pub fn read_wstring(&mut self) -> Result<Cow<'a, Utf16Str>> {

        let codeset = self.wide_codeset()
            .ok_or_else(|| invalid_data_err("Failed to read a wide string, no wide codeset was negotiated"))?;

        // The length of a wide string is in octets and not in code units.
        let len = self.read_sequence_len()?;
        if len % 2 != 0 {
            return Err(invalid_data_err(format!("Failed to read a wide string, its length of {len} octets is invalid")));
        }

        let mut raw = self.read_ushort_slice(len / 2)?;

        // The nul code unit is only there in GIOP 1.1, so it is stripped when it is, and an
        // empty wide string has none at all.
        if raw.last() == Some(&0) {
            match &mut raw {
                Cow::Borrowed(raw) => { *raw = &raw[..raw.len() - 1] }
                Cow::Owned(raw) => { raw.pop(); }
            }
        }

        // Code units read into a copy of their own cannot be borrowed past it.
        let decoded = match raw {
            Cow::Borrowed(raw) => codeset.decode_to_cow(raw),
            Cow::Owned(raw) => codeset.decode_to_cow(&raw).map(|s| Cow::Owned(s.into_owned())),
        };

        decoded.ok_or_else(|| invalid_data_err(format!("Failed to read a wide string, decoding it with {codeset:?} failed")))

    }

    /// These octets, which the reader just read, as a slice of the buffer they are in.
    fn share(&self, raw: &'a [u8]) -> Buf<u8, 1> {
        self.owner.slice_ref(raw)
    }

    /// Read an array of `len` values, without the length that precedes a sequence, as the
    /// data they are: shared out of the buffer this reader reads when their octets are
    /// aligned for the type and in the native order, and copied otherwise.
    #[track_caller]
    pub fn read_buf<T: CdrElement>(&mut self, len: usize) -> Result<Spectrum<T>> {

        let size = mem::size_of::<T>();
        let byte_len = len.checked_mul(size)
            .ok_or_else(|| invalid_data_err(format!("Failed to read {len} elements of {size} octets, their size overflows")))?;

        let raw = self.read_raw_slice(byte_len, size)?;
        let mut buf = self.share(raw);

        // The octets of the other order are swapped value by value, into a copy of their own
        // since the message is shared, and a single octet has no order of its own.
        if size > 1 && !self.endian.is_native() {
            buf.make_mut().as_mut_slice().chunks_exact_mut(size).for_each(<[u8]>::reverse);
        }

        // The buffer holds exactly the octets of `len` values, so only an invalid one fails.
        buf.cast([len])
            .map_err(|_| invalid_data_err(format!("Failed to read an array of {len} values, one of them is invalid")))

    }

    /// Read an array of `len` strings, decoded with this reader's codeset, as the data they
    /// are: shared out of the buffer this reader reads when the codeset leaves every one of
    /// them as it is on the wire, and copied otherwise.
    ///
    /// The strings are shared until the first one that the codeset does not leave as it
    /// is, from which they are all copied into a buffer of their own, the ones before it
    /// included, out of the octets already read.
    #[track_caller]
    pub fn read_string_buf(&mut self, len: usize) -> Result<Spectrum<str>> {

        // Every string is at least its length and its terminating nul, which bounds how
        // many of them there can be before any room is made for them.
        if len > self.remaining() / 5 {
            return Err(invalid_data_err(format!("Failed to read {len} strings, only {} octets remain", self.remaining())));
        }

        let codeset = self.codeset();
        let decode_failed = || invalid_data_err(format!("Failed to read a string, decoding it with {codeset:?} failed"));

        let start = self.pos;
        let window = &self.buf[start..];
        let mut buf = None::<BufMut<u8, 1>>;
        let mut spans = Vec::with_capacity(len);

        for _ in 0..len {

            let raw = self.read_raw_string()?;

            let buf = match &mut buf {
                Some(buf) => buf,
                buf @ None => match codeset.decode_is_native(raw) {
                    // Shared in place, at its span in the window.
                    Some(true) => {
                        let at = raw.as_ptr().addr() - window.as_ptr().addr();
                        spans.push(span(at, at + raw.len())?);
                        continue;
                    }
                    // The strings shared so far are copied out of the window, which they
                    // are all in, to the buffer that the rest are decoded into. Without
                    // the octets between them, they end before they did, in 32 bits.
                    Some(false) => {
                        let buf = buf.insert(BufMut::with_capacity(self.pos - start));
                        for span in &mut spans {
                            let at = buf.len();
                            buf.extend_from_slice(&window[span.start as usize..span.end as usize]);
                            *span = at as u32..buf.len() as u32;
                        }
                        buf
                    }
                    None => return Err(decode_failed()),
                }
            };

            let at = buf.len();
            codeset.decode_to_buf(raw, buf).ok_or_else(decode_failed)?;
            spans.push(span(at, buf.len())?);

        }

        let buf = match buf {
            Some(buf) => buf.freeze(),
            None => self.share(&self.buf[start..self.pos]),
        };

        // SAFETY: There is one span per string, each one being the octets of a string
        // within the buffer, which the codeset says are valid UTF-8: either native ones,
        // shared in place or copied out of the window, or the ones it appended.
        Ok(unsafe { Buf::from_spans_unchecked(buf, spans, [len]) })

    }

    /// Read the byte order flag that starts an encapsulation, and apply it to this 
    /// reader, so that the numbers it reads afterward are swapped when it is not the 
    /// native one.
    #[track_caller]
    pub fn read_endian(&mut self) -> Result<Endian> {
        // The byte order flag is TRUE (1) for little endian and FALSE (0) for big endian.
        let endian = if self.read_bool()? { Endian::Little } else { Endian::Big };
        self.set_endian(endian);
        Ok(endian)
    }

    /// Read an encapsulation, which is a sequence of octets that the CDR alignment restarts
    /// at the origin of, and return the reader of its own that it is read through.
    ///
    /// Nothing is copied because the alignment is tracked from the offset of the cursor and
    /// not from the address of the octets, see the note on this type: the returned reader
    /// simply borrows them in place, and it is left right past the byte order flag, which
    /// it has already applied.
    ///
    /// The codesets are **not** inherited, because the specification makes them a property
    /// of the encapsulation itself and not of the stream holding it: every usage of an
    /// encapsulation has to define the GIOP version and the codesets it is encoded with,
    /// the default being GIOP 1.0, so ISO 8859-1 for the strings and no wide strings at
    /// all, which is what the returned reader is left on. A usage that defines otherwise
    /// sets them on it, see [`Self::set_codesets`].
    #[track_caller]
    pub fn read_encapsulation(&mut self) -> Result<CdrReader<'a>> {

        let len = self.read_sequence_len()?;
        if len == 0 {
            return Err(invalid_data_err("Failed to read an encapsulation, it does not even hold a byte order flag"));
        }

        let raw = self.read_raw_slice(len, 1)?;

        let mut reader = self.substream(raw);
        reader.read_endian()?;

        Ok(reader)

    }

}

impl Debug for CdrReader<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CdrReader")
            .field("position", &self.pos)
            .field("remaining", &self.owner.slice_ref(self.remaining_slice()))
            .field("endian", &self.endian)
            .field("codesets", &self.codesets)
            .finish()
    }
}

// ==================== //
//         Write        //
// ==================== //

/// A structure for writing a CDR buffer, which grows as it goes, so that a write never
/// fails for the lack of room, and which is handed out as the [`BufMut`] it filled.
///
/// # Alignment
///
/// CDR aligns every primitive on its own size, relative to the origin of the stream, and
/// this writer tracks that alignment *logically*, from the offset of its position and not
/// from the address of the octets, just like [`CdrReader`] does. An encapsulation restarts
/// it at its own origin, and is therefore written in place, into the very same buffer, see
/// [`Self::write_encapsulation`].
///
/// The alignment padding it inserts is always zeroed, and every other octet of the stream
/// is one a primitive wrote, so a buffer that is recycled never leaks the octets of the
/// stream it held before, which are still there past its length, see [`BufMut::set_len`].
///
/// # Endianness
///
/// The octets are always written in the **native** order, which the byte order flag of the
/// message or of the encapsulation advertises, see [`Self::write_endian`]. Nothing is ever
/// swapped on its way out, a peer that reads in the other order being the one that swaps.
///
/// # Rewinding
///
/// The position can be moved back to an offset that was noted earlier, see
/// [`Self::set_position`], which is how a length that is only known afterward is filled in
/// without encoding into a buffer of its own first. Writing carries on from there, over
/// the octets that are already there, and the buffer keeps whichever of the two is longer.
pub struct CdrWriter {
    /// The buffer being filled, of which the first `pos` octets are the stream, and the
    /// rest the room a primitive booked and did not need, or a rewind left behind.
    buf: BufMut<u8, 1>,
    /// Offset in the buffer the next primitive is written at, which the CDR alignment is
    /// derived from, and where the stream stops.
    pos: usize,
    /// Offset in the buffer the CDR alignment is relative to, which is 0 for a message and
    /// the first octet of an encapsulation while one is being written.
    origin: usize,
    /// Codesets to use for the (wide) strings, when negotiated.
    codesets: Option<Codesets>,
}

impl CdrWriter {

    #[inline]
    pub fn new() -> Self {
        Self::from_buf(BufMut::new())
    }

    /// A writer with room for `len` octets already made, for a caller that knows roughly
    /// how long what it is about to write is.
    ///
    /// Everything of a message is written a field at a time, so a writer that starts with
    /// nothing grows two or three times before one of any ordinary size is down. The room
    /// is a floor and not a limit: a writer grows past it like any buffer, see
    /// [`BufMut::reserve`].
    #[inline]
    pub fn with_capacity(len: usize) -> Self {
        Self::from_buf(BufMut::with_capacity(len))
    }

    /// Carry on writing into a buffer that already holds octets, the CDR alignment staying
    /// relative to its very first one, so that a stream can be written in several goes,
    /// and a buffer recycled from a stream to the next, see [`Self::into_buf`].
    #[inline]
    pub fn from_buf(buf: BufMut<u8, 1>) -> Self {
        Self {
            pos: buf.len(),
            buf,
            origin: 0,
            codesets: None,
        }
    }

    #[inline]
    pub const fn codesets(&self) -> Option<Codesets> {
        self.codesets
    }

    #[inline]
    pub const fn set_codesets(&mut self, codesets: Option<Codesets>) {
        self.codesets = codesets;
    }

    #[inline]
    pub const fn with_codesets(mut self, codesets: Option<Codesets>) -> Self {
        self.codesets = codesets;
        self
    }

    /// The codeset of the narrow strings, defaulting to the one mandated by the
    /// specification when no codeset has been negotiated.
    #[inline]
    pub fn codeset(&self) -> Codeset {
        self.codesets.map(|codesets| codesets.codeset).unwrap_or(Codeset::Iso8859_1)
    }

    /// The codeset of the wide strings, there is no default one for those.
    #[inline]
    pub fn wide_codeset(&self) -> Option<WideCodeset> {
        self.codesets.map(|codesets| codesets.wide_codeset)
    }

    /// The octets that are written so far, which stop at the position: the ones past it are
    /// the room a primitive booked and did not need, and are left to the ones that follow.
    #[inline]
    pub fn buffer(&self) -> &[u8] {
        &self.buf.as_slice()[self.origin..self.pos]
    }

    /// Give the buffer that was filled back, from the origin to the position, see
    /// [`Self::from_buf`] for the way back, which is how a buffer is recycled from a stream
    /// to the next.
    ///
    /// The position is where the stream stops, so one that was moved back has to be moved
    /// to the end of the stream again before the buffer is taken, see
    /// [`Self::set_position`]. The octets before the origin are dropped, which moves the
    /// ones behind them to the front of the buffer.
    #[inline]
    pub fn into_buf(mut self) -> BufMut<u8, 1> {
        self.buf.truncate(self.pos);
        if self.origin != 0 {
            self.buf.as_mut_slice().copy_within(self.origin.., 0);
            self.buf.truncate(self.pos - self.origin);
        }
        self.buf
    }

    #[inline]
    pub fn into_sender(self) -> CdrSender {
        CdrSender::from_writer(self)
    }

    /// Drop everything that was written, the allocation being kept around so that the
    /// buffer can be recycled.
    #[inline]
    pub fn clear(&mut self) {
        self.buf.clear();
        self.pos = 0;
        self.origin = 0;
    }

    /// Offset the CDR alignment is counted from, which is where the stream being written
    /// starts: zero for one written from the very first octet of the buffer, and the first
    /// octet of the stream being written into it otherwise, see
    /// [`Self::write_encapsulation`] and [`Self::set_origin`].
    #[inline]
    pub const fn origin(&self) -> usize {
        self.origin
    }

    /// Move the origin the CDR alignment is counted from, which is what a stream written
    /// into the octets of another one restarts its alignment at: a GIOP message laid down
    /// behind the ones that are waiting to go out is aligned, and measured, from where it
    /// starts and not from where they do, see [`super::giop::MessageLen`].
    ///
    /// # Panics
    ///
    /// Panics if the offset is past the position, which the origin of the stream being
    /// written is never past.
    #[inline]
    pub fn set_origin(&mut self, origin: usize) {
        assert!(origin <= self.pos, "cannot move the CDR origin past the position of {}", self.pos);
        self.origin = origin;
    }

    /// Offset the next primitive is written at, which the CDR alignment is derived from.
    #[inline]
    pub const fn position(&self) -> usize {
        self.pos
    }

    /// Move the position to an offset that was noted earlier, so that a primitive whose
    /// value is only known afterward is written where it was kept for, see the note on
    /// this type. The octets past it are kept, and the stream stops at the position, so it
    /// has to be moved back to the end of the stream to carry it on.
    ///
    /// # Panics
    ///
    /// Panics if the offset is past the octets the buffer holds, or before the origin of
    /// the stream being written.
    #[inline]
    pub fn set_position(&mut self, pos: usize) {
        assert!(pos <= self.buf.len(), "cannot move the CDR position past the {} octets of the buffer", self.buf.len());
        assert!(pos >= self.origin, "cannot move the CDR position before the origin of the stream being written");
        self.pos = pos;
    }

    /// Pad the stream up to the given alignment, which is where the next primitive of that
    /// width would be written, whether or not one actually follows.
    ///
    /// The CDR aligns every primitive it writes on its own, so this is only for the
    /// alignment a protocol asks for on octets that are not one: the body of a GIOP 1.2
    /// message is padded up to eight whatever it turns out to hold, and only the protocol
    /// knows that the body starts there.
    ///
    /// The alignment is counted from the origin of the stream being written, which is the
    /// first octet of an encapsulation when one is being written into, and the padding is
    /// zeroed like the padding of any primitive.
    ///
    /// # Panics
    ///
    /// Panics if the alignment is not a power of two.
    #[track_caller]
    pub fn align(&mut self, align: usize) -> Result<()> {

        assert!(align.is_power_of_two(), "cannot align a CDR stream on {align}, which is not a power of two");

        let len = align_offset(self.pos - self.origin, align);
        if len == 0 {
            return Ok(());
        }

        // Aligned on one, so nothing of its own is skipped: these octets *are* the padding,
        // and they are zeroed here because a rewind may have left something in them.
        self.next_slice(1, len)?.fill(0);

        Ok(())

    }

    /// Make room for `len` octets aligned on `align`, and return the offset of the first
    /// one, the position being left right past them.
    ///
    /// The alignment padding it skips is zeroed, because it is never handed out to the
    /// caller and would otherwise be whatever the buffer holds there: the octets a recycled
    /// buffer or a rewind left behind.
    ///
    /// The only way this fails is the buffer failing to grow, which leaves it and the
    /// position exactly as they were.
    fn reserve_slice(&mut self, align: usize, len: usize) -> Result<usize> {

        debug_assert!(align.is_power_of_two());

        // No primitive is actually crossed by an empty chunk, so no padding is inserted
        // either, and the position keeps its alignment for the next one.
        if len == 0 {
            return Ok(self.pos);
        }

        let start = self.pos + align_offset(self.pos - self.origin, align);
        let end = start + len;

        // A stream is a GIOP message or an encapsulation, and both are measured by an ulong,
        // so neither can ever be longer than one. Checking it here is what makes the length
        // of an encapsulation impossible to overflow, see [`Self::write_encapsulation`].
        if end - self.origin > u32::MAX as usize {
            return Err(invalid_data_err(format!("Failed to write the CDR stream, its {} octets do not fit into an ulong length", end - self.origin)));
        }

        if end > self.buf.len() {
            self.buf.try_reserve(end - self.buf.len())?;
            self.buf.set_len(end);
        }

        // The padding is never handed out to the caller, so it is zeroed here, in case a
        // rewind or an earlier stream left something there.
        self.buf.as_mut_slice()[self.pos..start].fill(0);
        self.pos = end;

        Ok(start)

    }

    /// Same as [`Self::reserve_slice`], but the octets are handed out to be written.
    #[inline]
    fn next_slice(&mut self, align: usize, len: usize) -> Result<&mut [u8]> {
        let start = self.reserve_slice(align, len)?;
        Ok(&mut self.buf.as_mut_slice()[start..][..len])
    }

    /// Write a number at an offset that [`Self::reserve_slice`] made room for, without
    /// moving the position, which is how a length that is only known afterward is filled in.
    fn write_num_at<T: CdrNum>(&mut self, offset: usize, value: T) {

        let size_and_align = mem::size_of::<T>();
        let dst = &mut self.buf.as_mut_slice()[offset..][..size_and_align];

        // SAFETY: A CDR number holds no padding and every one of its bit patterns is valid,
        // so its octets can be copied out of it, and the destination holds exactly as many
        // of them. Both reads and writes are unaligned because neither the value nor the
        // destination is necessarily aligned for the type.
        unsafe {
            dst.as_mut_ptr().cast::<T>().write_unaligned(value);
        }

    }

    fn write_num<T: CdrNum>(&mut self, value: T) -> Result<()> {
        let size_and_align = mem::size_of::<T>();
        let offset = self.reserve_slice(size_and_align, size_and_align)?;
        self.write_num_at(offset, value);
        Ok(())
    }

    fn write_num_slice<T: CdrNum>(&mut self, values: &[T]) -> Result<()> {

        let size_and_align = mem::size_of::<T>();
        let len = values.len();
        let byte_len = size_and_align.checked_mul(len)
            .ok_or_else(|| invalid_data_err(format!("Failed to write {len} elements of {size_and_align} octets, their size overflows")))?;

        let dst = self.next_slice(size_and_align, byte_len)?;

        // SAFETY: See [`Self::write_num_at`], and the destination holds exactly as many
        // octets as the values do, which cannot overlap it because they are borrowed.
        unsafe {
            dst.as_mut_ptr().copy_from_nonoverlapping(values.as_ptr().cast::<u8>(), byte_len);
        }

        Ok(())

    }

    #[inline]
    pub fn write_octet_slice(&mut self, values: &[u8]) -> Result<()> {
        self.next_slice(1, values.len())?.copy_from_slice(values);
        Ok(())
    }

    #[inline]
    pub fn write_octet(&mut self, value: u8) -> Result<()> {
        self.write_num(value)
    }

    /// Write a slice of booleans, which are the very octets they are made of, 0 for false
    /// and 1 for true.
    #[inline]
    pub fn write_bool_slice(&mut self, values: &[bool]) -> Result<()> {
        // SAFETY: A bool has the same size and alignment as an u8, and only the 0 and 1 bit
        // patterns, which are the ones the CDR uses for it.
        self.write_octet_slice(unsafe { slice::from_raw_parts(values.as_ptr().cast::<u8>(), values.len()) })
    }

    #[inline]
    pub fn write_bool(&mut self, value: bool) -> Result<()> {
        self.write_octet(value as u8)
    }

    #[inline]
    pub fn write_ushort_slice(&mut self, values: &[u16]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_ushort(&mut self, value: u16) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_short_slice(&mut self, values: &[i16]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_short(&mut self, value: i16) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_ulong_slice(&mut self, values: &[u32]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_ulong(&mut self, value: u32) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_long_slice(&mut self, values: &[i32]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_long(&mut self, value: i32) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_ulonglong_slice(&mut self, values: &[u64]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_ulonglong(&mut self, value: u64) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_longlong_slice(&mut self, values: &[i64]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_longlong(&mut self, value: i64) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_float_slice(&mut self, values: &[f32]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_float(&mut self, value: f32) -> Result<()> {
        self.write_num(value)
    }

    #[inline]
    pub fn write_double_slice(&mut self, values: &[f64]) -> Result<()> {
        self.write_num_slice(values)
    }

    #[inline]
    pub fn write_double(&mut self, value: f64) -> Result<()> {
        self.write_num(value)
    }

    /// Write the values of that data, in column-major order and without the length that
    /// precedes a sequence, the counterpart of [`CdrReader::read_buf`].
    pub fn write_data<T: CdrElement, const DIM: usize>(&mut self, data: &Buf<T, DIM>) -> Result<()> {
        let bytes = data.as_bytes();
        self.next_slice(mem::size_of::<T>(), bytes.len())?.copy_from_slice(bytes);
        Ok(())
    }

    /// Write the strings of that data, in column-major order and without the length that
    /// precedes a sequence, the counterpart of [`CdrReader::read_string_buf`].
    pub fn write_string_data<const DIM: usize>(&mut self, data: &Buf<str, DIM>) -> Result<()> {
        data.iter().try_for_each(|s| self.write_string(s))
    }

    /// Write the length of a sequence, which is an ulong on the wire.
    pub fn write_sequence_len(&mut self, len: usize) -> Result<()> {
        let len = u32::try_from(len)
            .map_err(|_| invalid_data_err(format!("Failed to write a sequence length, {len} does not fit into an ulong")))?;
        self.write_ulong(len)
    }

    /// Write octets as the string they are, which is them and the terminating nul their
    /// length counts in, with no codeset applied to them at all, see
    /// [`CdrReader::read_raw_string`].
    #[track_caller]
    pub fn write_raw_string(&mut self, raw: &[u8]) -> Result<()> {

        // A nul is what ends a string, so one inside these octets would cut it short and
        // make what goes out say something else than what was handed over.
        if raw.contains(&0) {
            return Err(invalid_data_err("Failed to write a raw string, it holds a nul octet"));
        }

        self.write_sequence_len(raw.len() + 1)?;
        self.write_octet_slice(raw)?;
        self.write_octet(0)

    }

    /// Write a string, encoded with this writer's codeset and terminated by a nul octet,
    /// which its length counts in.
    ///
    /// The length is only known once the codeset has run, so it is filled in afterward. The
    /// codeset appends the octets to the buffer when the position is at its end; otherwise
    /// the position was moved back into a longer stream, whose octets past it are kept,
    /// and the octets are encoded aside to be copied over them.
    #[track_caller]
    pub fn write_string(&mut self, s: &str) -> Result<()> {

        let codeset = self.codeset();
        let encode_failed = || invalid_data_err(format!("Failed to write a string, encoding it with {codeset:?} failed"));

        // The octets directly follow the length as they are not aligned themselves.
        let len_at = self.reserve_slice(4, 4)?;
        let start = self.pos;

        if start == self.buf.len() {
            let len = codeset.encode_to_buf(s, &mut self.buf).ok_or_else(encode_failed)?;
            self.pos += len;
            // The stream is measured by an ulong, see [`Self::reserve_slice`], which the
            // octets are only checked against once they are there.
            if self.pos - self.origin > u32::MAX as usize {
                self.buf.truncate(start);
                self.pos = start;
                return Err(invalid_data_err("Failed to write a string, it is too long for an ulong length"));
            }
        } else {
            let mut buf = BufMut::new();
            codeset.encode_to_buf(s, &mut buf).ok_or_else(encode_failed)?;
            self.write_octet_slice(buf.as_slice())?;
        }

        self.write_octet(0)?;

        // The stream fits an ulong, and so does a string within it.
        self.write_num_at(len_at, (self.pos - start) as u32);

        Ok(())

    }

    /// Write a wide string, encoded with this writer's wide codeset.
    ///
    /// Its length is in octets and not in code units, and it is **not** nul terminated,
    /// which is how GIOP 1.2 and the encapsulations encode it, see
    /// [`CdrReader::read_wstring`]. The code units are encoded aside first, so that their
    /// length is known when it is written before them.
    #[track_caller]
    pub fn write_wstring(&mut self, s: &Utf16Str) -> Result<()> {

        let codeset = self.wide_codeset()
            .ok_or_else(|| invalid_data_err("Failed to write a wide string, no wide codeset was negotiated"))?;

        let mut units = BufMut::new();
        let len = codeset.encode_to_buf(s, &mut units)
            .ok_or_else(|| invalid_data_err(format!("Failed to write a wide string, encoding it with {codeset:?} failed")))?;

        let len = len.checked_mul(2)
            .and_then(|len| u32::try_from(len).ok())
            .ok_or_else(|| invalid_data_err("Failed to write a wide string, it is too long for an ulong length"))?;

        self.write_ulong(len)?;
        self.write_ushort_slice(units.as_slice())

    }

    /// Write the byte order flag that starts an encapsulation, or a GIOP message, which is
    /// always the native one because that is the order everything is written in, see the
    /// note on this type.
    #[inline]
    pub fn write_endian(&mut self) -> Result<()> {
        // The byte order flag is TRUE (1) for little endian and FALSE (0) for big endian.
        self.write_bool(Endian::NATIVE.is_little())
    }

    /// Write an encapsulation, which is a sequence of octets that the CDR alignment
    /// restarts at the origin of, the given function writing what it holds.
    ///
    /// Nothing is copied because the alignment is tracked from the offset of the position
    /// and not from the address of the octets, see the note on this type: the encapsulation
    /// is written in place, into the very same buffer. Its byte order flag is written
    /// before the function is called, and its length once it returns.
    ///
    /// The codesets are **not** inherited, because the specification makes them a property
    /// of the encapsulation itself and not of the stream holding it, see
    /// [`CdrReader::read_encapsulation`]: the function is given a writer on the ISO 8859-1
    /// of GIOP 1.0, and sets them itself when the encapsulation it writes defines
    /// otherwise.
    ///
    /// An encapsulation that fails to be written is dropped from the buffer, so the stream
    /// carries on exactly where it was.
    ///
    /// ```ignore
    /// let mut writer = CdrWriter::new();
    /// writer.write_octet(0x2A)?;
    /// writer.write_encapsulation(|encapsulation| {
    ///     encapsulation.write_ulong(0xDEADBEEF)
    /// })?;
    /// writer.write_octet(0x2B)?;
    /// ```
    pub fn write_encapsulation(&mut self, write: impl FnOnce(&mut Self) -> Result<()>) -> Result<()> {

        let restore_pos = self.pos;
        let restore_len = self.buf.len();

        // Room for the length, which is only known once the encapsulation is written, and
        // is filled in below.
        let len_at = self.reserve_slice(4, 4)?;
        let origin = self.pos;

        // The encapsulation is a stream of its own: its alignment restarts right here, and
        // its codesets are the default ones until it says otherwise.
        let outer_origin = mem::replace(&mut self.origin, origin);
        let outer_codesets = self.codesets.take();

        // The byte order flag is the very first octet of an encapsulation, and it is only
        // followed by what the function writes if it could be written at all. Note that the
        // room for it is already there, since the length was aligned on 4 octets.
        let res = self.write_endian().and_then(|()| write(self));

        self.origin = outer_origin;
        self.codesets = outer_codesets;

        if let Err(err) = res {
            // The stream carries on where it was, and the octets the encapsulation grew the
            // buffer by are dropped, the ones it wrote over being gone for good.
            self.pos = restore_pos;
            self.buf.truncate(restore_len);
            return Err(err);
        }

        // The length of a stream is capped to an ulong when its octets are reserved, so
        // this cannot overflow, see [`Self::reserve_slice`].
        let len = self.pos - origin;
        debug_assert!(u32::try_from(len).is_ok(), "the CDR stream grew past an ulong length");
        self.write_num_at(len_at, len as u32);

        Ok(())

    }

}

impl Default for CdrWriter {
    #[inline]
    fn default() -> Self {
        Self::new()
    }
}

impl Debug for CdrWriter {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CdrWriter")
            .field("position", &self.pos)
            .field("origin", &self.origin)
            .field("buffer", &self.buf)
            .field("codesets", &self.codesets)
            .finish()
    }
}

// ==================== //
//   Sender (for IO)    //
// ==================== //

/// A writer-like structure specialized for sending through I/O.
#[derive(Debug, Default)]
pub struct CdrSender {
    /// The main buffer where buffers might be interleaved.
    writer: CdrWriter,
    /// The full buffers that are interleaved into the buffer to be flushed.
    interleaved: Vec<Interleaved>,
}

#[derive(Debug)]
struct Interleaved {
    /// The position this is interleaved at.
    at: usize,
    /// The position already read of this buffer.
    pos: usize,
    /// The buffer to read from.
    buf: Buf<u8, 1>,
}

impl CdrSender {

    #[inline]
    pub const fn from_writer(writer: CdrWriter) -> Self {
        Self {
            writer,
            interleaved: Vec::new(),
        }
    }

    pub fn interleave(&mut self, buf: Buf<u8, 1>) {
        
        if buf.is_empty() {
            return;
        }

        self.interleaved.push(Interleaved {
            at: self.writer.position(),
            pos: 0,
            buf,
        });

    }

    /// The octets that go out next, which are the run at the front of the stream: the ones
    /// of the writer up to the point the next buffer was laid in at, or that buffer itself
    /// once the stream stands right on it, see [`Self::interleave`].
    ///
    /// This gives back exactly what [`Self::ack`] takes, so a caller hands these to a
    /// writer and says how much of them went out, and never reaches past what it was given.
    pub fn buffer(&self) -> &[u8] {

        let origin = self.writer.origin();

        let Some(interleaved) = self.interleaved.first() else {
            return self.writer.buffer();
        };

        // The buffer laid in right where the stream stands is the one that goes out now,
        // whatever the writer holds behind it waiting its turn.
        if interleaved.at == origin {
            return &interleaved.buf.as_slice()[interleaved.pos..];
        }

        // Otherwise what goes out is what the writer holds in front of it, and not one
        // octet past the point it was laid in at: those come after it and not before.
        &self.writer.buffer()[..interleaved.at - origin]

    }

    /// Take note that the given number of octets of [`Self::buffer`] have gone out, which
    /// is what moves the stream forward: what that call gives back afterward is whatever
    /// follows them.
    ///
    /// One call may span several runs, a writer that was handed more than the run at the
    /// front being told so once: each turn takes what the run at the front still holds and
    /// moves on to the one behind it.
    pub fn ack(&mut self, len: usize) {

        let mut remaining = len;

        while remaining != 0 {

            let begin = self.writer.origin();
            let mut end = self.writer.position();

            if let Some(interleaved) = self.interleaved.first_mut() {

                end = interleaved.at;

                if interleaved.at == begin {

                    let taken = usize::min(remaining, interleaved.buf.len() - interleaved.pos);
                    interleaved.pos += taken;
                    remaining -= taken;

                    if interleaved.pos == interleaved.buf.len() {
                        self.interleaved.remove(0);
                    }

                    continue;

                }

            }

            let size = end.strict_sub(begin);
            let taken = usize::min(remaining, size);
            assert!(taken != 0, "acknowledged more octets than were given out");

            self.writer.set_origin(begin + taken);
            remaining -= taken;

        }

    }

    pub fn drop_off_origin(&mut self) {

        let origin = self.writer.origin();
        let codesets = self.writer.codesets();
        self.writer = CdrWriter::from_buf(mem::take(&mut self.writer).into_buf())
            .with_codesets(codesets);

        for interleaved in &mut self.interleaved {
            interleaved.at -= origin;
        }
        
    }

}

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

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

// ==================== //
//      CDR typing      //
// ==================== //

/// Token sealing the sequence hooks of [`CdrRead`] and [`CdrWrite`]: it can only be built by
/// this module, so those hooks can only be called from here, by the sequence
/// implementations they exist for.
#[derive(Debug)]
pub struct CdrSealed(());

/// Read a value out of a [`CdrReader`], the `'d` lifetime being the lifetime of the octets
/// the reader reads, so that an implementor can borrow them in place instead of copying
/// them out, and an aggregate can borrow many disjoint parts of them at once.
pub trait CdrRead<'d>: Sized {

    fn read(buf: &mut CdrReader<'d>) -> Result<Self>;

    /// Read a whole run of `len` values at once, which only the CDR builtin types can do:
    /// every other type returns [`None`], leaving the reader untouched, and is then read one
    /// value at a time by the sequences.
    ///
    /// The run is borrowed straight out of the octets when they are aligned for the type and
    /// in the native order, and copied out of them when they are not, hence the
    /// clone-on-write, see [`CdrReader`].
    ///
    /// This is a hand-rolled specialization, statically dispatched on the element type, so
    /// neither the borrowed nor the value by value path costs anything at runtime.
    ///
    /// It is sealed by [`CdrSealed`], so only the sequence implementations of this module
    /// ever call it, which is why an implementor of this trait has nothing to do with the
    /// length prefix: the sequences deal with it themselves.
    #[doc(hidden)]
    fn _read_slice(reader: &mut CdrReader<'d>, len: usize, _sealed: CdrSealed) -> Option<Result<Cow<'d, [Self]>>>
    where
        Self: Clone,
    {
        let _ = (reader, len);
        None
    }

    /// Read a whole run of `len` values at once into a vector, the owned counterpart of
    /// [`Self::_read_slice`]: nothing being borrowed, it needs no `Clone` bound, so a
    /// sequence read into a [`Vec`] takes the same path whatever its element type is.
    #[doc(hidden)]
    fn _read_vec(reader: &mut CdrReader<'d>, len: usize, _sealed: CdrSealed) -> Option<Result<Vec<Self>>> {
        let _ = (reader, len);
        None
    }

}

/// Write a value into a [`CdrWriter`].
pub trait CdrWrite {

    fn write(&self, buf: &mut CdrWriter) -> Result<()>;

    /// Copy a whole run of values into the stream at once, which only the CDR builtin types
    /// can do, see [`CdrRead::_read_slice`].
    ///
    /// The `Self: Sized` bound keeps this trait dyn compatible, which it must stay because
    /// a message is encoded through a `&dyn CdrWrite`.
    #[doc(hidden)]
    fn _write_slice(values: &[Self], writer: &mut CdrWriter, _sealed: CdrSealed) -> Option<Result<()>>
    where
        Self: Sized,
    {
        let _ = (values, writer);
        None
    }

}

impl<'d> CdrReader<'d> {

    /// Read any value implementing [`CdrRead`], which may borrow the octets this reader
    /// reads, and many such values can be alive at the same time.
    #[inline]
    #[track_caller]
    pub fn read<T: CdrRead<'d>>(&mut self) -> Result<T> {
        T::read(self)
    }

    /// Read `len` values of any type implementing [`CdrRead`], without the length that
    /// precedes a sequence: this is a CDR array, and it is also what a sequence read in
    /// several goes is made of.
    ///
    /// A run of a CDR builtin type is taken out of the stream at once and borrowed in
    /// place when the octets allow it, hence the clone-on-write, see [`CdrReader`]. Any
    /// other type is read value by value into a sequence of its own.
    #[track_caller]
    pub fn read_slice<T: CdrRead<'d> + Clone + 'd>(&mut self, len: usize) -> Result<Cow<'d, [T]>> {

        if let Some(slice) = T::_read_slice(self, len, CdrSealed(())) {
            return slice;
        }

        // Falling back to read_vec just in case it is supported, plus it allows sharing
        // the code handling the reservation and reading of data.
        self.read_vec(len).map(Cow::Owned)

    }

    /// Read `len` values of any type implementing [`CdrRead`] into a vector, without the
    /// length that precedes a sequence, see [`Self::read_slice`]. The values are owned and
    /// never borrowed, so this needs no `Clone` bound, and a run of a CDR builtin type is
    /// still taken out of the stream at once.
    #[track_caller]
    pub fn read_vec<T: CdrRead<'d>>(&mut self, len: usize) -> Result<Vec<T>> {

        if let Some(vec) = T::_read_vec(self, len, CdrSealed(())) {
            return vec;
        }

        // The length is whatever the stream says, and nothing bounds it before the values
        // are actually read, so the room for them is reserved the fallible way.
        let mut ret = Vec::new();
        ret.try_reserve(len)
            .map_err(|err| out_of_memory_err(err, format!("Failed to allocate room for a sequence of {len} values")))?;

        for _ in 0..len {
            ret.push(T::read(self)?);
        }

        Ok(ret)

    }

}

impl CdrWriter {

    /// Write any value implementing [`CdrWrite`].
    #[inline]
    #[track_caller]
    pub fn write<T: CdrWrite + ?Sized>(&mut self, value: &T) -> Result<()> {
        value.write(self)
    }

    /// Write values of any type implementing [`CdrWrite`], without the length that
    /// precedes a sequence, the counterpart of [`CdrReader::read_slice`].
    ///
    /// This is the generic form of the `write_*_slice` methods and takes the very same
    /// bulk path they do when the element type is a CDR builtin one, any other type being
    /// written value by value.
    #[track_caller]
    pub fn write_slice<T: CdrWrite>(&mut self, values: &[T]) -> Result<()> {

        if let Some(res) = T::_write_slice(values, self, CdrSealed(())) {
            return res;
        }

        for value in values {
            self.write(value)?;
        }

        Ok(())

    }

}

// ==================== //
//       Numbers        //
// ==================== //

/// Implement the CDR types of a numeric primitive: the number itself, and the runs of them,
/// which the sequences borrow out of the stream or copy into it at once.
macro_rules! impl_cdr_type_num {
    ( $name:ident, $ty:ty ) => {
        pastey::paste! {

            impl<'d> CdrRead<'d> for $ty {

                #[inline]
                fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
                    reader.[<read_ $name>]()
                }

                #[inline]
                fn _read_slice(reader: &mut CdrReader<'d>, len: usize, _sealed: CdrSealed) -> Option<Result<Cow<'d, [Self]>>> {
                    Some(reader.[<read_ $name _slice>](len))
                }

                #[inline]
                fn _read_vec(reader: &mut CdrReader<'d>, len: usize, _sealed: CdrSealed) -> Option<Result<Vec<Self>>> {
                    Some(reader.[<read_ $name _slice>](len).map(Cow::into_owned))
                }

            }

            impl CdrWrite for $ty {

                #[inline]
                fn write(&self, writer: &mut CdrWriter) -> Result<()> {
                    writer.[<write_ $name>](*self)
                }

                #[inline]
                fn _write_slice(values: &[Self], writer: &mut CdrWriter, _sealed: CdrSealed) -> Option<Result<()>> {
                    Some(writer.[<write_ $name _slice>](values))
                }

            }

        }
    };
}

impl_cdr_type_num!(bool, bool);
impl_cdr_type_num!(octet, u8);
impl_cdr_type_num!(ushort, u16);
impl_cdr_type_num!(short, i16);
impl_cdr_type_num!(ulong, u32);
impl_cdr_type_num!(long, i32);
impl_cdr_type_num!(ulonglong, u64);
impl_cdr_type_num!(longlong, i64);
impl_cdr_type_num!(float, f32);
impl_cdr_type_num!(double, f64);

// ==================== //
//       Strings        //
// ==================== //

impl<'d> CdrRead<'d> for Cow<'d, str> {
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        reader.read_string()
    }
}

impl<'d> CdrRead<'d> for String {
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        reader.read_string().map(Cow::into_owned)
    }
}

impl CdrWrite for str {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_string(self)
    }
}

impl CdrWrite for String {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_string(self)
    }
}

impl<'d> CdrRead<'d> for Cow<'d, Utf16Str> {
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        reader.read_wstring()
    }
}

impl<'d> CdrRead<'d> for Utf16String {
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        reader.read_wstring().map(Cow::into_owned)
    }
}

impl CdrWrite for Utf16Str {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_wstring(self)
    }
}

impl CdrWrite for Utf16String {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_wstring(self)
    }
}

// ==================== //
//       Codesets       //
// ==================== //

impl CdrRead<'_> for Codesets {
    fn read(reader: &mut CdrReader<'_>) -> Result<Self> {
        let codeset_raw = reader.read_ulong()?;
        let wide_codeset_raw = reader.read_ulong()?;
        Ok(Self {
            codeset: Codeset::from_id(codeset_raw)
                .ok_or_else(|| invalid_data_err(format!("Codeset 0x{codeset_raw:08X} is not supported")))?,
            wide_codeset: WideCodeset::from_id(wide_codeset_raw)
                .ok_or_else(|| invalid_data_err(format!("Wide codeset 0x{wide_codeset_raw:08X} is not supported")))?,
        })
    }
}

impl CdrWrite for Codesets {
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_ulong(self.codeset.id())?;
        writer.write_ulong(self.wide_codeset.id())?;
        Ok(())
    }
}

// ==================== //
//      Containers      //
// ==================== //

impl<'d, T: CdrRead<'d> + Clone + 'd> CdrRead<'d> for Cow<'d, [T]> {
    /// A sequence of a CDR builtin type is read out of the stream at once, borrowed in
    /// place when the octets allow it, and one of a constructed type is read value by value
    /// into a sequence of its own.
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        let len = reader.read_sequence_len()?;
        reader.read_slice(len)
    }
}

impl<'d, T: CdrRead<'d>> CdrRead<'d> for Vec<T> {
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        let len = reader.read_sequence_len()?;
        reader.read_vec(len)
    }
}

impl<'d, T> CdrRead<'d> for Box<[T]>
where
    Vec<T>: CdrRead<'d>,
{
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        <Vec<T>>::read(reader).map(Vec::into_boxed_slice)
    }
}

/// A sequence of octets is the only one that is always borrowed straight out of the stream,
/// because a single byte needs neither alignment nor byte swapping, whatever the byte order
/// of the stream is.
///
/// A sequence of any other CDR builtin type is only borrowed when the octets happen to be
/// aligned for it and in the native order, so it is read as a [`Cow`] and it is up to the
/// caller to deal with the copy it may be, see [`CdrReader`].
impl<'d> CdrRead<'d> for &'d [u8] {
    #[inline]
    fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
        let len = reader.read_sequence_len()?;
        reader.read_raw_slice(len, 1)
    }
}

impl<T: CdrWrite> CdrWrite for [T] {
    /// A sequence copied into the stream at once when its element type is a CDR builtin
    /// one, and written value by value otherwise.
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_sequence_len(self.len())?;
        writer.write_slice(self)
    }
}

impl<T: CdrWrite> CdrWrite for Vec<T> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        <[T] as CdrWrite>::write(self, writer)
    }
}

impl<T, const N: usize> CdrWrite for [T; N]
where
    [T]: CdrWrite,
{
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        <[T] as CdrWrite>::write(&self[..], writer)
    }
}

impl<T: CdrWrite + ?Sized> CdrWrite for &'_ T {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        CdrWrite::write(&**self, writer)
    }
}

impl<T: CdrWrite + ?Sized> CdrWrite for Box<T> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        CdrWrite::write(&**self, writer)
    }
}

impl<T: CdrWrite + ?Sized> CdrWrite for Arc<T> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        CdrWrite::write(&**self, writer)
    }
}

impl<T: CdrWrite + ToOwned + ?Sized> CdrWrite for Cow<'_, T> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write::<T>(&**self)
    }
}

// ==================== //
//       Arc Buf        //
// ==================== //

/// A packed element that is a CDR primitive exactly as it is in memory, aligned on its own
/// size, so that an array of them is shared out of the octets it is read from, see
/// [`CdrReader::read_buf`].
///
/// A [`crate::dev::DevState`] is not one: it is an enum, which is an ulong on the wire.
pub trait CdrElement: PackedElement {}
impl CdrElement for bool {}
impl CdrElement for u8 {}
impl CdrElement for u16 {}
impl CdrElement for i16 {}
impl CdrElement for u32 {}
impl CdrElement for i32 {}
impl CdrElement for u64 {}
impl CdrElement for i64 {}
impl CdrElement for f32 {}
impl CdrElement for f64 {}
impl CdrElement for DevEnum {}

impl<T: CdrElement> CdrRead<'_> for Spectrum<T> {
    /// A sequence, shared out of the buffer the reader reads when it can, see
    /// [`CdrReader::read_buf`].
    #[inline]
    fn read(reader: &mut CdrReader<'_>) -> Result<Self> {
        let len = reader.read_sequence_len()?;
        reader.read_buf(len)
    }
}

impl CdrRead<'_> for Scalar<str> {
    /// A single string, shared out of the buffer the reader reads when it can, see
    /// [`CdrReader::read_string_buf`].
    #[inline]
    fn read(reader: &mut CdrReader<'_>) -> Result<Self> {
        Ok(reader.read_string_buf(1)?.into_scalar().expect("a single string was read"))
    }
}

impl CdrRead<'_> for Spectrum<str> {
    #[inline]
    fn read(reader: &mut CdrReader<'_>) -> Result<Self> {
        let len = reader.read_sequence_len()?;
        reader.read_string_buf(len)
    }
}

impl<T: CdrElement> CdrWrite for Spectrum<T> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_sequence_len(self.len())?;
        writer.write_data(self)
    }
}

impl CdrWrite for Spectrum<str> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write_sequence_len(self.len())?;
        writer.write_string_data(self)
    }
}

// ==================== //
//        Tuples        //
// ==================== //

/// Implement the CDR types of the tuples, which are the plain structures of the CDR: their
/// members follow each other with no length prefix, in the order they are declared in.
macro_rules! impl_cdr_type_tuple {
    ( $( $member:ident ),* ) => {

        impl<'d, $( $member: CdrRead<'d>, )*> CdrRead<'d> for ( $( $member, )* ) {
            #[inline]
            #[allow(unused)]
            fn read(reader: &mut CdrReader<'d>) -> Result<Self> {
                // The members of a tuple expression are evaluated left to right, which is
                // the order they are laid out in.
                Ok(( $( $member::read(reader)?, )* ))
            }
        }

        impl<$( $member: CdrWrite, )*> CdrWrite for ( $( $member, )* ) {
            #[inline]
            #[allow(unused, non_snake_case)]
            fn write(&self, writer: &mut CdrWriter) -> Result<()> {
                let ( $( $member, )* ) = self;
                $( writer.write($member)?; )*
                Ok(())
            }
        }

    };
}

impl_cdr_type_tuple!();
impl_cdr_type_tuple!(A);
impl_cdr_type_tuple!(A, B);
impl_cdr_type_tuple!(A, B, C);
impl_cdr_type_tuple!(A, B, C, D);
impl_cdr_type_tuple!(A, B, C, D, E);
impl_cdr_type_tuple!(A, B, C, D, E, F);
impl_cdr_type_tuple!(A, B, C, D, E, F, G);
impl_cdr_type_tuple!(A, B, C, D, E, F, G, H);
impl_cdr_type_tuple!(A, B, C, D, E, F, G, H, I);
impl_cdr_type_tuple!(A, B, C, D, E, F, G, H, I, J);
impl_cdr_type_tuple!(A, B, C, D, E, F, G, H, I, J, K);
impl_cdr_type_tuple!(A, B, C, D, E, F, G, H, I, J, K, L);

// ==================== //
//    Buffer adapters   //
// ==================== //

impl CdrWrite for BufMut<u8, 1> {
    #[inline]
    fn write(&self, writer: &mut CdrWriter) -> Result<()> {
        writer.write::<[u8]>(self.as_slice())
    }
}

// ==================== //
//         Utils        //
// ==================== //

/// Internal trait to mark a type as a CDR primitive, read without being checked.
///
/// # Safety
///
/// Every bit pattern of the type must be a valid value of it, which is why `bool` is not
/// one, and the type must hold no padding.
unsafe trait CdrNum: Copy + 'static {
    fn swap_bytes(self) -> Self;
}

macro_rules! impl_cdr_num {
    ( $ty:ty, $swap:expr ) => {
        unsafe impl CdrNum for $ty {
            #[inline]
            fn swap_bytes(self) -> Self {
                $swap(self)
            }
        }
    };
}

impl_cdr_num!(u8, u8::swap_bytes);
impl_cdr_num!(u16, u16::swap_bytes);
impl_cdr_num!(i16, i16::swap_bytes);
impl_cdr_num!(u32, u32::swap_bytes);
impl_cdr_num!(i32, i32::swap_bytes);
impl_cdr_num!(u64, u64::swap_bytes);
impl_cdr_num!(i64, i64::swap_bytes);
impl_cdr_num!(f32, |val: f32| f32::from_bits(val.to_bits().swap_bytes()));
impl_cdr_num!(f64, |val: f64| f64::from_bits(val.to_bits().swap_bytes()));

/// Internal function to get offset to apply to a pointer to align it.
#[inline]
const fn align_offset(addr: usize, align: usize) -> usize {
    (align - (addr & (align - 1))) & (align - 1)
}

/// Internal shorthand for the errors the reader/writer returns.
#[cold]
fn invalid_data_err(description: impl Into<Cow<'static, str>>) -> Error {
    Error::new(ErrorKind::InvalidData).with_description(description)
}

/// Internal shorthand for the error a failed allocation returns.
#[cold]
fn out_of_memory_err(error: TryReserveError, description: impl Into<Cow<'static, str>>) -> Error {
    Error::new(ErrorKind::Internal)
        .with_description(description)
        .with_source(Box::new(error))
}

/// The span of a string between these offsets, which must fit 32 bits.
#[inline]
fn span(start: usize, end: usize) -> Result<Range<u32>> {
    u32::try_from(end)
        .map(|end| start as u32..end)
        .map_err(|_| invalid_data_err(format!("Failed to read strings, their {end} octets do not fit into 32 bits")))
}


// ==================== //
//        Tests         //
// ==================== //

#[cfg(test)]
mod tests {

    use super::*;

    const NON_NATIVE: Endian = match Endian::NATIVE {
        Endian::Little => Endian::Big,
        Endian::Big => Endian::Little,
    };

    const CODESETS: Option<Codesets> = Some(Codesets {
        codeset: Codeset::Utf8,
        wide_codeset: WideCodeset::Utf16,
    });

    /// A copy of the given octets in a buffer, whose payload is 8-aligned, so that the
    /// tests control the address the reader sees and not the allocator.
    fn aligned(bytes: &[u8]) -> Buf<u8, 1> {
        Buf::from(bytes)
    }

    /// The CDR alignment is relative to the origin of the buffer and not to the address of
    /// its octets, so a reader over a slice starting anywhere aligns from its own origin.
    #[test]
    fn logical_align() {

        let mut bytes = vec![0u8; 16];
        bytes[0] = 0xAA;
        bytes[2..4].copy_from_slice(&0x1122u16.to_ne_bytes());
        bytes[4..8].copy_from_slice(&0x33445566u32.to_ne_bytes());
        bytes[8..16].copy_from_slice(&0x1122334455667788u64.to_ne_bytes());

        // The very same stream, once starting on an 8-aligned address, and once 2 octets
        // in, so that every primitive lands on an address that is not aligned for it.
        for origin in [0, 2] {

            let buf = aligned(&[&vec![0u8; origin][..], &bytes].concat());
            let mut reader = CdrReader::new(&buf).substream(&buf.as_bytes()[origin..]);

            assert_eq!(reader.read_octet().unwrap(), 0xAA);
            assert_eq!(reader.read_ushort().unwrap(), 0x1122);
            assert_eq!(reader.position(), 4);
            assert_eq!(reader.read_ulong().unwrap(), 0x33445566);
            assert_eq!(reader.read_ulonglong().unwrap(), 0x1122334455667788);
            assert!(reader.is_empty());

        }

    }

    /// A slice of numbers is borrowed straight out of the buffer when its address is
    /// aligned for the type and its octets are in the native order, and copied into an
    /// owned and aligned buffer when it is not.
    #[test]
    fn num_borrow() {

        let values = [0x11223344u32, 0x55667788];
        let bytes = values.map(u32::to_ne_bytes).concat();

        // The octets of a buffer are 8-aligned, so the values can be borrowed in place.
        let buf = aligned(&bytes);
        let borrowed = CdrReader::new(&buf).read_ulong_slice(2).unwrap();
        assert!(matches!(borrowed, Cow::Borrowed(_)));
        assert_eq!(&*borrowed, &values);

        // Not the native order, so they are copied out and swapped there, the buffer being
        // left untouched.
        let swapped = CdrReader::new(&buf).with_endian(NON_NATIVE).read_ulong_slice(2).unwrap();
        assert!(matches!(swapped, Cow::Owned(_)));
        assert_eq!(&*swapped, &values.map(u32::swap_bytes));
        assert_eq!(buf.as_bytes(), &bytes[..]);

        // The origin of this one is 2 octets in, so the values are 4-aligned relative to it
        // but their address is not, and they have to be copied out for that alone.
        let buf = aligned(&[&[0u8; 2][..], &bytes].concat());
        let copied = CdrReader::new(&buf).substream(&buf.as_bytes()[2..]).read_ulong_slice(2).unwrap();
        assert!(matches!(copied, Cow::Owned(_)));
        assert_eq!(&*copied, &values);

        // A single byte has no order of its own and is always aligned, so the octets and
        // the booleans are borrowed whatever the endianness is.
        let buf = aligned(&[0, 1, 0]);
        let mut reader = CdrReader::new(&buf).substream(&buf.as_bytes()[1..]).with_endian(NON_NATIVE);
        assert!(matches!(reader.read_bool_slice(2).unwrap(), Cow::Borrowed(_)));
        assert!(matches!(reader.read_octet_slice(0).unwrap(), Cow::Borrowed(_)));

    }

    /// An array read out of a buffer shares its octets when they are aligned and in the
    /// native order, outliving the buffer, and is copied otherwise, the buffer being left
    /// untouched.
    #[test]
    fn data_sharing() {

        let values = [0x11223344u32, 0x55667788];
        let bytes = values.map(u32::to_ne_bytes).concat();
        let buf = aligned(&[&[0u8; 4][..], &bytes].concat());

        let mut reader = CdrReader::new(&buf);
        reader.skip(4).unwrap();
        let shared = reader.read_buf::<u32>(2).unwrap();
        assert_eq!(shared.as_bytes().as_ptr(), buf.as_bytes()[4..].as_ptr());

        let mut reader = CdrReader::new(&buf).with_endian(NON_NATIVE);
        reader.skip(4).unwrap();
        let swapped = reader.read_buf::<u32>(2).unwrap();
        assert_eq!(swapped.as_slice(), &values.map(u32::swap_bytes));
        assert!(swapped.is_unique());
        assert_eq!(&buf.as_bytes()[4..], &bytes[..]);

        // A substream shares the very same buffer.
        assert!(!CdrReader::new(&buf).substream(&buf.as_bytes()[4..]).read_buf::<u32>(2).unwrap().is_unique());

        drop(buf);
        assert_eq!(shared.as_slice(), &values);

    }

    /// An array holding an invalid value is rejected, like a single one.
    #[test]
    fn data_invalid() {
        let buf = aligned(&[0, 1, 2]);
        assert!(CdrReader::new(&buf).read_buf::<bool>(3).is_err());
        assert!(CdrReader::new(&buf).read_buf::<bool>(2).is_ok());
    }

    /// Strings are shared out of the buffer when the codeset leaves them as they are on the
    /// wire, and transcoded into a buffer of their own otherwise: ISO 8859-1, the default
    /// codeset, has 'é' as a single octet, which is not UTF-8.
    #[test]
    fn string_buf() {

        let strings = Spectrum::<str>::from(&["ab", "", "cdé"][..]);

        for (codesets, shared) in [(CODESETS, true), (None, false)] {
            let mut writer = CdrWriter::new().with_codesets(codesets);
            writer.write_string_data(&strings).unwrap();
            writer.write_ulong(7).unwrap();
            let buf = writer.into_buf().freeze();
            let mut reader = CdrReader::new(&buf).with_codesets(codesets);
            let data = reader.read_string_buf(3).unwrap();
            assert_eq!(data, strings);
            assert_eq!(data.is_unique(), !shared);
            // The reader is left right past the strings, whichever way they were read.
            assert_eq!(reader.read_ulong().unwrap(), 7);
        }

        // Each string is at least 5 octets, so no room is made for more than fit.
        assert!(CdrReader::new(&aligned(&[0; 9])).read_string_buf(2).is_err());

    }

    /// Data are written and read back as the sequences they are.
    #[test]
    fn data_seq() {

        let doubles = Spectrum::<f64>::from(&[1.5, -2.0][..]);
        let strings = Spectrum::<str>::from(&["a", "bc"][..]);

        let mut writer = CdrWriter::new();
        writer.write(&(&doubles, &strings, &doubles)).unwrap();
        let buf = writer.into_buf().freeze();

        let mut reader = CdrReader::new(&buf);
        assert_eq!(reader.read::<Spectrum<f64>>().unwrap(), doubles);
        assert_eq!(reader.read::<Spectrum<str>>().unwrap(), strings);
        assert_eq!(reader.read::<Spectrum<f64>>().unwrap(), doubles);
        assert!(reader.is_empty());

    }

    /// Only the 0 and 1 bit patterns are valid booleans, so anything else is rejected
    /// before the slice is reinterpreted as booleans.
    #[test]
    fn bool_bits() {
        let buf = aligned(&[0, 1, 0xFF]);
        let mut reader = CdrReader::new(&buf);
        assert_eq!(&*reader.read_bool_slice(2).unwrap(), &[false, true]);
        assert!(reader.read_bool().is_err());
    }

    /// A read that does not fit leaves the reader exactly where it was, so nothing is ever
    /// half read, and the length of a slice must not overflow when it is scaled to octets.
    #[test]
    fn overflow() {

        let buf = aligned(&[1, 2, 3, 4, 5, 6]);
        let mut reader = CdrReader::new(&buf);

        reader.read_octet().unwrap();
        assert!(reader.read_ulong().is_err());  // Aligned on 4, so it would end past the 6th.
        assert!(reader.read_ulong_slice(usize::MAX).is_err());  // len * size overflows.
        assert_eq!(reader.position(), 1);
        assert_eq!(reader.read_ushort().unwrap(), u16::from_ne_bytes([3, 4]));

    }

    /// The octets are never modified, so a clone is a rewind point: what it reads ahead
    /// leaves the reader it was cloned from untouched.
    #[test]
    fn clone_read() {

        let buf = aligned(&0xDEADBEEFu32.to_ne_bytes());
        let reader = CdrReader::new(&buf);

        let mut ahead = reader.clone();
        assert_eq!(ahead.read_ulong().unwrap(), 0xDEADBEEF);

        assert_eq!(reader.position(), 0);
        assert_eq!(reader.clone().read_ulong().unwrap(), 0xDEADBEEF);

    }

    /// A string is borrowed straight out of the buffer when its codeset allows it, and its
    /// length counts the terminating nul octet in.
    #[test]
    fn str_borrow() {

        // "héllo" in UTF-8, the terminating nul octet being counted in its length.
        let buf = aligned(&[&7u32.to_ne_bytes()[..], b"h\xC3\xA9llo\0"].concat());

        let s = CdrReader::new(&buf).with_codesets(CODESETS).read_string().unwrap();
        assert!(matches!(s, Cow::Borrowed(_)));
        assert_eq!(&*s, "héllo");

        // Latin-1 does not map to UTF-8 one octet at a time, so it has to decode into a
        // string of its own, and the very same octets decode to something else.
        let latin1 = Some(Codesets { codeset: Codeset::Iso8859_1, ..CODESETS.unwrap() });
        let s = CdrReader::new(&buf).with_codesets(latin1).read_string().unwrap();
        assert!(matches!(s, Cow::Owned(_)));
        assert_eq!(&*s, "héllo");

    }

    /// A string is nul terminated and its length counts that octet in, so a length of zero
    /// is invalid, and so is a last octet that is not the nul one.
    #[test]
    fn str_nul() {

        let buf = aligned(&[&0u32.to_ne_bytes()[..]].concat());
        assert!(CdrReader::new(&buf).with_codesets(CODESETS).read_string().is_err());

        let buf = aligned(&[&5u32.to_ne_bytes()[..], b"hello"].concat());
        assert!(CdrReader::new(&buf).with_codesets(CODESETS).read_string().is_err());

        // Not enough octets left for the length it announces.
        let buf = aligned(&[&6u32.to_ne_bytes()[..], b"hello"].concat());
        assert!(CdrReader::new(&buf).with_codesets(CODESETS).read_string().is_err());

    }

    /// A wide string is measured in octets and not in code units, and those are borrowed
    /// in place under the same conditions as any other number.
    #[test]
    fn wstr_octets() {

        let units = [0x0068u16, 0x00E9, 0x006C, 0x006C, 0x006F];  // "héllo"
        let expected = Utf16String::from_str("héllo");

        // GIOP 1.1 counts the terminating nul code unit in, GIOP 1.2 does not, and both
        // are read the same way.
        for tail in [&[0u16][..], &[]] {

            let units = [&units[..], tail].concat();
            let len = (units.len() * 2) as u32;

            let mut bytes = len.to_ne_bytes().to_vec();
            for unit in &units {
                bytes.extend_from_slice(&unit.to_ne_bytes());
            }

            let buf = aligned(&bytes);
            let w = CdrReader::new(&buf).with_codesets(CODESETS).read_wstring().unwrap();
            assert!(matches!(w, Cow::Borrowed(_)));
            assert_eq!(&*w, &*expected);

            // The very same wide string in the other order: the code units cannot be
            // borrowed then, so neither can the string decoded out of the copy they are
            // swapped in.
            let mut swapped = len.swap_bytes().to_ne_bytes().to_vec();
            for unit in &units {
                swapped.extend_from_slice(&unit.swap_bytes().to_ne_bytes());
            }

            let buf = aligned(&swapped);
            let w = CdrReader::new(&buf).with_endian(NON_NATIVE).with_codesets(CODESETS).read_wstring().unwrap();
            assert!(matches!(w, Cow::Owned(_)));
            assert_eq!(&*w, &*expected);

        }

    }

    /// A wide string needs a negotiated wide codeset, there is no default one, and its
    /// length must be a whole number of code units.
    #[test]
    fn wstr_codeset() {

        let buf = aligned(&[&2u32.to_ne_bytes()[..], &0u16.to_ne_bytes()].concat());
        assert!(CdrReader::new(&buf).read_wstring().is_err());
        assert!(CdrReader::new(&buf).with_codesets(CODESETS).read_wstring().unwrap().is_empty());

        // An empty wide string, which GIOP 1.2 holds no nul code unit at all for.
        let buf = aligned(&0u32.to_ne_bytes());
        assert!(CdrReader::new(&buf).with_codesets(CODESETS).read_wstring().unwrap().is_empty());

        let buf = aligned(&[&3u32.to_ne_bytes()[..], b"abc"].concat());
        assert!(CdrReader::new(&buf).with_codesets(CODESETS).read_wstring().is_err());

    }

    /// An encapsulation restarts the CDR alignment at its own origin and carries its own
    /// byte order, and it is read in place, out of the very same octets.
    #[test]
    fn encaps_align() {

        // The encapsulation itself: its byte order flag, an octet, and an ulong aligned on
        // its own origin, so with 2 octets of padding in between.
        let mut inner = vec![Endian::NATIVE.is_little() as u8, 0xAA, 0, 0];
        inner.extend_from_slice(&0xDEADBEEFu32.to_ne_bytes());

        // The stream it is held in: an octet, then the encapsulation as a sequence of
        // octets, so its length lands on the next 4-aligned offset.
        let mut outer = vec![0x2A, 0, 0, 0];
        outer.extend_from_slice(&(inner.len() as u32).to_ne_bytes());
        outer.extend_from_slice(&inner);

        let buf = aligned(&outer);
        let mut reader = CdrReader::new(&buf);
        assert_eq!(reader.read_octet().unwrap(), 0x2A);

        let mut encapsulation = reader.read_encapsulation().unwrap();
        assert_eq!(encapsulation.endian(), Endian::NATIVE);
        assert_eq!(encapsulation.position(), 1);  // The byte order flag is already read.
        assert_eq!(encapsulation.read_octet().unwrap(), 0xAA);
        assert_eq!(encapsulation.read_ulong().unwrap(), 0xDEADBEEF);
        assert!(encapsulation.is_empty());
        assert!(reader.is_empty());

        // It borrows the very same octets, nothing was copied out of the buffer.
        assert!(buf.as_bytes().as_ptr_range().contains(&encapsulation.buffer().as_ptr()));

    }

    /// The byte order of an encapsulation is its own, so it applies to the numbers it holds
    /// and to nothing else, and so are its codesets: they are a property of the
    /// encapsulation itself, so nothing is inherited from the stream holding it and the
    /// defaults of GIOP 1.0 apply until a usage defines otherwise.
    #[test]
    fn encaps_endian() {

        let mut inner = vec![NON_NATIVE.is_little() as u8, 0, 0, 0];
        inner.extend_from_slice(&0xDEADBEEFu32.swap_bytes().to_ne_bytes());
        inner.extend_from_slice(&4u32.swap_bytes().to_ne_bytes());
        inner.extend_from_slice(b"abc\0");

        let mut outer = (inner.len() as u32).to_ne_bytes().to_vec();
        outer.extend_from_slice(&inner);
        outer.extend_from_slice(&0x11223344u32.to_ne_bytes());

        let buf = aligned(&outer);
        let mut reader = CdrReader::new(&buf).with_codesets(CODESETS);

        let mut encapsulation = reader.read_encapsulation().unwrap();
        assert_eq!(encapsulation.endian(), NON_NATIVE);
        assert_eq!(encapsulation.read_ulong().unwrap(), 0xDEADBEEF);

        // The codesets of the stream are not the ones of the encapsulation, which is left
        // on the ISO 8859-1 of GIOP 1.0, and holds no wide string at all.
        assert_eq!(encapsulation.codesets(), None);
        assert_eq!(encapsulation.codeset(), Codeset::Iso8859_1);
        assert_eq!(&*encapsulation.read_string().unwrap(), "abc");

        // The stream it was read out of kept its own byte order.
        assert_eq!(reader.endian(), Endian::NATIVE);
        assert_eq!(reader.read_ulong().unwrap(), 0x11223344);

    }

    /// An encapsulation that does not even hold its byte order flag is invalid, and one
    /// that announces more octets than there are left cannot be read either.
    #[test]
    fn encaps_flag() {

        let buf = aligned(&0u32.to_ne_bytes());
        assert!(CdrReader::new(&buf).read_encapsulation().is_err());

        let buf = aligned(&[&8u32.to_ne_bytes()[..], &[0; 4]].concat());
        assert!(CdrReader::new(&buf).read_encapsulation().is_err());

        // The byte order flag is a boolean, so nothing else than 0 or 1 will do.
        let buf = aligned(&[&1u32.to_ne_bytes()[..], &[0xFF]].concat());
        assert!(CdrReader::new(&buf).read_encapsulation().is_err());

    }

    /// Every primitive is aligned on its own size, relative to the origin of the stream,
    /// and the padding that is skipped is zeroed because it is never handed out.
    #[test]
    fn write_padding() {

        let mut writer = CdrWriter::new();

        writer.write_octet(0xFF).unwrap();
        assert_eq!(writer.position(), 1);

        writer.write_ulong(0x11223344).unwrap();
        assert_eq!(writer.position(), 8);

        writer.write_ulonglong(u64::MAX).unwrap();
        assert_eq!(writer.position(), 16);

        let buf = writer.into_buf().freeze();
        assert_eq!(buf.len(), 16);
        assert_eq!(&buf.as_bytes()[1..4], &[0, 0, 0]);
        assert_eq!(&buf.as_bytes()[4..8], &0x11223344u32.to_ne_bytes());

    }

    /// A codeset that encodes to fewer octets than its worst case must not leave a hole in
    /// the stream, the room it did not need going to the primitives that follow.
    #[test]
    fn write_room_back() {

        // Latin-1 encodes 'é' to a single octet where its UTF-8 source takes two.
        let codesets = Some(Codesets { codeset: Codeset::Iso8859_1, ..CODESETS.unwrap() });

        let mut writer = CdrWriter::new().with_codesets(codesets);
        writer.write_string("héllo").unwrap();
        writer.write_octet(0xAA).unwrap();

        // The 5 octets of the string, its nul one, and the octet right after it.
        let buf = writer.into_buf().freeze();
        assert_eq!(buf.as_bytes(), [&6u32.to_ne_bytes()[..], b"h\xE9llo\0\xAA"].concat());

    }

    /// The unused room is given back to the buffer as well, and not only to the primitives
    /// that follow, so that a write that grew the buffer for its worst case does not leave
    /// a tail past the position when nothing follows it.
    #[test]
    fn write_room_shrink() {

        let codesets = Some(Codesets { codeset: Codeset::Iso8859_1, ..CODESETS.unwrap() });

        let mut writer = CdrWriter::new().with_codesets(codesets);
        writer.write_string("héllo").unwrap();

        // Worst case is the 6 octets of the UTF-8 source plus the nul one, so the buffer
        // grew to 11 octets, and the encoding only needed 10 of them.
        assert_eq!(writer.position(), 10);
        assert_eq!(writer.buf.len(), 10);

    }

    /// Only the tail the write itself grew is given back, so a position that was moved back
    /// into a longer stream keeps everything past it.
    #[test]
    fn write_rewind_tail() {

        let codesets = Some(Codesets { codeset: Codeset::Iso8859_1, ..CODESETS.unwrap() });

        let mut writer = CdrWriter::new().with_codesets(codesets);
        writer.write_octet_slice(&[0xAA; 32]).unwrap();

        writer.set_position(0);
        writer.write_string("héllo").unwrap();

        assert_eq!(writer.position(), 10);
        assert_eq!(writer.buf.len(), 32);
        assert_eq!(writer.buf.as_bytes()[10..], [0xAA; 22]);

    }

    /// The position moves back to an offset that was noted earlier, which is how a length
    /// that is only known afterward is filled in without encoding into a buffer of its own.
    #[test]
    fn write_rewind_len() {

        let mut writer = CdrWriter::new();

        let len_at = writer.position();
        writer.write_ulong(0).unwrap();  // Placeholder, written below.

        let start = writer.position();
        writer.write_octet_slice(b"abc").unwrap();
        let end = writer.position();

        writer.set_position(len_at);
        writer.write_ulong((end - start) as u32).unwrap();
        writer.set_position(end);

        // The octets that were written over are the only ones that changed.
        let buf = writer.into_buf().freeze();
        assert_eq!(buf.as_bytes(), [&3u32.to_ne_bytes()[..], b"abc"].concat());

    }

    /// An encapsulation restarts the CDR alignment at its own origin, and is written in
    /// place, into the very same buffer, its length being filled in once it is written.
    #[test]
    fn write_encaps() {

        let mut writer = CdrWriter::new();
        writer.write_octet(0x2A).unwrap();

        writer.write_encapsulation(|encapsulation| {

            // The length of the encapsulation was aligned on 4, so its origin is the 8th
            // octet, and the byte order flag is already written.
            assert_eq!(encapsulation.position(), 9);

            encapsulation.write_octet(0xAA)?;

            // Aligned on 4 relative to the origin of the encapsulation, which is the 12th
            // octet of the buffer and not the 16th.
            encapsulation.write_ulong(0xDEADBEEF)?;
            assert_eq!(encapsulation.position(), 16);

            Ok(())

        }).unwrap();

        let buf = writer.into_buf().freeze();
        assert_eq!(buf.len(), 16);
        assert_eq!(&buf.as_bytes()[4..8], &8u32.to_ne_bytes());  // 8 octets of encapsulation.

        let mut reader = CdrReader::new(&buf);
        assert_eq!(reader.read_octet().unwrap(), 0x2A);

        let mut encapsulation = reader.read_encapsulation().unwrap();
        assert_eq!(encapsulation.read_octet().unwrap(), 0xAA);
        assert_eq!(encapsulation.read_ulong().unwrap(), 0xDEADBEEF);
        assert!(encapsulation.is_empty());
        assert!(reader.is_empty());

    }

    /// An encapsulation that fails to be written is dropped from the buffer, so the stream
    /// carries on exactly where it was.
    #[test]
    fn write_encaps_failed() {

        let wide = Utf16String::from_str("hello");

        let mut writer = CdrWriter::new();
        writer.write_ulonglong(1).unwrap();

        let res = writer.write_encapsulation(|encapsulation| {
            encapsulation.write_ulong(0xDEADBEEF)?;
            // No wide codeset has been negotiated, so this one cannot be encoded.
            encapsulation.write_wstring(&wide)
        });

        assert!(res.is_err());
        assert_eq!(writer.position(), 8);

        writer.write_ulonglong(2).unwrap();

        let buf = writer.into_buf().freeze();
        assert_eq!(buf.len(), 16);
        assert_eq!(&*CdrReader::new(&buf).read_ulonglong_slice(2).unwrap(), &[1, 2]);

    }

    /// A buffer that is recycled leaks nothing of the stream it held before through the
    /// octets no primitive covers, which the writer zeroes.
    #[test]
    fn write_recycle() {

        let mut writer = CdrWriter::from_buf(BufMut::from(&[0xFF; 16][..]));

        // Carrying on where the buffer stopped, the CDR alignment being relative to its
        // very first octet.
        assert_eq!(writer.position(), 16);
        writer.write_octet(1).unwrap();

        writer.clear();
        assert_eq!(writer.position(), 0);

        writer.write_octet(1).unwrap();
        writer.write_ulong(2).unwrap();

        let buf = writer.into_buf().freeze();
        assert_eq!(buf.as_bytes(), [&[1, 0, 0, 0][..], &2u32.to_ne_bytes()].concat());

    }

    /// A buffer that cannot grow is an error and not an abort, which is what the writer
    /// reports instead of the octets it could not make room for.
    #[test]
    fn write_no_grow() {

        // Far more octets than an allocation could ever hold, so the reservation fails
        // without even being attempted.
        let mut buf = BufMut::<u8, 1>::new();
        assert!(buf.try_reserve(usize::MAX).is_err());
        assert!(buf.is_empty());

        let mut writer = CdrWriter::from_buf(BufMut::from(&[0xAA; 8][..]));
        writer.set_position(0);

        // The buffer holds these ones already, so nothing has to grow for them.
        writer.write_ulonglong(1).unwrap();
        assert_eq!(writer.into_buf().freeze().as_bytes(), &1u64.to_ne_bytes());

    }

    /// An encapsulation holds encapsulations of its own, each of them restarting the CDR
    /// alignment at its own origin, and each being closed when its writer is dropped.
    #[test]
    fn write_encaps_nested() {

        let mut writer = CdrWriter::new();

        writer.write_encapsulation(|outer| {

            outer.write_octet(0xAA)?;

            outer.write_encapsulation(|inner| {
                inner.write_octet(0xBB)?;
                // Aligned on 4 relative to the origin of the inner encapsulation only.
                inner.write_ulong(0xDEADBEEF)
            })?;

            outer.write_ulong(0x11223344)

        }).unwrap();

        let buf = writer.into_buf().freeze();
        let mut reader = CdrReader::new(&buf);

        let mut outer = reader.read_encapsulation().unwrap();
        assert_eq!(outer.read_octet().unwrap(), 0xAA);

        let mut inner = outer.read_encapsulation().unwrap();
        assert_eq!(inner.read_octet().unwrap(), 0xBB);
        assert_eq!(inner.read_ulong().unwrap(), 0xDEADBEEF);
        assert!(inner.is_empty());

        assert_eq!(outer.read_ulong().unwrap(), 0x11223344);
        assert!(outer.is_empty());
        assert!(reader.is_empty());

    }

    /// Everything a writer writes is what a reader reads back out of it, the encapsulation
    /// included, which is the whole point of the two of them.
    #[test]
    fn round_trip() {

        let wide = Utf16String::from_str("héllo");

        let mut writer = CdrWriter::new().with_codesets(CODESETS);

        writer.write_endian().unwrap();
        writer.write_octet(0x2A).unwrap();
        writer.write_bool(true).unwrap();
        writer.write_short(-2).unwrap();
        writer.write_ulong(0xDEADBEEF).unwrap();
        writer.write_ulonglong(0x1122334455667788).unwrap();
        writer.write_double(1.5).unwrap();
        writer.write_string("hello").unwrap();
        writer.write_wstring(&wide).unwrap();
        writer.write_sequence_len(3).unwrap();
        writer.write_long_slice(&[-1, 0, 1]).unwrap();

        writer.write_encapsulation(|encapsulation| {
            // The codesets are not inherited, an encapsulation carries its own.
            assert_eq!(encapsulation.codesets(), None);
            encapsulation.write_octet(0xAA)?;
            encapsulation.write_float_slice(&[0.5, 1.5])?;
            encapsulation.write_string("abc")
        }).unwrap();

        let buf = writer.into_buf().freeze();
        let mut reader = CdrReader::new(&buf).with_codesets(CODESETS);

        assert_eq!(reader.read_endian().unwrap(), Endian::NATIVE);
        assert_eq!(reader.read_octet().unwrap(), 0x2A);
        assert!(reader.read_bool().unwrap());
        assert_eq!(reader.read_short().unwrap(), -2);
        assert_eq!(reader.read_ulong().unwrap(), 0xDEADBEEF);
        assert_eq!(reader.read_ulonglong().unwrap(), 0x1122334455667788);
        assert_eq!(reader.read_double().unwrap(), 1.5);
        assert_eq!(&*reader.read_string().unwrap(), "hello");
        assert_eq!(&*reader.read_wstring().unwrap(), &*wide);
        assert_eq!(reader.read_sequence_len().unwrap(), 3);
        assert_eq!(&*reader.read_long_slice(3).unwrap(), &[-1, 0, 1]);

        let mut encapsulation = reader.read_encapsulation().unwrap();
        assert_eq!(encapsulation.read_octet().unwrap(), 0xAA);
        assert_eq!(&*encapsulation.read_float_slice(2).unwrap(), &[0.5, 1.5]);
        assert_eq!(&*encapsulation.read_string().unwrap(), "abc");
        assert!(encapsulation.is_empty());
        assert!(reader.is_empty());

    }


    /// A sequence of a CDR builtin type is read out of the stream and copied into it at
    /// once, while one of a constructed type is read and written value by value: this is
    /// the hand-rolled specialization of [`CdrRead::_read_slice`].
    #[test]
    fn num_seq() {

        let mut writer = CdrWriter::new().with_codesets(CODESETS);
        writer.write::<[u32]>(&[1, 2, 3]).unwrap();
        writer.write::<[&str]>(&["hello", "world"]).unwrap();

        let buf = writer.into_buf().freeze();
        let mut reader = CdrReader::new(&buf).with_codesets(CODESETS);

        let numbers = reader.read::<Cow<'_, [u32]>>().unwrap();
        assert!(matches!(numbers, Cow::Borrowed(_)));
        assert_eq!(&*numbers, &[1, 2, 3]);

        // A string is not a builtin type, so the sequence of them is owned, even though
        // each of its strings still borrows the stream in place.
        let strings = reader.read::<Cow<'_, [Cow<'_, str>]>>().unwrap();
        assert!(matches!(strings, Cow::Owned(_)));
        assert!(matches!(strings[0], Cow::Borrowed(_)));
        assert_eq!(&*strings, &["hello", "world"]);

        assert!(reader.is_empty());

    }

    /// A value that is held rather than borrowed goes out exactly like the borrowed form of
    /// it, whichever of the owning types is holding it, so that a caller never has to spell
    /// a sequence out one value at a time to write it.
    #[test]
    fn owned_write() {

        let strings = vec!["hello".to_string(), "world".to_string()];
        let cows = vec![Cow::Borrowed("hello"), Cow::Owned("world".to_string())];

        // Each of these writes the very same sequence of two strings, and the borrowed
        // form is the reference the others are checked against.
        let mut reference = CdrWriter::new().with_codesets(CODESETS);
        reference.write::<[&str]>(&["hello", "world"]).unwrap();
        let reference = reference.into_buf().freeze();

        let mut writers: Vec<(&str, CdrWriter)> = Vec::new();

        macro_rules! written {
            ($name:literal, $value:expr) => {{
                let mut writer = CdrWriter::new().with_codesets(CODESETS);
                writer.write($value).unwrap();
                writers.push(($name, writer));
            }};
        }

        written!("Vec<String>", &strings);
        written!("Vec<Cow<str>>", &cows);
        written!("[String]", &strings[..]);
        written!("Box<[String]>", &strings.clone().into_boxed_slice());
        written!("Arc<[String]>", &Arc::<[String]>::from(&strings[..]));
        written!("&Vec<String>", &&strings);

        for (name, writer) in writers {
            assert_eq!(writer.buffer(), reference.as_bytes(), "{name} was not written like the borrowed form");
        }

        // A single string is the same story, held or borrowed.
        let mut reference = CdrWriter::new().with_codesets(CODESETS);
        reference.write("hello").unwrap();
        let reference = reference.into_buf().freeze();

        for value in [&"hello".to_string() as &dyn CdrWrite, &Box::<str>::from("hello"), &Arc::<str>::from("hello")] {
            let mut writer = CdrWriter::new().with_codesets(CODESETS);
            writer.write(value).unwrap();
            assert_eq!(writer.buffer(), reference.as_bytes());
        }

    }

    /// The octets of a sequence are the only ones that are always borrowed straight out of
    /// the stream, whatever its byte order is, which is what an object key or the octets of
    /// an encapsulation are read as.
    #[test]
    fn octet_seq() {

        // A stream in the other byte order: only the length is swapped, an octet has no
        // order of its own.
        let mut bytes = 3u32.swap_bytes().to_ne_bytes().to_vec();
        bytes.extend_from_slice(&[1, 2, 3]);

        let buf = aligned(&bytes);

        let mut reader = CdrReader::new(&buf).with_endian(NON_NATIVE);
        assert_eq!(reader.read::<&[u8]>().unwrap(), &[1, 2, 3]);

        let mut reader = CdrReader::new(&buf).with_endian(NON_NATIVE);
        assert_eq!(reader.read::<&[u8]>().unwrap(), &[1, 2, 3]);

        // A sequence of a wider type cannot be borrowed in that order, so it is read as the
        // copy it has to be.
        let mut bytes = 2u32.swap_bytes().to_ne_bytes().to_vec();
        bytes.extend_from_slice(&[1, 2, 3, 4]);

        let buf = aligned(&bytes);

        let mut reader = CdrReader::new(&buf).with_endian(NON_NATIVE);
        assert!(matches!(reader.read::<Cow<'_, [u16]>>().unwrap(), Cow::Owned(_)));

    }

    /// A tuple is a plain CDR structure, its members following each other with no length
    /// prefix, and the ones that borrow the stream do so all at once out of a single read.
    #[test]
    fn tuples() {

        let mut writer = CdrWriter::new().with_codesets(CODESETS);
        writer.write(&(&[1u32, 2, 3][..], "hello", 0x2Au8)).unwrap();

        let buf = writer.into_buf().freeze();
        let mut reader = CdrReader::new(&buf).with_codesets(CODESETS);

        let (values, name, tail) = reader.read::<(Cow<'_, [u32]>, Cow<'_, str>, u8)>().unwrap();

        assert!(matches!(values, Cow::Borrowed(_)));
        assert!(matches!(name, Cow::Borrowed(_)));
        assert_eq!(&*values, &[1, 2, 3]);
        assert_eq!(&*name, "hello");
        assert_eq!(tail, 0x2A);
        assert!(reader.is_empty());

    }

    /// A message is encoded through a `&dyn CdrWrite`, so that trait has to stay dyn
    /// compatible, which the sealed run hook is bounded on `Self: Sized` for.
    #[test]
    fn dyn_write() {

        let values: [&dyn CdrWrite; 3] = [&0x2Au8, &0xDEADBEEFu32, &"hello"];

        let mut writer = CdrWriter::new().with_codesets(CODESETS);
        for value in values {
            writer.write(value).unwrap();
        }

        let buf = writer.into_buf().freeze();
        let mut reader = CdrReader::new(&buf).with_codesets(CODESETS);

        assert_eq!(reader.read::<u8>().unwrap(), 0x2A);
        assert_eq!(reader.read::<u32>().unwrap(), 0xDEADBEEF);
        assert_eq!(&*reader.read::<Cow<'_, str>>().unwrap(), "hello");

    }

    /// The owned containers are read out of the very same sequence, whether its values were
    /// borrowed out of the stream or copied out of it.
    #[test]
    fn owned_read() {

        let mut writer = CdrWriter::new();
        writer.write::<[i32]>(&[-1, 0, 1]).unwrap();

        let buf = writer.into_buf().freeze();

        assert_eq!(CdrReader::new(&buf).read::<Vec<i32>>().unwrap(), vec![-1, 0, 1]);
        assert_eq!(&*CdrReader::new(&buf).read::<Box<[i32]>>().unwrap(), &[-1, 0, 1]);

    }


    /// A writer that takes only a few octets at a time is what the origin of a sender is
    /// there for: what it swallowed is acknowledged, and the next call gives out what is
    /// left and never those octets again.
    #[test]
    fn sender_partial_ack() {

        let mut sender = CdrWriter::new().into_sender();
        sender.write_octet_slice(&[0xAA; 100]).unwrap();

        let mut out = Vec::new();

        while !sender.buffer().is_empty() {
            let taken = usize::min(3, sender.buffer().len());
            out.extend_from_slice(&sender.buffer()[..taken]);
            sender.ack(taken);
        }

        assert_eq!(out, [0xAA; 100]);

    }

    /// A buffer laid into a sender goes out at the point it was laid in at and nowhere
    /// else: the octets the writer already held in front of it come first, and the ones
    /// written after it come behind it, which is what a request laid down again in an
    /// earlier version rides on, see [`CdrSender::interleave`].
    #[test]
    fn sender_interleave() {

        let mut sender = CdrWriter::new().into_sender();

        sender.write_octet_slice(&[0xAA; 8]).unwrap();
        sender.interleave(aligned(&[0xBB; 8]));
        sender.write_octet_slice(&[0xCC; 8]).unwrap();

        // Three at a time, so that every run is crossed by a write that stops inside it.
        let mut out = Vec::new();

        while !sender.buffer().is_empty() {
            let taken = usize::min(3, sender.buffer().len());
            out.extend_from_slice(&sender.buffer()[..taken]);
            sender.ack(taken);
        }

        assert_eq!(out, [&[0xAA; 8][..], &[0xBB; 8], &[0xCC; 8]].concat());

    }

    /// One acknowledgement may span several runs, which is a writer that was handed more
    /// than the run at the front of the stream: each run takes what it still holds and the
    /// rest goes to the one behind it.
    #[test]
    fn sender_ack_spans_runs() {

        let mut sender = CdrWriter::new().into_sender();

        sender.write_octet_slice(&[0xAA; 8]).unwrap();
        sender.interleave(aligned(&[0xBB; 8]));
        sender.write_octet_slice(&[0xCC; 8]).unwrap();

        // Past the eight octets of the first run, so it spills into the buffer laid in
        // behind it and then into what was written after that one.
        sender.ack(20);

        assert_eq!(sender.buffer(), [0xCC; 4]);

    }

    /// Moving off the octets that have gone out leaves the stream exactly where it was:
    /// what is left to go out is the very same, the buffers laid in moving along with it.
    #[test]
    fn sender_drop_off() {

        let mut sender = CdrWriter::new().into_sender();

        sender.write_octet_slice(&[0xAA; 8]).unwrap();
        sender.interleave(aligned(&[0xBB; 8]));
        sender.write_octet_slice(&[0xCC; 8]).unwrap();

        sender.ack(4);
        sender.drop_off_origin();

        let mut out = Vec::new();

        while !sender.buffer().is_empty() {
            let taken = sender.buffer().len();
            out.extend_from_slice(&sender.buffer()[..taken]);
            sender.ack(taken);
        }

        assert_eq!(out, [&[0xAA; 4][..], &[0xBB; 8], &[0xCC; 8]].concat());

    }

    /// Acknowledging more than was ever given out is a bug of the caller, and it is caught
    /// where it happens rather than moving the stream to somewhere it never stood.
    #[test]
    #[should_panic]
    fn sender_over_ack() {
        let mut sender = CdrWriter::new().into_sender();
        sender.write_octet_slice(&[0xAA; 8]).unwrap();
        sender.ack(9);
    }

}