foundation_core 0.0.5

// 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()
    }
}

#[cfg(test)]
mod buffered_reader_tests {
    use super::*;

    #[test]
    fn can_buffered_reader_peek() {
        let content = b"alexander_wonderbat";
        let mut reader = BufferedReader::new(&content[..]);

        let mut content_to_read = vec![0; 5];
        reader
            .peek(&mut content_to_read)
            .expect("should read data correctly");

        assert_eq!(b"alexa", &content_to_read[..]);

        assert_eq!(content, reader.buffer());
    }
}

// -- 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_all`] reads the underlying size of the buffer provided data, consuming
    /// all that is requested which moves both the peek cursor and the actual data cursor
    /// which means that part of the internal buffer read from the underlying reader will
    /// at some point be discarded as it should be.
    ///
    /// # Errors
    /// Returns an error if reading from the underlying reader fails.
    pub fn read_size(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let read = match self.nextby(buf.len()) {
            Ok(state) => match state {
                PeekState::Request(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(ending)
                }
                PeekState::LessThanRequested => {
                    let ending = if buf.len() > self.buffer.len() {
                        self.buffer.len()
                    } else {
                        buf.len()
                    };
                    for (index, elem) in self.buffer[0..ending].iter().enumerate() {
                        buf[index] = *elem;
                    }
                    Ok(ending)
                }
                _ => return Ok(0),
            },
            Err(err) => Err(err),
        };

        match read {
            Ok(val) => {
                self.skip();
                Ok(val)
            }
            Err(err) => Err(err),
        }
    }

    /// [`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 whole data within the underlying stream into provided buffer
    /// returning the total length of bytes read out after pulling all the data within the
    /// stream until EOF.
    ///
    /// # 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)?;

        // get len of buffer
        let buffer_len = self.buffer.len();

        // get current peek position
        let original_peek = self.peek_pos;

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

        buf.extend_from_slice(slice);

        self.peek_pos = buffer_len;
        self.skip();

        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> {
    fn peek(&mut self, buf: &mut [u8]) -> std::result::Result<usize, PeekError> {
        match self.nextby(buf.len()) {
            Ok(state) => match state {
                PeekState::Request(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())
                }
                PeekState::ZeroLengthInput => Err(PeekError::ZeroLengthNotAllowed),
                _ => unreachable!("We should never hit this state"),
            },
            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
    }
}

#[cfg(test)]
mod byte_buffered_buffer_pointer {
    use std::io::Cursor;
    use std::sync::Arc;
    use std::sync::Mutex;

    use super::*;

    #[test]
    fn fails_read_all_the_data_when_limit_is_reached() {
        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(5, reader));

        let mut binding = buffer.lock().unwrap();

        let mut read_data = Vec::new();

        let read_length = binding.read_all(&mut read_data, Some(2));
        assert!(read_length.is_err())
    }

    #[test]
    fn can_read_all_the_data() {
        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(5, reader));

        let mut binding = buffer.lock().unwrap();

        let mut read_data = Vec::new();

        let read_length = binding
            .read_all(&mut read_data, None)
            .expect("read data successfully");
        assert_eq!(read_length, content.len());

        let data_4_str = str::from_utf8(&read_data[0..read_length]).expect("convert into string");
        assert_eq!(data_4_str, "alexander_wonderbat");
    }

    #[test]
    fn can_read_bytes_until_found() {
        const NO_BAR: &[u8] = b"_";
        let content = b"alexander_wonderbat\n";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();

        let mut new_line = Vec::new();
        let read = binding.read_bytes_until(NO_BAR, &mut new_line);

        assert_eq!(10, read.unwrap());
        let data4 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(new_line, data4);
    }

    #[test]
    fn can_read_bytes_until_not_found() {
        const NO_BAR: &[u8] = b"-";
        let content = b"alexander_wonderbat\n";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();

        let mut new_line = Vec::new();
        let read = binding.read_bytes_until(NO_BAR, &mut new_line);

        assert_eq!(content.len(), read.unwrap());
        let data4 = vec![
            97, 108, 101, 120, 97, 110, 100, 101, 114, 95, 119, 111, 110, 100, 101, 114, 98, 97,
            116, 10,
        ];
        assert_eq!(new_line, data4);
    }

    #[test]
    fn can_peek_line() {
        let content = b"alexander_wonderbat\n";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();

        let mut new_line = String::new();
        let read = binding.next_line(&mut new_line);

        assert_eq!(content.len(), read.unwrap());
        assert_eq!(new_line, "alexander_wonderbat\n");

        let data4 = vec![
            97, 108, 101, 120, 97, 110, 100, 101, 114, 95, 119, 111, 110, 100, 101, 114, 98, 97,
            116, 10,
        ];
        assert_eq!(binding.scan(), &data4);
    }

    #[test]
    fn can_read_line() {
        let content = b"alexander_wonderbat\n";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();

        let mut new_line = String::new();
        let read = binding.read_line(&mut new_line);

        assert_eq!(content.len(), read.unwrap());
        assert_eq!(new_line, "alexander_wonderbat\n");
    }

    #[test]
    fn can_peek_until_a_signal_and_consume() {
        const SIGNAL: &[u8] = b"_";

        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();
        assert!(matches!(
            binding.next_until(SIGNAL).expect("should peek"),
            PeekState::Request(_)
        ));

        let result = binding.consume();
        let content = result.unwrap();

        let data4 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(content, data4);

        let data_4_str = str::from_utf8(&data4).expect("convert into string");
        assert_eq!(data_4_str, "alexander_");
    }

    #[test]
    fn can_peek_until_a_signal() {
        const SIGNAL: &[u8] = b"_";

        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();
        let result = binding.next_until(SIGNAL);
        assert!(matches!(result, Ok(PeekState::Request(_))));

        let PeekState::Request(content) = result.unwrap() else {
            panic!("Failed expectation")
        };

        let data4 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(content, &data4);

        let data_4_str = str::from_utf8(&data4).expect("convert into string");
        assert_eq!(data_4_str, "alexander_");
    }

    #[test]
    fn can_peek_until_a_signal_but_not_found() {
        const SIGNAL: &[u8] = b"-";

        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        let mut binding = buffer.lock().unwrap();
        let result = binding.next_until(SIGNAL);
        assert!(matches!(result, Ok(PeekState::EndOfFile)));
    }

    #[test]
    fn can_take_peeked_data_with_multi_mutex() {
        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let buffer = Mutex::new(ByteBufferPointer::new(10, reader));

        assert_eq!(
            buffer.lock().unwrap().peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.lock().unwrap().fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.lock().unwrap().peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );

        assert_eq!(
            buffer.lock().unwrap().forward_by(9).expect("capture 3"),
            PeekState::Continue
        );

        assert_eq!(buffer.lock().unwrap().scan(), &data3[0..9]);
        assert_eq!(buffer.lock().unwrap().full_scan(), &data3[0..10]);

        assert_eq!(
            buffer.lock().unwrap().forward_by(10).expect("capture 4"),
            PeekState::EndOfBuffered
        );

        let amt = buffer.lock().unwrap().fill_up().expect("should fill up");

        assert_eq!(
            buffer.lock().unwrap().forward_by(amt).expect("capture 3"),
            PeekState::Continue
        );

        let data4 = vec![
            97, 108, 101, 120, 97, 110, 100, 101, 114, 95, 119, 111, 110, 100, 101, 114, 98, 97,
            116,
        ];
        assert_eq!(buffer.lock().unwrap().scan(), &data4);
    }

    #[test]
    fn can_read_data_via_byte_buffer_pointer() {
        let content = b"alexander_wonderbat";
        let mut source = Cursor::new(content.to_vec());
        let reader = OwnedReader::atomic(&mut source);
        let mut buffer = ByteBufferPointer::new(128, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data = vec![97];

        assert_eq!(
            buffer.peek().expect("capture 1"),
            PeekState::Request(data.as_ref())
        );

        let data2 = vec![97, 108];
        assert_eq!(
            buffer.peek_by(2).expect("capture 2"),
            PeekState::Request(data2.as_ref())
        );

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );
    }

    #[test]
    fn can_take_peeked_data() {
        let content = b"alexander_wonderbat";
        let mut source = Cursor::new(content.to_vec());
        let reader = OwnedReader::atomic(&mut source);
        let mut buffer = ByteBufferPointer::new(128, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );
    }

    #[test]
    fn can_take_unmove_data() {
        let content = b"alexander_wonderbat";
        let mut source = Cursor::new(content.to_vec());
        let reader = OwnedReader::atomic(&mut source);
        let mut buffer = ByteBufferPointer::new(128, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );

        assert_eq!(
            buffer.forward_by(10).expect("capture 3"),
            PeekState::Continue
        );

        assert_eq!(buffer.scan(), &data3[0..10]);

        assert_eq!(
            buffer.unforward_by(5).expect("capture 3"),
            PeekState::Continue
        );

        assert_eq!(buffer.scan(), &data3[0..5]);
    }

    #[test]
    fn can_skip_data() {
        let content = b"alexander_wonderbat";
        let mut source = Cursor::new(content.to_vec());
        let reader = OwnedReader::atomic(&mut source);
        let mut buffer = ByteBufferPointer::new(128, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );

        assert_eq!(
            buffer.forward_by(10).expect("capture 3"),
            PeekState::Continue
        );

        assert_eq!(buffer.scan(), &data3[0..10]);

        assert_eq!(
            buffer.unforward_by(5).expect("capture 3"),
            PeekState::Continue
        );

        buffer.skip();

        let data4: Vec<u8> = vec![];
        assert_eq!(buffer.scan(), &data4);

        let data5: Vec<u8> = vec![110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(5).expect("capture 3"),
            PeekState::Request(data5.as_ref())
        );
    }

    #[test]
    fn can_use_buffered_reader_move_next_data() {
        let content = b"alexander_wonderbat";
        let mut reader = BufferedReader::new(BufferedWriter::new(Cursor::new(content.to_vec())));
        let mut buffer = ByteBufferPointer::new(128, OwnedReader::atomic(&mut reader));

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(0, buffer.data_cursor());
        assert_eq!(0, buffer.peek_cursor());
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );
        assert_eq!(
            buffer.forward_by(10).expect("capture 3"),
            PeekState::Continue
        );
        assert_eq!(buffer.scan(), &data3[0..10]);
        assert!(!buffer.greater_than_40_percent());
        assert_eq!(buffer.consume().expect("capture 3"), data3.clone());
        assert!(buffer.greater_than_40_percent());
    }

    #[test]
    fn can_move_next_data() {
        let content = b"alexander_wonderbat";
        let mut source = Cursor::new(content.to_vec());
        let reader = OwnedReader::atomic(&mut source);
        let mut buffer = ByteBufferPointer::new(128, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(0, buffer.data_cursor());
        assert_eq!(0, buffer.peek_cursor());
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );
        assert_eq!(
            buffer.forward_by(10).expect("capture 3"),
            PeekState::Continue
        );
        assert_eq!(buffer.scan(), &data3[0..10]);
        assert!(!buffer.greater_than_40_percent());
        assert_eq!(buffer.consume().expect("capture 3"), data3.clone());
        assert!(buffer.greater_than_40_percent());
    }

    #[test]
    fn can_pull_more_data() {
        let content = b"alexander_wonderbat";

        let mut source = Cursor::new(content.to_vec());
        let reader = OwnedReader::atomic(&mut source);
        let mut buffer = ByteBufferPointer::new(10, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );

        assert_eq!(
            buffer.forward_by(9).expect("capture 3"),
            PeekState::Continue
        );

        assert_eq!(buffer.scan(), &data3[0..9]);
        assert_eq!(buffer.full_scan(), &data3[0..10]);

        assert_eq!(
            buffer.forward_by(10).expect("capture 4"),
            PeekState::EndOfBuffered
        );

        let amt = buffer.fill_up().expect("should fill up");

        assert_eq!(
            buffer.forward_by(amt).expect("capture 3"),
            PeekState::Continue
        );

        let data4 = vec![
            97, 108, 101, 120, 97, 110, 100, 101, 114, 95, 119, 111, 110, 100, 101, 114, 98, 97,
            116,
        ];
        assert_eq!(buffer.scan(), &data4);
    }

    #[test]
    fn can_pull_more_data_with_mutex() {
        let content = b"alexander_wonderbat";

        let reader = OwnedReader::Sync(Arc::new(Mutex::new(Cursor::new(content.to_vec()))));
        let mut buffer = ByteBufferPointer::new(10, reader);

        assert_eq!(
            buffer.peek().expect("less than requested"),
            PeekState::LessThanRequested
        );

        buffer.fill_up().expect("should fill up");

        let data3 = vec![97, 108, 101, 120, 97, 110, 100, 101, 114, 95];
        assert_eq!(
            buffer.peek_by(10).expect("capture 3"),
            PeekState::Request(data3.as_ref())
        );

        assert_eq!(
            buffer.forward_by(9).expect("capture 3"),
            PeekState::Continue
        );

        assert_eq!(buffer.scan(), &data3[0..9]);
        assert_eq!(buffer.full_scan(), &data3[0..10]);

        assert_eq!(
            buffer.forward_by(10).expect("capture 4"),
            PeekState::EndOfBuffered
        );

        let amt = buffer.fill_up().expect("should fill up");

        assert_eq!(
            buffer.forward_by(amt).expect("capture 3"),
            PeekState::Continue
        );

        let data4 = vec![
            97, 108, 101, 120, 97, 110, 100, 101, 114, 95, 119, 111, 110, 100, 101, 114, 98, 97,
            116,
        ];
        assert_eq!(buffer.scan(), &data4);
    }
}

// -- 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
    }
}

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> {}

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

#[cfg(test)]
mod shared_write_reader_tests {
    use super::*;
    use std::io::Cursor;

    #[test]
    fn test_shared_write_reader_write_and_read_position() {
        let cursor = Cursor::new(Vec::new());
        let shared = SharedWriteReader::new(cursor);

        let mut writer = shared.clone();
        writer
            .write_all(b"hello world")
            .expect("write should succeed");

        let inner = shared.inner.lock().expect("lock should not be poisoned");
        assert_eq!(inner.get_ref(), b"hello world");
    }

    #[test]
    fn test_shared_write_reader_multiple_writers() {
        let cursor = Cursor::new(Vec::new());
        let shared = SharedWriteReader::new(cursor);

        let mut w1 = shared.clone();
        let mut w2 = shared.clone();

        w1.write_all(b"hello ").expect("write should succeed");
        w2.write_all(b"world").expect("write should succeed");

        let inner = shared.inner.lock().expect("lock should not be poisoned");
        let data = inner.get_ref();
        assert_eq!(data, b"hello world");
    }

    #[test]
    fn test_shared_write_reader_from_arc() {
        let arc = Arc::new(Mutex::new(Cursor::new(Vec::new())));
        let shared = SharedWriteReader::from_arc(arc.clone());

        let mut writer = shared.clone();
        writer
            .write_all(b"test data")
            .expect("write should succeed");

        let guard = arc.lock().expect("lock should not be poisoned");
        let data = guard.get_ref();
        assert_eq!(data, b"test data");
    }
}

#[cfg(test)]
mod shared_byte_buffer_stream_tests {
    use super::*;

    #[test]
    fn test_shared_write_reader_with_cursor() {
        let cursor = Cursor::new(Vec::new());
        let shared = SharedWriteReader::new(cursor);

        let mut w1 = shared.clone();
        w1.write_all(b"hello").expect("write should succeed");

        let r1 = shared.clone();
        let inner = r1.inner.lock().expect("lock not poisoned");
        assert_eq!(inner.get_ref(), b"hello");
    }

    #[test]
    fn test_shared_write_reader_threaded() {
        let cursor = Cursor::new(Vec::new());
        let shared = SharedWriteReader::new(cursor);

        let mut handles = vec![];
        for i in 0..3 {
            let s = shared.clone();
            let handle = std::thread::spawn(move || {
                let mut writer = s.clone();
                let msg = format!("thread_{}", i);
                writer
                    .write_all(msg.as_bytes())
                    .expect("write should succeed");
            });
            handles.push(handle);
        }

        for handle in handles {
            handle.join().expect("thread should join");
        }

        let inner = shared.inner.lock().expect("lock not poisoned");
        let data = inner.get_ref();
        assert!(data.starts_with(b"thread_"));
    }
}

// ============================================================================
// SharedBuffer Tests
// ============================================================================

#[cfg(test)]
mod shared_buffer_tests {
    use super::*;

    #[test]
    fn test_shared_buffer_write_and_read() {
        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");
    }

    #[test]
    fn test_shared_buffer_empty_read() {
        let (_, mut reader) = SharedBuffer::split();

        let mut buf = [0u8; 5];
        let bytes_read = reader.read(&mut buf).unwrap();

        assert_eq!(bytes_read, 0);
    }

    #[test]
    fn test_shared_buffer_multiple_writes() {
        let (mut writer, mut reader) = SharedBuffer::split();

        writer.write_all(b"hello").unwrap();
        writer.write_all(b" ").unwrap();
        writer.write_all(b"world").unwrap();

        let mut buf = [0u8; 11];
        reader.read_exact(&mut buf).unwrap();

        assert_eq!(&buf, b"hello world");
    }

    #[test]
    fn test_shared_buffer_partial_read() {
        let (mut writer, mut reader) = SharedBuffer::split();

        writer.write_all(b"hello world").unwrap();

        let mut buf = [0u8; 5];
        reader.read_exact(&mut buf).unwrap();
        assert_eq!(&buf, b"hello");

        reader.read_exact(&mut buf).unwrap();
        assert_eq!(&buf, b" worl");
    }

    #[test]
    fn test_shared_buffer_clone_writer() {
        let (mut writer, mut reader) = SharedBuffer::split();
        let mut writer2 = writer.clone();

        writer.write_all(b"hello").unwrap();
        writer2.write_all(b" ").unwrap();
        writer.write_all(b"world").unwrap();

        let mut buf = [0u8; 11];
        reader.read_exact(&mut buf).unwrap();

        assert_eq!(&buf, b"hello world");
    }

    #[test]
    fn test_shared_buffer_clone_reader() {
        let (mut writer, mut reader) = SharedBuffer::split();
        let mut reader2 = reader.clone();

        writer.write_all(b"hello").unwrap();

        let mut buf1 = [0u8; 5];
        let mut buf2 = [0u8; 5];

        reader.read_exact(&mut buf1).unwrap();
        reader2.read_exact(&mut buf2).unwrap();

        assert_eq!(&buf1, b"hello");
        assert_eq!(&buf2, b"hello");
    }

    #[test]
    fn test_shared_buffer_with_initial() {
        let buffer = SharedBuffer::with_initial(b"initial data");
        let (_, mut reader) = (
            SharedBufferWriter::new(buffer.inner.clone()),
            SharedBufferReader::new(buffer.inner),
        );

        let mut buf = [0u8; 12];
        reader.read_exact(&mut buf).unwrap();

        assert_eq!(&buf, b"initial data");
    }

    #[test]
    fn test_shared_buffer_len() {
        let buffer = SharedBuffer::with_initial(b"hello");
        assert_eq!(buffer.len(), 5);

        let (mut writer, _) = SharedBuffer::split();
        assert_eq!(writer.inner.lock().unwrap().len(), 0);

        writer.write_all(b"test").unwrap();
        assert_eq!(writer.inner.lock().unwrap().len(), 4);
    }

    #[test]
    fn test_shared_buffer_threaded() {
        use std::thread;

        let (mut writer, mut reader) = SharedBuffer::split();

        let handle = thread::spawn(move || {
            writer.write_all(b"from thread").unwrap();
        });

        thread::sleep(std::time::Duration::from_millis(10));

        let mut buf = [0u8; 11];
        reader.read_exact(&mut buf).unwrap();

        handle.join().unwrap();

        assert_eq!(&buf, b"from thread");
    }

    // ========================================================================
    // Tests for into_buffered_stream() - SharedByteBufferStream integration
    // ========================================================================

    #[test]
    fn test_into_buffered_stream_read() {
        let (mut writer, reader) = SharedBuffer::split();
        writer.write_all(b"hello world").unwrap();

        let mut stream = reader.into_buffered_stream();

        let mut buf = [0u8; 11];
        let bytes_read = stream.read(&mut buf).unwrap();

        assert_eq!(bytes_read, 11);
        assert_eq!(&buf, b"hello world");
    }

    #[test]
    fn test_into_buffered_stream_read_line() {
        let (mut writer, reader) = SharedBuffer::split();
        writer.write_all(b"line 1\nline 2\nline 3\n").unwrap();

        let mut stream = reader.into_buffered_stream();

        let mut line = String::new();
        let bytes_read = stream.read_line(&mut line).unwrap();
        assert_eq!(bytes_read, 7); // "line 1\n"
        assert_eq!(line, "line 1\n");

        line.clear();
        let bytes_read = stream.read_line(&mut line).unwrap();
        assert_eq!(bytes_read, 7); // "line 2\n"
        assert_eq!(line, "line 2\n");

        line.clear();
        let bytes_read = stream.read_line(&mut line).unwrap();
        assert_eq!(bytes_read, 7); // "line 3\n"
        assert_eq!(line, "line 3\n");
    }

    #[test]
    fn test_into_buffered_stream_sse_format() {
        // Simulate SSE format: "data: hello\n\n"
        let (mut writer, reader) = SharedBuffer::split();
        writer.write_all(b"data: hello\n\n").unwrap();

        let mut stream = reader.into_buffered_stream();

        // Read all data
        let mut buf = Vec::new();
        stream.read_to_end(&mut buf).unwrap();

        assert_eq!(buf, b"data: hello\n\n");
    }

    #[test]
    fn test_into_buffered_stream_multiple_lines() {
        let (mut writer, reader) = SharedBuffer::split();
        writer.write_all(b"event: test\ndata: payload\n\n").unwrap();

        let mut stream = reader.into_buffered_stream();

        // Read line by line
        let mut line = String::new();

        stream.read_line(&mut line).unwrap();
        assert_eq!(line, "event: test\n");

        line.clear();
        stream.read_line(&mut line).unwrap();
        assert_eq!(line, "data: payload\n");

        line.clear();
        stream.read_line(&mut line).unwrap();
        assert_eq!(line, "\n");
    }

    #[test]
    fn test_into_buffered_stream_empty_buffer() {
        let (_, reader) = SharedBuffer::split();
        let mut stream = reader.into_buffered_stream();

        let mut buf = [0u8; 5];
        let bytes_read = stream.read(&mut buf).unwrap();

        assert_eq!(bytes_read, 0);
    }

    #[test]
    fn test_into_buffered_stream_partial_reads() {
        let (mut writer, reader) = SharedBuffer::split();
        writer.write_all(b"hello world").unwrap();

        let mut stream = reader.into_buffered_stream();

        // Read in chunks
        let mut buf = [0u8; 5];
        let bytes_read = stream.read(&mut buf).unwrap();
        assert_eq!(bytes_read, 5);
        assert_eq!(&buf, b"hello");

        let bytes_read = stream.read(&mut buf).unwrap();
        assert_eq!(bytes_read, 5);
        assert_eq!(&buf, b" worl");

        let bytes_read = stream.read(&mut buf).unwrap();
        assert_eq!(bytes_read, 1);
        assert_eq!(&buf[0..1], b"d");
    }

    #[test]
    fn test_into_buffered_stream_read_exact() {
        let (mut writer, reader) = SharedBuffer::split();
        writer.write_all(b"exact data").unwrap();

        let mut stream = reader.into_buffered_stream();

        let mut buf = [0u8; 10];
        stream.read_exact(&mut buf).unwrap();

        assert_eq!(&buf, b"exact data");
    }
}

#[cfg(test)]
mod buffered_writer_tests {
    use std::io::Cursor;

    use super::*;

    #[test]
    fn can_buffered_writer_peek() {
        let content = b"alexander_wonderbat";
        let mut reader = BufferedReader::new(BufferedWriter::new(Cursor::new(content.to_vec())));

        let mut content_to_read = vec![0; 5];
        reader
            .peek(&mut content_to_read)
            .expect("should read data correctly");

        assert_eq!(b"alexa", &content_to_read[..]);

        assert_eq!(content, reader.buffer());
    }
}