foundation_core 0.0.6

// use crate::compati::Mutex;
use std::cell::RefCell;
use std::io::{BufRead, BufReader, BufWriter, Cursor, IoSlice, IoSliceMut, Read, Result, Write};

use foundation_nostd::comp::basic::{Mutex, RwLock};
use std::rc::Rc;
use std::sync::atomic::AtomicPtr;
use std::sync::Arc;

use crate::err;
use derive_more::derive::From;

/// Trait for read streams that support reading with a timeout.
///
/// # Errors
/// Returns an error if reading fails or the requested size exceeds buffer capacity.
pub trait ReadTimeoutOperations: Read {
    /// Reads data from the stream with a timeout.
    ///
    /// # Returns
    /// The number of bytes read, or an error if reading fails or the requested size exceeds buffer capacity.
    ///
    /// # Errors
    /// Returns an error if reading fails or the requested size exceeds buffer capacity.
    fn read_timeout_into(
        &mut self,
        buf: &mut [u8],
        timeout: std::time::Duration,
    ) -> std::result::Result<usize, std::io::Error>;

    /// Sets the read timeout for this stream.
    ///
    /// # Errors
    /// Returns an error if the underlying stream does not support setting read timeouts
    /// or if setting the timeout fails.
    fn set_read_timeout_as(
        &mut self,
        timeout: std::time::Duration,
    ) -> std::result::Result<(), std::io::Error>;

    /// Gets the current read timeout for this stream.
    ///
    /// # Errors
    /// Returns an error if the underlying stream does not support getting read timeouts
    /// or if retrieving the timeout fails.
    fn get_current_read_timeout(
        &self,
    ) -> std::result::Result<Option<std::time::Duration>, std::io::Error>;
}

// BufferCapacity Trait

pub trait BufferCapacity {
    fn read_buffer(&self) -> &[u8];
    fn read_capacity(&self) -> usize;

    fn buffer_length(&self) -> usize {
        self.read_buffer().len()
    }
}

// -- Reader

pub struct BufferedReader<T: ?Sized> {
    inner: BufReader<T>,
}

// -- Constructor

impl<T: Read> BufferedReader<T> {
    pub fn with_capacity(capacity: usize, inner: T) -> Self {
        Self {
            inner: BufReader::with_capacity(capacity, inner),
        }
    }

    pub fn new(inner: T) -> Self {
        Self {
            inner: BufReader::new(inner),
        }
    }
}

impl<T: Read + Write> BufferedReader<BufferedWriter<T>> {
    pub fn get_core_ref(&self) -> &T {
        self.inner.get_ref().get_inner_ref()
    }

    pub fn get_core_mut(&mut self) -> &mut T {
        self.inner.get_mut().get_inner_mut()
    }
}

impl<T: ReadTimeoutOperations> ReadTimeoutOperations for BufferedReader<T> {
    fn read_timeout_into(
        &mut self,
        buf: &mut [u8],
        timeout: std::time::Duration,
    ) -> std::result::Result<usize, std::io::Error> {
        self.inner.get_mut().read_timeout_into(buf, timeout)
    }

    fn set_read_timeout_as(
        &mut self,
        timeout: std::time::Duration,
    ) -> std::result::Result<(), std::io::Error> {
        self.inner.get_mut().set_read_timeout_as(timeout)
    }

    fn get_current_read_timeout(
        &self,
    ) -> std::result::Result<Option<std::time::Duration>, std::io::Error> {
        self.inner.get_ref().get_current_read_timeout()
    }
}

impl<T: Read> BufferedReader<T> {
    pub fn get_ref(&self) -> &BufReader<T> {
        &self.inner
    }

    pub fn get_mut(&mut self) -> &mut BufReader<T> {
        &mut self.inner
    }

    pub fn get_inner_ref(&self) -> &T {
        self.inner.get_ref()
    }

    pub fn get_inner_mut(&mut self) -> &mut T {
        self.inner.get_mut()
    }

    pub fn capacity(&mut self) -> usize {
        self.inner.capacity()
    }

    pub fn buffer(&self) -> &[u8] {
        self.inner.buffer()
    }

    pub fn buffer_len(&self) -> usize {
        self.inner.buffer().len()
    }

    pub fn consume(&mut self, amount: usize) {
        self.inner.consume(amount);
    }
}

impl<T: BufferCapacity> BufferCapacity for BufferedReader<T> {
    fn read_buffer(&self) -> &[u8] {
        self.inner.buffer()
    }

    fn read_capacity(&self) -> usize {
        self.inner.capacity()
    }
}

// -- Implement Read for Size? for both BufRead & Read

impl<T: Read + ?Sized> BufRead for BufferedReader<T> {
    fn fill_buf(&mut self) -> Result<&[u8]> {
        self.inner.fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.inner.consume(amt);
    }
}

impl<T: Read + ?Sized> Read for BufferedReader<T> {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        self.inner.read(buf)
    }

    fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize> {
        self.inner.read_vectored(bufs)
    }

    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        self.inner.read_exact(buf)
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize> {
        self.inner.read_to_end(buf)
    }

    fn read_to_string(&mut self, buf: &mut String) -> Result<usize> {
        self.inner.read_to_string(buf)
    }
}

impl<T: Write + ?Sized> Write for BufferedReader<T> {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        self.inner.get_mut().write(buf)
    }

    fn write_all(&mut self, buf: &[u8]) -> Result<()> {
        self.inner.get_mut().write_all(buf)
    }

    fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
        self.inner.get_mut().write_vectored(bufs)
    }

    fn flush(&mut self) -> Result<()> {
        self.inner.get_mut().flush()
    }
}

// -- Writer

pub struct BufferedWriter<T: ?Sized + Write> {
    inner: BufWriter<T>,
}

// -- Constructor

impl<T: Write> BufferedWriter<T> {
    pub fn new(inner: T) -> Self {
        Self {
            inner: BufWriter::new(inner),
        }
    }

    pub fn with_capacity(capacity: usize, inner: T) -> Self {
        Self {
            inner: BufWriter::with_capacity(capacity, inner),
        }
    }
}

impl<T: ReadTimeoutOperations + Write> ReadTimeoutOperations for BufferedWriter<T> {
    fn read_timeout_into(
        &mut self,
        buf: &mut [u8],
        timeout: std::time::Duration,
    ) -> std::result::Result<usize, std::io::Error> {
        self.inner.get_mut().read_timeout_into(buf, timeout)
    }

    fn set_read_timeout_as(
        &mut self,
        timeout: std::time::Duration,
    ) -> std::result::Result<(), std::io::Error> {
        self.inner.get_mut().set_read_timeout_as(timeout)
    }

    fn get_current_read_timeout(
        &self,
    ) -> std::result::Result<Option<std::time::Duration>, std::io::Error> {
        self.inner.get_ref().get_current_read_timeout()
    }
}

impl<T: Read + Write + BufferCapacity> BufferedWriter<T> {
    pub fn read_capacity(&mut self) -> usize {
        self.inner.get_ref().read_capacity()
    }

    pub fn read_buffer(&self) -> &[u8] {
        self.inner.get_ref().read_buffer()
    }
}

impl<T: Read + Write> BufferedWriter<BufferedReader<T>> {
    pub fn get_core_ref(&self) -> &T {
        self.inner.get_ref().get_inner_ref()
    }

    pub fn get_core_mut(&mut self) -> &mut T {
        self.inner.get_mut().get_inner_mut()
    }
}

impl<T: Write> BufferedWriter<T> {
    pub fn get_ref(&self) -> &BufWriter<T> {
        &self.inner
    }

    pub fn get_mut(&mut self) -> &mut BufWriter<T> {
        &mut self.inner
    }

    pub fn get_inner_ref(&self) -> &T {
        self.inner.get_ref()
    }

    pub fn get_inner_mut(&mut self) -> &mut T {
        self.inner.get_mut()
    }

    pub fn capacity(&mut self) -> usize {
        self.inner.capacity()
    }

    pub fn buffer(&self) -> &[u8] {
        self.inner.buffer()
    }
}

// -- Implement Write for Size? for both BufRead & Read

impl<T: Write + ?Sized> Write for BufferedWriter<T> {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        self.inner.write(buf)
    }

    fn write_all(&mut self, buf: &[u8]) -> Result<()> {
        self.inner.write_all(buf)
    }

    fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
        self.inner.write_vectored(bufs)
    }

    fn flush(&mut self) -> Result<()> {
        self.inner.flush()
    }
}

impl<T: Write + Read + ?Sized> Read for BufferedWriter<T> {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        self.inner.get_mut().read(buf)
    }

    fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize> {
        self.inner.get_mut().read_vectored(bufs)
    }

    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        self.inner.get_mut().read_exact(buf)
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize> {
        self.inner.get_mut().read_to_end(buf)
    }

    fn read_to_string(&mut self, buf: &mut String) -> Result<usize> {
        self.inner.get_mut().read_to_string(buf)
    }
}

impl<T: Write + BufRead + ?Sized> BufRead for BufferedWriter<T> {
    fn fill_buf(&mut self) -> Result<&[u8]> {
        self.inner.get_mut().fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.inner.get_mut().consume(amt);
    }
}

impl<T: Write + BufRead + BufferCapacity> PeekableReadStream for BufferedWriter<T> {
    fn peek(&mut self, buf: &mut [u8]) -> std::result::Result<usize, PeekError> {
        if buf.len() > self.get_inner_ref().read_capacity() {
            return Err(PeekError::BiggerThanCapacity {
                requested: buf.len(),
                buffer_capacity: self.get_inner_ref().read_capacity(),
            });
        }

        let mut last_len = 0;
        while self.read_buffer().len() < buf.len() {
            self.inner.get_mut().fill_buf()?;
            let current_len = self.get_inner_ref().read_buffer().len();
            if last_len == current_len {
                break;
            }
            last_len = current_len;
        }

        let buffer = self.get_inner_ref().read_buffer();
        let ending = if buffer.len() < buf.len() {
            buffer.len()
        } else {
            buf.len()
        };

        for (index, elem) in buffer[0..ending].iter().enumerate() {
            buf[index] = *elem;
        }
        Ok(ending)
    }
}

pub type BufferedStream<T> = BufferedWriter<BufferedReader<T>>;

/// Returns a new `BufferedStream` type for `T` which is wrapping a `BufferedReader` and `BufferedWriter`
/// to support both read and write.
pub fn buffered_stream<T: Write + Read>(inner: T) -> BufferedStream<T> {
    BufferedWriter::new(BufferedReader::new(inner))
}

/// Returns a new `BufferedStream` type for `T` which is wrapping a `BufferedReader` and `BufferedWriter`
/// to support both read and write with a customized capacity.
pub fn buffered_stream_with_capacity<T: Write + Read>(
    capacity: usize,
    inner: T,
) -> BufferedStream<T> {
    BufferedWriter::with_capacity(capacity, BufferedReader::with_capacity(capacity, inner))
}

#[derive(From, Debug)]
pub enum PeekError {
    ZeroLengthNotAllowed,
    NotSupported,
    LockAcquisitionError,
    BiggerThanCapacity {
        requested: usize,
        buffer_capacity: usize,
    },
    IOError(std::io::Error),
}

impl std::error::Error for PeekError {}

impl core::fmt::Display for PeekError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{self:?}")
    }
}

/// Trait for read streams that support peeking ahead without consuming data.
///
/// # Errors
/// Returns an error if peeking fails or the requested size exceeds buffer capacity.
/// Trait for read streams that support peeking ahead without consuming data.
///
/// Implementors provide a way to inspect upcoming bytes from the stream without
/// advancing the stream's read cursor.
///
/// Methods on this trait may return a `PeekError` when the peek operation
/// fails (for example due to an I/O error) or when the requested size is not
/// supported by the underlying buffer.
pub trait PeekableReadStream: Read {
    /// Peeks data from the stream into `buf` without consuming it.
    ///
    /// Copies up to `buf.len()` bytes of upcoming data into `buf`. The returned
    /// value is the number of bytes actually written into `buf`. The stream's
    /// read position is not advanced by this operation.
    ///
    /// # Errors
    ///
    /// Returns `PeekError::BiggerThanCapacity` if the requested buffer length
    /// exceeds the read buffer capacity of the underlying stream.
    ///
    /// Returns `PeekError::IOError(e)` if an underlying I/O error occurs while
    /// attempting to fill or inspect the buffer.
    ///
    /// Other `PeekError` variants may be returned for implementation-specific
    /// error conditions (for example locking failures or unsupported operations).
    fn peek(&mut self, buf: &mut [u8]) -> std::result::Result<usize, PeekError>;
}

/// Trait for read streams that can be split into a separate connection.
pub trait SplitReadStream: Read + Sized {
    /// Split the connection into a separate handle.
    ///
    /// # Errors
    /// Returns an error if the stream cannot be split.
    fn split_connection(&self) -> std::io::Result<Self>;
}

impl<T: Read> PeekableReadStream for BufferedReader<T> {
    fn peek(&mut self, buf: &mut [u8]) -> std::result::Result<usize, PeekError> {
        if buf.len() > self.inner.capacity() {
            return Err(PeekError::BiggerThanCapacity {
                requested: buf.len(),
                buffer_capacity: self.inner.capacity(),
            });
        }

        let mut last_len = 0;
        while self.inner.buffer().len() < buf.len() {
            self.inner.fill_buf()?;
            let current_len = self.inner.buffer().len();
            if last_len == current_len {
                break;
            }
            last_len = current_len;
        }

        let buffer = self.inner.buffer();

        let ending = if buffer.len() < buf.len() {
            buffer.len()
        } else {
            buf.len()
        };

        for (index, elem) in buffer[0..ending].iter().enumerate() {
            buf[index] = *elem;
        }

        Ok(ending)
    }
}

pub enum OwnedReader<T: Read> {
    RefCell(Rc<RefCell<T>>),
    Atomic(Arc<AtomicPtr<T>>),
    Sync(Arc<Mutex<T>>),
    RWrite(Arc<RwLock<T>>),
}

impl<T: Read> OwnedReader<T> {
    pub fn rwrite(reader: Arc<RwLock<T>>) -> Self {
        Self::RWrite(reader)
    }

    pub fn sync(reader: Arc<Mutex<T>>) -> Self {
        Self::Sync(reader)
    }

    pub fn atomic(reader: &mut T) -> Self {
        Self::Atomic(Arc::new(AtomicPtr::new(reader)))
    }

    pub fn ref_cell(reader: Rc<RefCell<T>>) -> Self {
        Self::RefCell(reader)
    }
}

impl<T: Read> OwnedReader<T> {
    /// Executes a function with a reference to the underlying reader.
    ///
    /// # Panics
    /// Panics if the lock cannot be acquired.
    pub fn do_ref<F, V>(&self, caller: F) -> V
    where
        F: Fn(&T) -> V,
    {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    (caller)(atomic_reader)
                }
            }
            Self::Sync(core) => {
                let guard = core.lock().expect("can acquire");
                (caller)(&guard)
            }
            Self::RWrite(core) => {
                let guard = core.read().expect("can acquire");
                (caller)(&guard)
            }
            Self::RefCell(core) => {
                let guard = core.borrow();
                (caller)(&guard)
            }
        }
    }

    /// Executes a function once with a reference to the underlying reader.
    ///
    /// # Panics
    /// Panics if the lock cannot be acquired.
    pub fn do_once<F, V>(&self, caller: F) -> V
    where
        F: FnOnce(&T) -> V,
    {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    (caller)(atomic_reader)
                }
            }
            Self::Sync(core) => {
                let guard = core.lock().expect("can acquire");
                (caller)(&guard)
            }
            Self::RWrite(core) => {
                let guard = core.read().expect("can acquire");
                (caller)(&guard)
            }
            Self::RefCell(core) => {
                let guard = core.borrow();
                (caller)(&guard)
            }
        }
    }

    /// Executes a function once with a mutable reference to the underlying reader.
    ///
    /// # Panics
    /// Panics if the lock cannot be acquired.
    pub fn do_once_mut<F, V>(&self, caller: F) -> V
    where
        F: FnOnce(&mut T) -> V,
    {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    (caller)(atomic_reader)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                (caller)(&mut *guard)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                (caller)(&mut *guard)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                (caller)(&mut *guard)
            }
        }
    }

    /// Executes a mutable function with a mutable reference to the underlying reader.
    ///
    /// # Panics
    /// Panics if the lock cannot be acquired.
    pub fn do_mut<F, V>(&self, mut caller: F) -> V
    where
        F: FnMut(&mut T) -> V,
    {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    (caller)(atomic_reader)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                (caller)(&mut guard)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                (caller)(&mut guard)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                (caller)(&mut guard)
            }
        }
    }
}

impl<T: Read> Clone for OwnedReader<T> {
    fn clone(&self) -> Self {
        match self {
            Self::Atomic(core) => Self::Atomic(Arc::clone(core)),
            Self::RWrite(core) => Self::RWrite(Arc::clone(core)),
            Self::RefCell(core) => Self::RefCell(core.clone()),
            Self::Sync(core) => Self::Sync(core.clone()),
        }
    }
}

impl<T: Read + Write> Write for OwnedReader<T> {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.write(buf)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.write(buf)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.write(buf)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.write(buf)
            }
        }
    }

    fn flush(&mut self) -> Result<()> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.flush()
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.flush()
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.flush()
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.flush()
            }
        }
    }
}

impl<T: ReadTimeoutOperations> ReadTimeoutOperations for OwnedReader<T> {
    fn read_timeout_into(
        &mut self,
        buf: &mut [u8],
        timeout: std::time::Duration,
    ) -> std::result::Result<usize, std::io::Error> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.read_timeout_into(buf, timeout)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.read_timeout_into(buf, timeout)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.read_timeout_into(buf, timeout)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.read_timeout_into(buf, timeout)
            }
        }
    }

    fn set_read_timeout_as(
        &mut self,
        timeout: std::time::Duration,
    ) -> std::result::Result<(), std::io::Error> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.set_read_timeout_as(timeout)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.set_read_timeout_as(timeout)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.set_read_timeout_as(timeout)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.set_read_timeout_as(timeout)
            }
        }
    }

    fn get_current_read_timeout(
        &self,
    ) -> std::result::Result<Option<std::time::Duration>, std::io::Error> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.get_current_read_timeout()
                }
            }
            Self::Sync(core) => {
                let guard = core.lock().expect("can acquire");
                guard.get_current_read_timeout()
            }
            Self::RWrite(core) => {
                let guard = core.write().expect("can acquire");
                guard.get_current_read_timeout()
            }
            Self::RefCell(core) => {
                let guard = core.borrow_mut();
                guard.get_current_read_timeout()
            }
        }
    }
}

impl<T: Read> Read for OwnedReader<T> {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.read(buf)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.read(buf)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.read(buf)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.read(buf)
            }
        }
    }

    fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.read_vectored(bufs)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.read_vectored(bufs)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.read_vectored(bufs)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.read_vectored(bufs)
            }
        }
    }

    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.read_exact(buf)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.read_exact(buf)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.read_exact(buf)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.read_exact(buf)
            }
        }
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.read_to_end(buf)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.read_to_end(buf)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.read_to_end(buf)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.read_to_end(buf)
            }
        }
    }

    fn read_to_string(&mut self, buf: &mut String) -> Result<usize> {
        match self {
            Self::Atomic(core) => {
                let ptr = core.load(std::sync::atomic::Ordering::Acquire);
                unsafe {
                    let atomic_reader: &mut T = &mut *ptr;
                    atomic_reader.read_to_string(buf)
                }
            }
            Self::Sync(core) => {
                let mut guard = core.lock().expect("can acquire");
                guard.read_to_string(buf)
            }
            Self::RWrite(core) => {
                let mut guard = core.write().expect("can acquire");
                guard.read_to_string(buf)
            }
            Self::RefCell(core) => {
                let mut guard = core.borrow_mut();
                guard.read_to_string(buf)
            }
        }
    }
}

// Bare metal platforms usually have very small amounts of RAM
// (in the order of hundreds of KB)
pub const DEFAULT_READ_SIZE: usize = if cfg!(target_os = "espidf") {
    512
} else {
    8 * 1024
};

/// `SharedPointerReader` defines a shared buffer reader pointer that allows reading through
/// a underlying buffered stream.

/// Marker trait for transport types that can report whether they are
/// Unix-domain sockets (and thus non-poolable after a single HTTP exchange).
pub trait IsUnixTransport {
    /// Whether the underlying transport is a Unix-domain socket.
    fn is_unix(&self) -> bool;

    /// Whether an HTTP connection over this transport may be returned to the
    /// pool for reuse.
    ///
    /// WHY: Unix sockets close after every response, so they are never pooled —
    /// the default reflects that. Transports that carry their own poolability
    /// decision (e.g. a caller-provided stream: a `WireGuard` overlay pools, an
    /// SSH `docker system dial-stdio` channel does not) override this.
    fn should_pool(&self) -> bool {
        !self.is_unix()
    }
}

pub struct SharedByteBufferStream<T: Read>(OwnedReader<ByteBufferPointer<T>>);

impl<T: Read> core::fmt::Debug for SharedByteBufferStream<T> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        // Use the internal OwnedReader to borrow the inner ByteBufferPointer
        // and produce a concise debug representation without requiring `T: Debug`.
        self.0.do_once_mut(|inner| {
            f.debug_struct("SharedByteBufferStream")
                .field("pull_amount", &inner.pull_amount)
                .field("buffer_len", &inner.buffer.len())
                .field("pos", &inner.pos)
                .field("peek_pos", &inner.peek_pos)
                .field("distance", &inner.distance())
                .finish()
        })
    }
}

impl<T: Read> SharedByteBufferStream<T> {
    pub fn do_ref<F, V>(&self, caller: F) -> V
    where
        F: Fn(&ByteBufferPointer<T>) -> V,
    {
        self.0.do_ref(caller)
    }

    pub fn do_once<F, V>(&self, caller: F) -> V
    where
        F: FnOnce(&ByteBufferPointer<T>) -> V,
    {
        self.0.do_once(caller)
    }

    pub fn do_once_mut<F, V>(&self, caller: F) -> V
    where
        F: FnOnce(&mut ByteBufferPointer<T>) -> V,
    {
        self.0.do_once_mut(caller)
    }

    pub fn do_mut<F, V>(&self, caller: F) -> V
    where
        F: FnMut(&mut ByteBufferPointer<T>) -> V,
    {
        self.0.do_mut(caller)
    }

    /// Read a line from the stream into the provided buffer.
    ///
    /// Delegates to the inner `ByteBufferPointer`'s `read_line` method.
    /// Returns the number of bytes read, or an error if reading fails.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying stream fails.
    pub fn read_line(&mut self, buf: &mut String) -> std::io::Result<usize> {
        self.0.do_once_mut(|inner| inner.read_line(buf))
    }

    /// Probe the stream by filling the buffer if needed and returning the available byte count.
    ///
    /// If the buffer already has unconsumed data, returns its length without reading.
    /// Otherwise calls `fill_up()` to try pulling more bytes from the underlying reader.
    ///
    /// # Errors
    /// Returns the underlying `io::Error` if the fill fails (e.g. WouldBlock,
    /// connection reset, etc.).
    pub fn probe(&mut self) -> std::io::Result<usize> {
        self.0.do_once_mut(|inner| {
            if inner.is_empty_or_consumed() {
                let _ = inner.fill_up()?;
            }
            Ok(inner.remaining())
        })
    }

    /// Whether the underlying transport is a Unix-domain socket.
    ///
    /// Delegates through `ByteBufferPointer::with_inner` →
    /// `RawStream::is_unix()` → `Connection::is_unix()`.
    #[must_use]
    pub fn is_unix(&self) -> bool
    where
        T: IsUnixTransport,
    {
        self.0.do_ref(|bbp| bbp.with_inner(|raw| raw.is_unix()))
    }

    /// Whether an HTTP connection over this stream may be returned to the pool.
    ///
    /// Delegates through `ByteBufferPointer::with_inner` →
    /// `RawStream::should_pool()` → `Connection::should_pool()`.
    #[must_use]
    pub fn should_pool(&self) -> bool
    where
        T: IsUnixTransport,
    {
        self.0.do_ref(|bbp| bbp.with_inner(|raw| raw.should_pool()))
    }

    /// Run `f` against the innermost reader `T` by shared reference.
    ///
    /// WHY: callers above `foundation_core` (e.g. `foundation_netio`) sometimes need
    /// to reach the concrete transport — for instance to register a readiness waker
    /// on a non-blocking overlay socket — without the buffered wrappers in the way.
    /// Same access path as [`Self::is_unix`], generalised over the return type.
    ///
    /// The closure runs under the stream's lock; keep it short and non-reentrant.
    pub fn with_inner_ref<V>(&self, f: impl Fn(&T) -> V) -> V {
        self.0.do_ref(|bbp| bbp.with_inner(|raw| f(raw)))
    }
}

// Implement cloning for the [`SharedByteBufferStream`].
impl<T: Read> Clone for SharedByteBufferStream<T> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

// Constructors

impl<T: Read> SharedByteBufferStream<T> {
    pub fn ref_cell(reader: T) -> Self {
        let wrapped_reader = OwnedReader::ref_cell(Rc::new(RefCell::new(reader)));
        let byte_reader = ByteBufferPointer::new(DEFAULT_READ_SIZE, wrapped_reader);
        Self(OwnedReader::ref_cell(Rc::new(RefCell::new(byte_reader))))
    }

    pub fn ref_cell_with_capacity(capacity: usize, reader: T) -> Self {
        let wrapped_reader = OwnedReader::ref_cell(Rc::new(RefCell::new(reader)));
        let byte_reader = ByteBufferPointer::new(capacity, wrapped_reader);
        Self(OwnedReader::ref_cell(Rc::new(RefCell::new(byte_reader))))
    }

    pub fn rwrite(reader: T) -> Self {
        let wrapped_reader = OwnedReader::rwrite(Arc::new(RwLock::new(reader)));
        let byte_reader = ByteBufferPointer::new(DEFAULT_READ_SIZE, wrapped_reader);
        Self(OwnedReader::rwrite(Arc::new(RwLock::new(byte_reader))))
    }

    pub fn rwrite_with_capacity(capacity: usize, reader: T) -> Self {
        let wrapped_reader = OwnedReader::rwrite(Arc::new(RwLock::new(reader)));
        let byte_reader = ByteBufferPointer::new(capacity, wrapped_reader);
        Self(OwnedReader::rwrite(Arc::new(RwLock::new(byte_reader))))
    }

    pub fn sync(reader: &Arc<Mutex<T>>) -> Self {
        let wrapped_reader = OwnedReader::sync(Arc::clone(reader));
        let byte_reader = ByteBufferPointer::new(DEFAULT_READ_SIZE, wrapped_reader);
        Self(OwnedReader::sync(Arc::new(Mutex::new(byte_reader))))
    }
}

impl<T: Read> SharedByteBufferStream<BufferedReader<T>> {
    pub fn with_owned_reader(reader: T) -> Self {
        let wrapped_reader = BufferedReader::new(reader);
        Self::rwrite(wrapped_reader)
    }
}

impl<T: Read> SharedByteBufferStream<BufferedReader<T>> {
    pub fn with_buffered_owned_reader(reader: T) -> Self {
        Self::rwrite(BufferedReader::new(reader))
    }
}

// Public Methods

impl<T: Read> PeekableReadStream for SharedByteBufferStream<T> {
    fn peek(&mut self, buf: &mut [u8]) -> std::result::Result<usize, PeekError> {
        self.0
            .do_once_mut(|binding| match binding.peekby(buf.len()) {
                Ok(state) => match state {
                    // `Stalled` carries real bytes too — the peer simply has not
                    // finished speaking. A peeker wants to look at whatever is
                    // there (a whole short request may already have arrived), so
                    // both states hand their data over.
                    PeekState::Request(data) | PeekState::Stalled(data) => {
                        let ending = if buf.len() > data.len() {
                            data.len()
                        } else {
                            buf.len()
                        };

                        for (index, elem) in data[0..ending].iter().enumerate() {
                            buf[index] = *elem;
                        }
                        Ok(data.len())
                    }
                    // No bytes are currently peekable. `NoNext` in particular is
                    // reached when the reader is drained and reported EOF — a
                    // peer that connects and immediately disconnects hits it, so
                    // it is an ordinary end-of-stream, not an unreachable state.
                    PeekState::ZeroLengthInput
                    | PeekState::NoNext
                    | PeekState::LessThanRequested
                    | PeekState::EndOfBuffered
                    | PeekState::EndOfFile
                    | PeekState::Continue => Ok(0),
                },
                Err(err) => Err(PeekError::IOError(err)),
            })
    }
}

impl<T: Read + Write> std::io::Write for SharedByteBufferStream<T> {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        self.0.do_once_mut(|binding| binding.write(buf))
    }

    fn flush(&mut self) -> Result<()> {
        self.0.do_once_mut(Write::flush)
    }
}

impl<T: ReadTimeoutOperations> ReadTimeoutOperations for SharedByteBufferStream<T> {
    fn read_timeout_into(
        &mut self,
        buf: &mut [u8],
        timeout: std::time::Duration,
    ) -> std::result::Result<usize, std::io::Error> {
        self.0
            .do_once_mut(|binding| binding.read_timeout_into(buf, timeout))
    }

    fn set_read_timeout_as(
        &mut self,
        timeout: std::time::Duration,
    ) -> std::result::Result<(), std::io::Error> {
        self.0
            .do_once_mut(|binding| binding.set_read_timeout_as(timeout))
    }

    fn get_current_read_timeout(
        &self,
    ) -> std::result::Result<Option<std::time::Duration>, std::io::Error> {
        self.0
            .do_once_mut(|binding| binding.get_current_read_timeout())
    }
}

impl<T: Read> std::io::Read for SharedByteBufferStream<T> {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        self.0.do_once_mut(|binding| binding.read(buf))
    }

    fn read_vectored(&mut self, buf: &mut [IoSliceMut<'_>]) -> Result<usize> {
        self.0.do_once_mut(|binding| binding.read_vectored(buf))
    }

    /// All-or-nothing `read_exact`.
    ///
    /// WHY: the standard `read_exact` loops over `read`, copying what it gets
    /// into `buf`. If a later `read` fails — which on a non-blocking stream means
    /// `WouldBlock` — it returns `Err` and the caller drops `buf`. But those
    /// bytes were already *consumed* from the stream, so a retry silently
    /// resumes mid-message. That corrupts any framed protocol (it is exactly how
    /// an h2 handshake ends up validating a SETTINGS frame as its preface).
    ///
    /// HOW: buffer the whole request first via `peekby`, which consumes nothing.
    /// Only once all `buf.len()` bytes are in memory do we consume them, so this
    /// call either fully succeeds or leaves the stream untouched.
    ///
    /// The two ways of coming up short are kept strictly apart, because callers
    /// act on them very differently:
    ///
    /// - **Transient** (the peer is just slow): `peekby` reports `Stalled`, or
    ///   propagates `WouldBlock` when it buffered nothing at all. Either way the
    ///   caller sees `WouldBlock` and retries. Nothing was consumed, so the retry
    ///   sees the message from the start.
    /// - **Terminal** (the peer closed mid-message): `peekby` returns a short
    ///   `Request`, which it only does by breaking on `fill_up() == 0`, and
    ///   `Read::read` returning `0` *is* end-of-stream. No further bytes will ever
    ///   arrive, so this reports `UnexpectedEof` as `read_exact` must. Reporting
    ///   `WouldBlock` here would invite the caller to retry forever.
    ///
    /// Blocking readers are unaffected: `peekby` fills until satisfied or EOF.
    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        if buf.is_empty() {
            return Ok(());
        }
        self.0.do_once_mut(|binding| {
            let needed = buf.len();
            match binding.peekby(needed) {
                // Fully buffered: the consuming read below cannot now stall.
                Ok(PeekState::Request(data)) if data.len() >= needed => binding.read_exact(buf),

                // Short because the peer paused, not because it left. More bytes
                // are still coming, so this is a retry, not a failure.
                Ok(PeekState::Stalled(_)) => Err(crate::err!(
                    WouldBlock,
                    "only part of the {needed} bytes have arrived; retry once the peer sends more"
                )),

                // Short, or nothing at all. With `Stalled` handled above, `peekby`
                // only returns `Ok` here after the reader signalled EOF (see the
                // invariant above), so the stream has ended mid-message.
                Ok(_) => Err(crate::err!(
                    UnexpectedEof,
                    "failed to fill whole buffer: stream ended before {needed} bytes arrived"
                )),

                // Reader error — `WouldBlock` included — passed through verbatim.
                // Nothing has been consumed.
                Err(err) => Err(err),
            }
        })
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize> {
        self.0.do_once_mut(|binding| binding.read_to_end(buf))
    }

    fn read_to_string(&mut self, buf: &mut String) -> Result<usize> {
        self.0.do_once_mut(|binding| binding.read_to_string(buf))
    }
}

// SAFETY: SharedByteBufferStream is Send when T is Send AND it uses Arc-based synchronization.
//
// The HTTP client creates SharedByteBufferStream via rwrite() which uses Arc<RwLock<T>>.
// Arc<RwLock<T>> is Send when T: Send. The RefCell variant (Rc<RefCell<T>>) is only
// used in single-threaded contexts and never crosses thread boundaries.
//
// In HttpRequestTask:
//   - stream_holder stores SharedByteBufferStream<RawStream>
//   - Created via SharedByteBufferStream::rwrite() (task.rs:194)
//   - RawStream is Send, RWrite uses Arc<RwLock<_>>
//   - Therefore safe to send across threads
unsafe impl<T: Read + Send> Send for SharedByteBufferStream<T> {}

// SAFETY: SharedByteBufferStream is Sync when T is Send + Sync AND it uses
// Arc-based synchronization. The multi-threaded (`multi` feature) path always
// uses rwrite() (Arc<RwLock<T>>) or sync() (Arc<Mutex<T>>) — both are Sync
// when T: Send + Sync. The RefCell variant (Rc<RefCell<T>>) is only used on
// single-threaded paths and never shared across threads.
//
// See Feature 00 — SendSafeBody Sync for motivation.
unsafe impl<T: Read + Send + Sync> Sync for SharedByteBufferStream<T> {}

pub struct ByteBufferPointer<T: Read> {
    reader: OwnedReader<T>,
    pull_amount: usize,
    peek_pos: usize,
    buffer: Vec<u8>,
    pos: usize,
}

// Constructors

impl<T: Read> ByteBufferPointer<T> {
    #[must_use]
    pub fn new(pull_amount: usize, reader: OwnedReader<T>) -> Self {
        Self {
            buffer: Vec::with_capacity(pull_amount),
            pull_amount,
            peek_pos: 0,
            reader,
            pos: 0,
        }
    }

    pub fn from_reader(pull_amount: usize, reader: T) -> Self {
        let wrapped_reader = OwnedReader::rwrite(Arc::new(RwLock::new(reader)));
        Self::new(pull_amount, wrapped_reader)
    }

    pub fn reader(reader: T) -> Self {
        let wrapped_reader = OwnedReader::rwrite(Arc::new(RwLock::new(reader)));
        Self::new(DEFAULT_READ_SIZE, wrapped_reader)
    }

    /// Access the inner reader for transport-level queries like `is_unix()`.
    pub fn with_inner<F, V>(&self, f: F) -> V
    where
        F: Fn(&T) -> V,
    {
        self.reader.do_ref(|inner| f(inner))
    }
}

impl<T: ReadTimeoutOperations> ReadTimeoutOperations for ByteBufferPointer<T> {
    fn read_timeout_into(
        &mut self,
        buf: &mut [u8],
        timeout: std::time::Duration,
    ) -> std::result::Result<usize, std::io::Error> {
        self.reader.read_timeout_into(buf, timeout)
    }

    fn set_read_timeout_as(
        &mut self,
        timeout: std::time::Duration,
    ) -> std::result::Result<(), std::io::Error> {
        self.reader.set_read_timeout_as(timeout)
    }

    fn get_current_read_timeout(
        &self,
    ) -> std::result::Result<Option<std::time::Duration>, std::io::Error> {
        self.reader.get_current_read_timeout()
    }
}

// Methods

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Data<'a> {
    Request(&'a [u8]), // data you resulted
    Consumed(Vec<u8>),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PeekState<'a> {
    Request(&'a [u8]), // data you resulted
    /// Fewer bytes than requested, because the non-blocking reader stalled
    /// (`WouldBlock`) rather than ended. The bytes are real and peekable, but
    /// more may still arrive.
    ///
    /// This is deliberately distinct from a short [`Self::Request`], which means
    /// the stream *ended* early: a caller that needs every byte must retry on
    /// `Stalled` and fail on a short `Request`. Collapsing the two makes a slow
    /// peer indistinguishable from a hung-up one.
    Stalled(&'a [u8]),
    LessThanRequested, // when data is way less than requested position
    EndOfBuffered,     // end of buffered data, so consume and read more
    EndOfFile,         // end of file, the real underlying stream is finished
    NoNext, // indicates the peek cursor is still at the same position as the data cursor.
    Continue,
    ZeroLengthInput, // indicates you can continue to peek into the buffer
}

#[allow(unused)]
impl<T: Read> ByteBufferPointer<T> {
    /// Returns the distance between the peek position and the actual cursor
    /// position.
    #[inline]
    #[must_use]
    pub fn distance(&self) -> usize {
        self.peek_pos - self.pos
    }

    #[must_use]
    pub fn peek_cursor(&self) -> usize {
        self.peek_pos
    }

    #[must_use]
    pub fn data_cursor(&self) -> usize {
        self.pos
    }

    /// Returns the total length of the string being accumulated on.
    #[inline]
    #[must_use]
    pub fn len(&self) -> usize {
        self.buffer.len()
    }

    #[inline]
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.buffer.len() == 0
    }

    /// Returns the number of unconsumed bytes in the buffer.
    #[inline]
    #[must_use]
    pub fn remaining(&self) -> usize {
        self.buffer.len().saturating_sub(self.pos)
    }

    /// Returns true when the buffer has no data or all data has been consumed.
    #[inline]
    #[must_use]
    pub fn is_empty_or_consumed(&self) -> bool {
        self.remaining() == 0
    }

    #[inline]
    pub fn uncapture(&mut self) {
        self.uncapture_by(1);
    }

    #[inline]
    pub fn uncapture_by(&mut self, by: usize) {
        if (self.pos - by) > 0 {
            self.pos -= by;
        } else {
            self.pos = 0;
        }
    }

    /// `full_scan` returns the whole buffer as is, so you see the entire
    /// content regardless of cursors position.
    #[inline]
    #[must_use]
    pub fn full_scan(&self) -> &[u8] {
        &self.buffer[..]
    }

    /// scan returns the whole string slice currently at the points of where
    /// the main pos (position) cursor till the end.
    #[inline]
    #[must_use]
    pub fn scan(&self) -> &[u8] {
        &self.buffer[self.pos..self.peek_pos]
    }

    #[inline]
    #[must_use]
    pub fn greater_than_40_percent(&self) -> bool {
        // if we have not moved at all the ignore
        if self.pos == 0 {
            return false;
        }

        let buffer_length = self.buffer.len();
        #[allow(clippy::cast_precision_loss)]
        let percentage = (buffer_length as f64 / self.pos as f64);
        percentage > 0.4
    }

    #[inline]
    pub fn truncate(&mut self, force: bool) {
        // if we have not moved at all the ignore
        if self.pos == 0 {
            return;
        }

        let distance = self.peek_pos - self.pos;
        let buffer_length = self.buffer.len();
        #[allow(clippy::cast_precision_loss)]
        let percentage = (buffer_length as f64 / self.pos as f64);
        let should_truncate = percentage > 0.7 || force;

        if !should_truncate {
            return;
        }

        let slice = &self.buffer[self.pos..buffer_length];
        let mut slice_copy = vec![0; slice.len()];
        slice_copy.truncate(0);
        slice_copy.extend_from_slice(slice);
        let slice_length = slice_copy.len();

        // move the slice to the start of the buffer.
        for (index, byte) in slice_copy.iter().enumerate() {
            self.buffer[index] = *byte;
        }

        self.buffer.truncate(slice_length);
        self.pos = 0;
        self.peek_pos = self.pos + distance;
    }

    /// [`Self::fill_up`] fills the internal buffer with the pull amount
    /// which allows you to continue to collect the relevant
    /// set of data which we match is right in the correct set of
    /// bytes until we indicate to the pointer to consume the data until
    /// the position cursor.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    #[inline]
    pub fn fill_up(&mut self) -> std::io::Result<usize> {
        // truncate buffer if we have read most of it.
        let force = self.greater_than_40_percent();
        self.truncate(force);

        let len = self.buffer.len();
        // tracing::trace!("Current buffer: {} -> {:?}", len, &self.buffer[..len.min(64)]);

        // extract and add more to buffer from reader.
        // self.reader.fill_buf()?;
        let mut copied = vec![0; self.pull_amount];
        let read = match self.reader.read(&mut copied) {
            Ok(read_size) => read_size,
            Err(err) => {
                tracing::trace!("Failed to read data from reader due to: {:?}", &err);
                return Err(err);
            }
        };
        // tracing::trace!("Copied buffer: {:?} -> {:?}", read, &copied[..read]);

        // copy into the buffer the data just extracted from the buffer.
        // let location_before_extend = self.buffer.len();
        self.buffer.extend_from_slice(&copied[0..read]);
        // tracing::trace!(
        //     "Extended buffer: {:?} -> {:?}",
        //     self.buffer.len(),
        //     &self.buffer[0..self.buffer.len()],
        // );

        Ok(read)
    }

    /// [`Self::fill_all`] reads the whole underlying reader into the underlying
    /// buffer, allowing you to extract the remaining data within the stream
    /// as fully in-memory.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    #[inline]
    pub fn fill_all(&mut self, read_limit: Option<usize>) -> std::io::Result<usize> {
        let mut total_read = 0;

        // pull the amount of data until we reach EOF
        let mut copied = vec![0; self.pull_amount];
        loop {
            let read = self.reader.read(&mut copied)?;

            if read == 0 {
                break;
            }

            total_read += read;

            if let Some(limited_val) = read_limit {
                if limited_val < total_read {
                    return Err(crate::err!(
                        Interrupted,
                        "LimitReadTriggered",
                        "Crossed read limit {}",
                        limited_val,
                    ));
                }
            }

            self.buffer.extend_from_slice(&copied[0..read]);
        }

        Ok(total_read)
    }

    /// Forward by 1 by calling the [`Self::forward_by`] method underneath.
    ///
    /// # Errors
    /// Returns an error if forwarding fails.
    #[inline]
    pub fn forward(&mut self) -> std::io::Result<PeekState<'_>> {
        self.forward_by(1)
    }

    /// [`Self::forward_by`] provides method to move the peek cursor by a certain amount.
    /// Generally this is used external as the logic is generally backed into the
    /// `next*` and `read*` methods but in cases where you intend to progress the
    /// cursor your seek using the `peek*` methods this provides that surface.
    ///
    /// # Errors
    /// Returns an error if forwarding fails.
    #[inline]
    pub fn forward_by(&mut self, by: usize) -> std::io::Result<PeekState<'_>> {
        let buffer_length = self.buffer.len();
        let new_peek = self.peek_pos + by;
        if new_peek > buffer_length {
            self.peek_pos = buffer_length;
            Ok(PeekState::EndOfBuffered)
        } else {
            self.peek_pos = new_peek;
            Ok(PeekState::Continue)
        }
    }

    /// Unforward by 1 by calling the [`Self::unforward_by`] method underneath.
    ///
    /// # Errors
    /// Returns an error if unforwarding fails.
    #[inline]
    pub fn unforward(&mut self) -> std::io::Result<PeekState<'_>> {
        self.unforward_by(1)
    }

    /// [`Self::unforward_by`] moves your peek position backwards at which if it moves
    /// all the way back, will forever stay at the last know position of
    /// the actual data cursor.
    ///
    /// # Errors
    /// Returns an error if unforwarding fails.
    #[inline]
    pub fn unforward_by(&mut self, by: usize) -> std::io::Result<PeekState<'_>> {
        let new_peek = self.peek_pos - by;
        if new_peek <= self.pos {
            self.peek_pos = self.pos;
            Ok(PeekState::Continue)
        } else {
            self.peek_pos = new_peek;
            Ok(PeekState::Continue)
        }
    }

    /// [`Self::consume_some`] consumes the amount of data that has been peeked over-so far, returning
    /// that to the caller, this also moves the position of the data cursor
    /// forward to the location of the skip cursor.
    pub fn consume_some(&mut self) -> Option<Vec<u8>> {
        let from = self.pos;
        let until_pos = self.peek_pos;
        if from == until_pos {
            return None;
        }
        let slice = &self.buffer[from..until_pos];
        let slice_string = String::from_utf8(slice.to_vec());
        self.pos = self.peek_pos;

        let mut slice_copy = vec![0; slice.len()];
        slice_copy.truncate(0);
        slice_copy.extend_from_slice(slice);

        Some(slice_copy)
    }

    /// [`Self::consume`] consumes the amount of data that has been peeked over-so far, returning
    /// that to the caller, this also moves the position of the data cursor
    /// forward to the location of the skip cursor.
    ///
    /// # Errors
    /// Returns an error if consuming fails.
    pub fn consume(&mut self) -> std::io::Result<Vec<u8>> {
        match self.consume_some() {
            Some(item) => Ok(item),
            None => Ok(vec![]),
        }
    }

    /// [`Self::skip`] skip the amount of data that has been peeked over-so far, returning
    /// that to the caller, this also moves the position of the data cursor
    /// forward to the location of the skip cursor.
    pub fn skip(&mut self) {
        let from = self.pos;
        let until_pos = self.peek_pos;
        if from == until_pos {
            return;
        }
        self.pos = self.peek_pos;
    }

    /// [`Self::peek`] peeks into the future by 1 position without actually permanently
    /// change the peek cursor's position.
    ///
    /// # Errors
    /// Returns an error if peeking fails.
    #[inline]
    pub fn peek(&self) -> std::io::Result<PeekState<'_>> {
        self.peek_by(1)
    }

    /// [`Self::peek_by`] returns the available data if there is available within the forward movement
    /// being requested, the peek cursor is never adjusted but only used to look forward.
    ///
    /// WARNING: Do not use this method and then call [`Self::consume`] has it has no effect
    // or you may consume far less than intended. This is intended to let you peek forward
    /// without actual changes to peek cursor which is used in consume to extract the data
    /// already seen by calling all `next_*` methods.
    ///
    /// # Errors
    /// Returns an error if peeking fails.
    #[inline]
    pub fn peek_by(&self, by: usize) -> std::io::Result<PeekState<'_>> {
        let until_pos = self.peek_pos + by;

        // if we are further than the current buffer and the actual content of
        // the reading buffer, then indicate we are beyond available data which
        // requires user to fill up the buffer.
        let buffer_length = self.buffer.len();

        // data is less than requested
        if until_pos > buffer_length {
            return Ok(PeekState::LessThanRequested);
        }

        let from = self.peek_pos;
        Ok(PeekState::Request(&self.buffer[from..until_pos]))
    }

    // [`Self::peek_until`] returns the peek state for the requested data until the delimiter is seen
    /// at which point the underlying reference to the data is either shared in a
    /// [`PeekState::Request`] if fully read else return [`PeekState::EndOfFile`] if the
    /// source returns EOF or if the data available is less than requested.
    ///
    /// It runs in a tight loop and ensures to acquire as much data as needed.
    ///
    /// It catches the initial position of the peek cursor then returns back to that
    /// position after stopping. So the peek cursor will not adjust though the internal
    /// buffer may have been filled up using memory attempting to find the giving signal
    /// and this might read the underlying reader until its EOF or OOM occurs if the signal
    /// is never found.
    ///
    /// WARNING: Do not use this method and then call [`Self::consume`] has it has no effect
    /// or you may consume far less than intended. This is intended to let you peek forward
    /// without actual changes to peek cursor which is used in consume to extract the data
    /// already seen by calling all `next_*` methods.
    ///
    /// # Errors
    /// Returns an error if peeking or reading from the underlying reader fails.
    #[inline]
    pub fn peek_until<'a>(&'a mut self, signal: &[u8]) -> std::io::Result<PeekState<'a>> {
        if signal.is_empty() {
            return Ok(PeekState::ZeroLengthInput);
        }
        let current_peek_pos = self.peek_pos;
        loop {
            let state = self.peek_by(signal.len())?;
            match state {
                PeekState::Request(data) => {
                    if data == signal {
                        break;
                    }
                }
                PeekState::EndOfBuffered | PeekState::LessThanRequested => {
                    // if at the end then return with EndOfFile
                    if self.fill_up()? == 0 {
                        return Ok(PeekState::EndOfFile);
                    }
                    continue;
                }
                PeekState::Continue => continue,
                PeekState::EndOfFile => {
                    return Ok(PeekState::EndOfFile);
                }
                _ => {
                    unreachable!("Should never hit this types")
                }
            }

            self.peek_pos += signal.len();
        }

        // account for the found block where we break and ensure we can capture
        // the last peek correctly.
        self.peek_pos += signal.len();

        let slice = &self.buffer[self.pos..self.peek_pos];

        // restores peek position
        self.peek_pos = current_peek_pos;

        Ok(PeekState::Request(slice))
    }

    /// [`Self::peekby2`] provides a friendly method which calls [`Self::peekby`] underneath
    /// to peek into the distance without modifying cursor position.
    ///
    /// This means the cursor is at that position it was after peek into the need size from
    /// the current position.
    ///
    /// WARNING: Do not use this method and then call [`Self::consume`] has it has no effect
    /// or you may consume far less than intended. This is intended to let you peek forward
    /// without actual changes to peek cursor which is used in consume to extract the data
    /// already seen by calling all `next_*` methods.
    ///
    /// # Errors
    /// Returns an error if peeking fails or if there's no data available.
    pub fn peekby2(&mut self, size: usize) -> std::io::Result<&[u8]> {
        self.peekby(size).map(|item| match item {
            // A stall still yields the bytes it holds; this returns what is
            // peekable now, and short results are already part of its contract.
            PeekState::Request(inner) | PeekState::Stalled(inner) => Ok(inner),
            PeekState::NoNext => Err(crate::err!(UnexpectedEof, "No more data to pull through")),
            PeekState::ZeroLengthInput => Err(crate::err!(WriteZero, "Provided zero size request")),
            _ => unreachable!("Should not trigger this stage"),
        })?
    }

    /// [`Self::peekby`] returns a portion of the underlying buffer for the specified
    /// size using a tight loop until the requested size is of data has being pulled
    /// into the internal peek buffer for peeking .
    ///
    /// This moves forward the peek cursor forward temporarily until the requested size is achieved
    /// and if the loop stops and the current buffer size does not match then
    /// we return what's already acquired, so you need to be aware that this can happen
    /// since generally it means we have reached EOF from the internal readers perspective —
    /// or, on a non-blocking reader, that the peer has paused mid-message.
    ///
    /// WARNING: Do not use this method and then call [`Self::consume`] has it has no effect
    /// or you may consume far less than intended. This is intended to let you peek forward
    /// without actual changes to peek cursor which is used in consume to extract the data
    /// already seen by calling all `next_*` methods.
    ///
    /// # Errors
    /// Returns an error if peeking or reading from the underlying reader fails.
    /// `WouldBlock` is returned only when the stall leaves *nothing* buffered; a
    /// stall on top of buffered bytes yields those bytes as a short
    /// [`PeekState::Request`] instead.
    #[inline]
    pub fn peekby(&mut self, size: usize) -> std::io::Result<PeekState<'_>> {
        if size == 0 {
            return Ok(PeekState::ZeroLengthInput);
        }

        // Set when the fill loop stopped because the reader stalled rather than
        // ended, so the short result below can be labelled `Stalled` not `Request`.
        let mut stalled = false;

        loop {
            // Check remaining unconsumed data from peek_pos, not total buffer length.
            let available = self.buffer.len().cast_signed() - self.peek_pos.cast_signed();
            let rem = size.cast_signed() - available;

            if rem <= 0 {
                break;
            }

            // request more data so we get to enough to actually resolve the
            // requested size.
            match self.fill_up() {
                // EOF: the peer will send no more, so hand back what we have.
                Ok(0) => break,
                Ok(_) => {}
                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
                    // A non-blocking reader with nothing *more* to give yet. Bytes
                    // already buffered are still perfectly good to peek at, and
                    // discarding them by propagating `WouldBlock` strands any peer
                    // whose whole message is shorter than `size` — the caller would
                    // ask again, stall again, and never see the bytes sitting in
                    // the buffer.
                    //
                    // Only a stall with *nothing* buffered is a pure "come back
                    // later"; that still surfaces `WouldBlock` so callers can park.
                    if available > 0 {
                        stalled = true;
                        break;
                    }
                    return Err(err);
                }
                Err(err) => return Err(err),
            }
        }

        let buffer_len = self.buffer.len();

        let mut until_pos = self.peek_pos + size;
        if until_pos > buffer_len {
            until_pos = buffer_len;
        }

        let slice = &self.buffer[self.peek_pos..until_pos];
        if slice.is_empty() {
            return Ok(PeekState::NoNext);
        }

        if stalled && slice.len() < size {
            return Ok(PeekState::Stalled(slice));
        }

        Ok(PeekState::Request(slice))
    }

    /// [`Self::nextby`] returns a portion of the underlying buffer for the specified
    /// size using a tight loop until the requested size is of data has being pulled
    /// into the internal peek buffer for peeking .
    ///
    /// This moves forward the cursor forward until the requested size is achieved
    /// and if the loop stops and the current buffer size does not match then
    /// we return what's already acquired, so you need to be aware that this can happen
    /// since generally it means we have reached EOF from the internal readers perspective.
    ///
    /// This moves the peek cursor forward.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    #[inline]
    pub fn nextby(&mut self, size: usize) -> std::io::Result<PeekState<'_>> {
        if size == 0 {
            return Ok(PeekState::ZeroLengthInput);
        }

        let mut cached_error: Option<std::io::Error> = None;

        loop {
            // Check remaining unconsumed data (from peek_pos to end of buffer),
            // not total buffer length. After prior reads consume data, peek_pos
            // advances but buffer.len() stays the same until truncation. We must
            // compare against what's actually available to read.
            let available = self.buffer.len().cast_signed() - self.peek_pos.cast_signed();
            let rem = size.cast_signed() - available;

            if rem <= 0 {
                break;
            }

            // request more data so we get to enough to actually resolve the
            // requested size.
            match self.fill_up() {
                Ok(fill_size) => {
                    if fill_size == 0 {
                        break;
                    }
                }
                Err(err) => {
                    cached_error = Some(err);
                    break;
                }
            }
        }

        let buffer_len = self.buffer.len();

        let original_peek = self.peek_pos;

        // never overflows, how much exactly did we have before we
        // requested for more data
        let total_data_left = buffer_len - original_peek;

        // we were less than even what is requested or exactly what was requested
        // move peek to the end and move on
        if total_data_left <= size {
            self.peek_pos = buffer_len;
        } else {
            // if we basically indicative that the length to wanted is basically
            self.peek_pos = original_peek + size;
        }

        let slice = &self.buffer[original_peek..self.peek_pos];

        if self.peek_pos == original_peek {
            // if we never moved beyond our existing position and cached error
            // is not None then return error instead.
            if let Some(cached_error) = cached_error.take() {
                return Err(cached_error);
            }

            return Ok(PeekState::NoNext);
        }
        Ok(PeekState::Request(slice))
    }

    /// [`Self::read_size`] copies up to `buf.len()` unconsumed bytes into `buf` and
    /// consumes exactly the bytes it copied. This is what [`Read::read`] uses.
    ///
    /// Bytes are taken from the data cursor, the first byte not yet consumed. Bytes
    /// you have only peeked at (through the `next*`/`forward*` methods) are not
    /// consumed yet, so this returns them too. Afterwards the data cursor has moved
    /// forward by the number of bytes returned. The peek cursor stays where it was,
    /// unless it was behind the new data cursor; then it moves up to it. Nothing
    /// past what was returned is discarded.
    ///
    /// The internal buffer is filled from the underlying reader until it holds
    /// `buf.len()` unconsumed bytes, or the reader reports EOF or an error. A short
    /// count therefore means EOF, or that the reader failed after some bytes were
    /// already buffered. `Ok(0)` means EOF with nothing left, or an empty `buf`.
    ///
    /// # Errors
    /// Returns the underlying reader's error (including `WouldBlock`) if it fails
    /// before any unconsumed byte is available. If some bytes are already
    /// buffered, they are returned instead and the error is left for the next call.
    pub fn read_size(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }

        let mut cached_error: Option<std::io::Error> = None;
        while self.remaining() < buf.len() {
            match self.fill_up() {
                Ok(0) => break,
                Ok(_) => {}
                Err(err) => {
                    cached_error = Some(err);
                    break;
                }
            }
        }

        // `fill_up` may compact the buffer, which moves both cursors, so the
        // data cursor is read only after filling.
        let available = self.remaining();
        if available == 0 {
            return match cached_error {
                Some(err) => Err(err),
                None => Ok(0),
            };
        }

        let copied = available.min(buf.len());
        let from = self.pos;
        buf[..copied].copy_from_slice(&self.buffer[from..from + copied]);

        self.pos = from + copied;
        if self.peek_pos < self.pos {
            self.peek_pos = self.pos;
        }

        Ok(copied)
    }

    /// [`Self::nextby`] provides a more friendly API ontop [`Self::nextby`] returning
    /// the slice `&[u8]` without the [`PeekState`] wrapper.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails or no data is available.
    pub fn nextby2(&mut self, size: usize) -> std::io::Result<&[u8]> {
        self.nextby(size).map(|item| match item {
            PeekState::Request(inner) => Ok(inner),
            PeekState::NoNext => Err(crate::err!(UnexpectedEof, "No more data to pull through")),
            PeekState::ZeroLengthInput => Err(crate::err!(WriteZero, "Provided zero size request")),
            _ => unreachable!("Should not trigger this stage"),
        })?
    }

    /// [`Self::next_until`] returns the peek state for the requested data until the delimiter is seen
    /// at which point the underlying reference to the data is either shared in a
    /// [`PeekState::Request`] if fully read else return [`PeekState::EndOfFile`] if the
    /// source returns EOF or if the data available is less than requested.
    ///
    /// It runs in a tight loop and ensures to acquire as much data as needed.
    ///
    /// This moves the peek cursor forward.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    #[inline]
    pub fn next_until<'a>(&'a mut self, signal: &[u8]) -> std::io::Result<PeekState<'a>> {
        if signal.is_empty() {
            return Ok(PeekState::ZeroLengthInput);
        }
        loop {
            let state = self.peek_by(signal.len())?;
            // tracing::trace!(
            //     "Next_Until: using signal: {:?} against {:?}",
            //     &signal,
            //     &state,
            // );
            match state {
                PeekState::Request(data) => {
                    // tracing::trace!(
                    //     "Next_Until: using signal: {:?}(len={}) against {:?} as {:?}",
                    //     &signal,
                    //     signal.len(),
                    //     &data,
                    //     String::from_utf8(data.to_vec()),
                    // );
                    if data == signal {
                        break;
                    }
                }
                PeekState::EndOfBuffered | PeekState::LessThanRequested => {
                    // if at the end then return with EndOfFile
                    if self.fill_up()? == 0 {
                        return Ok(PeekState::EndOfFile);
                    }
                    continue;
                }
                PeekState::Continue => continue,
                PeekState::EndOfFile => {
                    return Ok(PeekState::EndOfFile);
                }
                _ => {
                    unreachable!("Should never hit this types")
                }
            }

            self.peek_pos += signal.len();
        }

        // account for the found block where we break and ensure we can capture
        // the last peek correctly.
        self.peek_pos += signal.len();
        Ok(PeekState::Request(&self.buffer[self.pos..self.peek_pos]))
    }

    /// [`Self::next_line`] reads the provided line into a string without consuming the read content.
    /// Allowing you to perform further operation on the data.
    ///
    /// If the newline byte is not found then it reads all the bytes into the provided buffer ontil
    /// it hits EOF or `EndOfFile` but this is key, it won't move the cursor forward, just copies the
    /// underlying data over.
    ///
    /// This moves the peek cursor forward.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    pub fn next_line(&mut self, buf: &mut String) -> std::io::Result<usize> {
        const NEWLINE_SLICE: &[u8] = b"\n";
        let read = {
            let inner = self.next_until(NEWLINE_SLICE)?;
            match inner {
                PeekState::Request(c) => {
                    unsafe {
                        let buf_vec = buf.as_mut_vec();
                        buf_vec.extend_from_slice(c);
                    };
                    Ok(c.len())
                }
                PeekState::EndOfFile => Ok(0),
                PeekState::ZeroLengthInput => {
                    Err(crate::err!(WriteZero, "Provided zero size request"))
                }
                _ => unreachable!("Should never trigger"),
            }
        };

        match read {
            Ok(inner) => {
                if inner == 0 {
                    let slice = &self.buffer[self.pos..self.peek_pos];
                    let slice_length = slice.len();

                    unsafe {
                        let mut buf_vec = buf.as_mut_vec();
                        buf_vec.extend_from_slice(slice);
                    };

                    return Ok(slice_length);
                }

                Ok(inner)
            }
            Err(err) => Err(err),
        }
    }

    /// [`Self::next_bytes_until`] reads the provided bytes into a Vec without consuming the read cursor.
    /// Allowing you to perform further operation on the data.
    ///
    /// If the target byte is not found then it reads all the bytes into the provided buffer ontil
    /// it hits EOF or `EndOfFile` but this is key, it won't move the cursor forward, just copies the
    /// underlying data over.
    ///
    /// This moves the peek cursor forward.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    pub fn next_bytes_until(&mut self, target: &[u8], buf: &mut Vec<u8>) -> std::io::Result<usize> {
        let read = {
            let inner = self.next_until(target)?;
            match inner {
                PeekState::Request(c) => {
                    unsafe {
                        buf.extend_from_slice(c);
                    };
                    Ok(c.len())
                }
                PeekState::EndOfFile => Ok(0),
                PeekState::ZeroLengthInput => {
                    Err(crate::err!(WriteZero, "Provided zero size request"))
                }
                _ => unreachable!("Should never trigger"),
            }
        };

        match read {
            Ok(inner) => {
                if inner == 0 {
                    let slice = &self.buffer[self.pos..self.peek_pos];
                    let slice_length = slice.len();

                    unsafe {
                        buf.extend_from_slice(slice);
                    };

                    return Ok(slice_length);
                }

                Ok(inner)
            }
            Err(err) => Err(err),
        }
    }

    /// [`Self::read_bytes_until`] reads the provided bytes into a Vec consuming read cursor, until it
    /// finds the relevant else stopping.
    ///
    /// If the new line byte is not found then everything read till that point is appended to the
    /// string and the cursor is consumed.
    ///
    /// This moves the peek cursor forward.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    pub fn read_bytes_until(&mut self, target: &[u8], buf: &mut Vec<u8>) -> std::io::Result<usize> {
        let read = {
            let inner = self.next_until(target)?;
            match inner {
                PeekState::Request(c) => {
                    unsafe {
                        buf.extend_from_slice(c);
                    };
                    Ok(c.len())
                }
                PeekState::EndOfFile => Ok(0),
                PeekState::ZeroLengthInput => {
                    Err(crate::err!(WriteZero, "Provided zero size request"))
                }
                _ => unreachable!("Should never trigger"),
            }
        };

        match read {
            Ok(inner) => {
                if inner == 0 {
                    let slice = &self.buffer[self.pos..self.peek_pos];
                    let slice_length = slice.len();
                    unsafe {
                        buf.extend_from_slice(slice);
                    };

                    self.skip();
                    return Ok(slice_length);
                }

                self.skip();
                Ok(inner)
            }
            Err(err) => Err(err),
        }
    }

    /// [`Self::read_line`] reads the provided line into a string and consumes the read data moving
    /// the cusrsor forward.
    ///
    /// If the new line byte is not found then everything read till that point is appended to the
    /// string and the cursor is consumed.
    ///
    /// This moves the peek cursor forward.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    pub fn read_line(&mut self, buf: &mut String) -> std::io::Result<usize> {
        const NEWLINE_SLICE: &[u8] = b"\n";
        let read = {
            let inner = self.next_until(NEWLINE_SLICE)?;
            match inner {
                PeekState::Request(c) => {
                    match String::from_utf8(c.to_vec()) {
                        Ok(inner) => {
                            buf.push_str(&inner);
                            Ok(inner.len())
                        }
                        Err(err) => {
                            tracing::error!(
                                "Received invalid utf8 data {:?} with error: {:?}",
                                &c,
                                err,
                            );

                            // fallback into character by character handling
                            buf.extend(c.iter().map(|b| char::from(*b)));

                            Ok(c.len())
                        }
                    }
                }
                PeekState::EndOfFile => Ok(0),
                PeekState::ZeroLengthInput => {
                    Err(err!(InvalidInput, "Zero length input not allowed"))
                }
                _ => unreachable!("Should never trigger"),
            }
        };

        match read {
            Ok(inner) => {
                if inner == 0 {
                    let slice = &self.buffer[self.pos..self.peek_pos];
                    let slice_length = slice.len();

                    unsafe {
                        let mut buf_vec = buf.as_mut_vec();
                        buf_vec.extend_from_slice(slice);
                    };

                    self.skip();
                    return Ok(slice_length);
                }

                self.skip();
                Ok(inner)
            }
            Err(err) => Err(err),
        }
    }

    /// [`Self::read_all`] pulls the rest of the underlying stream, up to EOF, and appends
    /// every unconsumed byte to `buf`, consuming all of them.
    ///
    /// Like [`Self::read_size`], it starts at the data cursor, so bytes that were only
    /// peeked at are included. Returns the number of bytes appended.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying stream fails or if the read limit is exceeded.
    pub fn read_all(
        &mut self,
        buf: &mut Vec<u8>,
        read_limit: Option<usize>,
    ) -> std::io::Result<usize> {
        self.fill_all(read_limit)?;

        let buffer_len = self.buffer.len();
        let slice = &self.buffer[self.pos..buffer_len];
        let slice_length = slice.len();

        buf.extend_from_slice(slice);

        self.peek_pos = buffer_len;
        self.pos = buffer_len;

        Ok(slice_length)
    }
}

impl<T: Read + Write> Write for ByteBufferPointer<T> {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        self.reader.write(buf)
    }

    fn flush(&mut self) -> Result<()> {
        self.reader.flush()
    }
}

impl<T: Read> Read for ByteBufferPointer<T> {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        self.read_size(buf)
    }
}

impl<T: Read> PeekableReadStream for ByteBufferPointer<T> {
    /// Copies upcoming bytes into `buf` without moving either cursor, as the
    /// trait requires. It looks from the peek cursor (see [`ByteBufferPointer::peekby`]),
    /// so when no `next*`/`forward*` peek is outstanding, a following
    /// [`Read::read`] returns the same bytes.
    ///
    /// Returns `Ok(0)` once the stream is drained and at EOF.
    fn peek(&mut self, buf: &mut [u8]) -> std::result::Result<usize, PeekError> {
        match self.peekby(buf.len()) {
            Ok(PeekState::Request(data) | PeekState::Stalled(data)) => {
                let copied = data.len().min(buf.len());
                buf[..copied].copy_from_slice(&data[..copied]);
                Ok(copied)
            }
            Ok(PeekState::ZeroLengthInput) => Err(PeekError::ZeroLengthNotAllowed),
            // `peekby` returns `NoNext` when nothing is buffered and the reader
            // is at EOF; the remaining states carry no bytes either.
            Ok(
                PeekState::NoNext
                | PeekState::LessThanRequested
                | PeekState::EndOfBuffered
                | PeekState::EndOfFile
                | PeekState::Continue,
            ) => Ok(0),
            Err(err) => Err(PeekError::IOError(err)),
        }
    }
}

impl<T: Read> BufferCapacity for ByteBufferPointer<T> {
    fn read_buffer(&self) -> &[u8] {
        &self.buffer
    }

    fn read_capacity(&self) -> usize {
        self.pull_amount
    }
}

// -- Cursor

pub struct BufferedCapacityCursor<T>(std::io::Cursor<T>);

impl BufferCapacity for BufferedCapacityCursor<&str> {
    fn read_buffer(&self) -> &[u8] {
        self.0.get_ref().as_bytes()
    }

    fn read_capacity(&self) -> usize {
        self.0.get_ref().len()
    }
}

impl BufferCapacity for BufferedCapacityCursor<String> {
    fn read_buffer(&self) -> &[u8] {
        self.0.get_ref().as_bytes()
    }

    fn read_capacity(&self) -> usize {
        self.0.get_ref().len()
    }
}

impl BufferCapacity for BufferedCapacityCursor<&[u8]> {
    fn read_buffer(&self) -> &[u8] {
        self.0.get_ref()
    }

    fn read_capacity(&self) -> usize {
        self.0.get_ref().len()
    }
}

impl BufferCapacity for BufferedCapacityCursor<Vec<u8>> {
    fn read_buffer(&self) -> &[u8] {
        self.0.get_ref()
    }

    fn read_capacity(&self) -> usize {
        self.0.get_ref().len()
    }
}

impl<T> BufferedCapacityCursor<T> {
    pub fn new(cursor: std::io::Cursor<T>) -> Self {
        Self(cursor)
    }

    /// `get_ref` returns the reference to the wrapped `Cursor<T>`.
    pub fn get_ref(&self) -> &Cursor<T> {
        &self.0
    }

    /// `get_mut` returns the mutable reference to the wrapped `Cursor<T>`.
    pub fn get_mut(&mut self) -> &mut Cursor<T> {
        &mut self.0
    }

    /// `get_inner_mut` returns a immutable reference to the  inner content
    /// of the wrapped `Cursor<T>`.
    pub fn get_inner_ref(&self) -> &T {
        self.0.get_ref()
    }

    /// `get_inner_mut` returns a mutable reference to the  inner content
    /// of the wrapped `Cursor<T>`.
    pub fn get_inner_mut(&mut self) -> &mut T {
        self.0.get_mut()
    }
}

// ============================================================================
// SharedBuffer - Thread-safe shared buffer for producer-consumer patterns
// ============================================================================

/// Writer handle for `SharedBuffer`.
///
/// Holds a clone of the `Arc<Mutex<Vec<u8>>>` and implements `Write`.
///
/// # Thread Safety
/// This type is `Send + Sync` and can be safely shared across threads.
#[derive(Clone)]
pub struct SharedBufferWriter {
    inner: Arc<Mutex<Vec<u8>>>,
}

impl Write for SharedBufferWriter {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedBufferWriter lock poisoned");
        guard.extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> Result<()> {
        Ok(())
    }
}

impl SharedBufferWriter {
    /// Create a new writer from a shared `Arc<Mutex<Vec<u8>>>`.
    pub fn new(inner: Arc<Mutex<Vec<u8>>>) -> Self {
        Self { inner }
    }

    /// Get a clone of the underlying Arc.
    #[must_use]
    pub fn clone_arc(&self) -> Arc<Mutex<Vec<u8>>> {
        Arc::clone(&self.inner)
    }
}

/// Reader handle for `SharedBuffer`.
///
/// Holds a clone of the `Arc<Mutex<Vec<u8>>>` and implements `Read`.
/// Tracks read position independently from writers.
///
/// # Thread Safety
/// This type is `Send + Sync` and can be safely shared across threads.
#[derive(Clone)]
pub struct SharedBufferReader {
    inner: Arc<Mutex<Vec<u8>>>,
    position: usize,
}

impl Read for SharedBufferReader {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        let guard = self.inner.lock().expect("SharedBufferReader lock poisoned");
        let available = guard.len().saturating_sub(self.position);
        if available == 0 {
            return Ok(0);
        }
        let to_read = buf.len().min(available);
        buf[..to_read].copy_from_slice(&guard[self.position..self.position + to_read]);
        drop(guard);
        self.position += to_read;
        Ok(to_read)
    }
}

impl SharedBufferReader {
    /// Create a new reader from a shared `Arc<Mutex<Vec<u8>>>`.
    pub fn new(inner: Arc<Mutex<Vec<u8>>>) -> Self {
        Self { inner, position: 0 }
    }

    /// Create a new reader with a specific starting position.
    pub fn with_position(inner: Arc<Mutex<Vec<u8>>>, position: usize) -> Self {
        Self { inner, position }
    }

    /// Get the current read position.
    #[must_use]
    pub fn position(&self) -> usize {
        self.position
    }

    /// Convert this reader into a `SharedByteBufferStream`.
    #[must_use]
    pub fn into_buffered_stream(self) -> SharedByteBufferStream<SharedBufferReadStream> {
        SharedByteBufferStream::rwrite(SharedBufferReadStream::new(
            Arc::clone(&self.inner),
            self.position,
        ))
    }
}

/// A readable stream that wraps `SharedBuffer` for use with `SharedByteBufferStream`.
pub struct SharedBufferReadStream {
    inner: Arc<Mutex<Vec<u8>>>,
    position: usize,
}

impl SharedBufferReadStream {
    /// Create a new `SharedBufferReadStream` from a shared buffer.
    ///
    /// # Arguments
    /// * `inner` - The shared buffer
    /// * `position` - The starting read position
    ///
    /// # Returns
    /// A `SharedBufferReadStream` that reads from the shared buffer.
    pub fn new(inner: Arc<Mutex<Vec<u8>>>, position: usize) -> Self {
        Self { inner, position }
    }
}

impl std::io::Read for SharedBufferReadStream {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        let guard = self.inner.lock().expect("SharedBuffer lock poisoned");
        let available = guard.len().saturating_sub(self.position);
        if available == 0 {
            return Ok(0);
        }
        let to_read = buf.len().min(available);
        buf[..to_read].copy_from_slice(&guard[self.position..self.position + to_read]);
        drop(guard);
        self.position += to_read;
        Ok(to_read)
    }
}

/// A thread-safe shared buffer for producer-consumer patterns.
///
/// This type provides a convenient way to create a shared buffer with
/// separate writer and reader handles. The buffer is wrapped in
/// `Arc<Mutex<>>` for thread-safe access.
///
/// # Examples
///
/// ```rust
/// use foundation_core::io::ioutils::SharedBuffer;
/// use std::io::{Read, Write};
///
/// // Create writer and reader handles
/// let (mut writer, mut reader) = SharedBuffer::split();
///
/// // Write data
/// writer.write_all(b"hello").unwrap();
///
/// // Read data
/// let mut buf = [0u8; 5];
/// reader.read_exact(&mut buf).unwrap();
/// assert_eq!(&buf, b"hello");
/// ```
pub struct SharedBuffer {
    inner: Arc<Mutex<Vec<u8>>>,
}

impl SharedBuffer {
    /// Create a new `SharedBuffer` with separate writer and reader handles.
    ///
    /// # Returns
    /// A tuple of (`SharedBufferWriter`, `SharedBufferReader`) that share
    /// the same underlying buffer.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use foundation_core::io::ioutils::SharedBuffer;
    /// use std::io::{Read, Write};
    ///
    /// let (mut writer, mut reader) = SharedBuffer::split();
    /// writer.write_all(b"hello").unwrap();
    /// let mut buf = [0u8; 5];
    /// reader.read_exact(&mut buf).unwrap();
    /// assert_eq!(&buf, b"hello");
    /// ```
    #[must_use]
    pub fn split() -> (SharedBufferWriter, SharedBufferReader) {
        let inner = Arc::new(Mutex::new(Vec::new()));
        let writer = SharedBufferWriter::new(Arc::clone(&inner));
        let reader = SharedBufferReader::new(inner);
        (writer, reader)
    }

    /// Create a new `SharedBuffer` with initial content.
    ///
    /// # Arguments
    /// * `initial` - Initial bytes to populate the buffer with
    ///
    /// # Returns
    /// A `SharedBuffer` containing the initial data.
    #[must_use]
    pub fn with_initial(initial: &[u8]) -> Self {
        let inner = Arc::new(Mutex::new(initial.to_vec()));
        Self { inner }
    }

    /// Get the current length of the buffer.
    ///
    /// # Panics
    /// Panics if the mutex lock is poisoned (i.e., another thread panicked while holding the lock).
    ///
    /// # Returns
    /// The number of bytes currently in the buffer.
    #[must_use]
    pub fn len(&self) -> usize {
        let guard = self.inner.lock().expect("SharedBuffer lock poisoned");
        guard.len()
    }

    /// Check if the buffer is empty.
    ///
    /// # Returns
    /// `true` if the buffer contains no bytes, `false` otherwise.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Get a clone of the underlying Arc, e.g. to build a
    /// [`SharedBufferWriter`] or [`SharedBufferReader`] over a buffer made with
    /// [`Self::with_initial`].
    #[must_use]
    pub fn clone_arc(&self) -> Arc<Mutex<Vec<u8>>> {
        Arc::clone(&self.inner)
    }
}

impl Default for SharedBuffer {
    fn default() -> Self {
        Self {
            inner: Arc::new(Mutex::new(Vec::new())),
        }
    }
}

// ============================================================================
// SharedWriteReader - Thread-safe shared Read+Write handle
// ============================================================================

/// A thread-safe shared handle that supports both Read and Write operations.
///
/// WHY: Some use cases require a single handle that can be shared across threads
/// where both reading and writing operations are needed on the same underlying resource.
///
/// WHAT: This type wraps any `T: Read + Write` with `Arc<Mutex<>>` for thread-safe access.
///
/// HOW: Provides cloneable handles that share the same underlying Read+Write resource.
pub struct SharedWriteReader<T: Read + Write> {
    inner: Arc<Mutex<T>>,
}

impl<T: Read + Write> Clone for SharedWriteReader<T> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<T: Read + Write> SharedWriteReader<T> {
    /// Create a new `SharedWriteReader` from a `Read`+`Write` type.
    ///
    /// # Returns
    /// A `SharedWriteReader` that owns the wrapped value.
    pub fn new(inner: T) -> Self {
        Self {
            inner: Arc::new(Mutex::new(inner)),
        }
    }

    /// Create a new `SharedWriteReader` from an existing `Arc<Mutex<T>>`.
    ///
    /// # Returns
    /// A `SharedWriteReader` wrapping the provided Arc.
    pub fn from_arc(arc: Arc<Mutex<T>>) -> Self {
        Self { inner: arc }
    }

    /// Clone the underlying Arc reference.
    #[must_use]
    pub fn clone_arc(&self) -> Arc<Mutex<T>> {
        Arc::clone(&self.inner)
    }
}

impl<T: Read + Write> Read for SharedWriteReader<T> {
    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.read(buf)
    }

    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.read_exact(buf)
    }

    fn read_to_end(&mut self, buf: &mut Vec<u8>) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.read_to_end(buf)
    }

    fn read_to_string(&mut self, buf: &mut String) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.read_to_string(buf)
    }

    fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.read_vectored(bufs)
    }
}

impl<T: Read + Write> Write for SharedWriteReader<T> {
    fn write(&mut self, buf: &[u8]) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.write(buf)
    }

    fn write_all(&mut self, buf: &[u8]) -> Result<()> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.write_all(buf)
    }

    fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.write_vectored(bufs)
    }

    fn flush(&mut self) -> Result<()> {
        let mut guard = self.inner.lock().expect("SharedWriteReader lock poisoned");
        guard.flush()
    }
}

unsafe impl<T: Read + Write + Send> Send for SharedWriteReader<T> {}
unsafe impl<T: Read + Write + Send + Sync> Sync for SharedWriteReader<T> {}