base-endpoint 0.2.0

// Released under the MIT License.
// Copyright, 2026, by Samuel Williams.

//! Plaintext byte-stream capabilities for established transports.
//!
//! These contracts describe one direction, independently of whether the transport encrypts its bytes. Scheduler stream handles forward directly into them. Packet transports keep their message boundaries and datagram metadata in separate packet capabilities rather than flattening packets into this interface, so not every endpoint connection is a stream.

use base_scheduler::{Cancellation, OperationResult, SocketStreamReader, SocketStreamWriter};
use std::{future::Future, io};

/// A sequential input direction. A nonempty read returns zero at EOF.
/// Independent input and output directions must be able to make progress.
pub trait ReadableStream: Send {
    /// Read a possibly short prefix. Empty storage does not establish EOF.
    fn read_into(
        &mut self,
        buffer: &mut [u8],
        cancellation: &Cancellation,
    ) -> impl Future<Output = OperationResult<usize>> + Send;

    /// Attempt one read now, without waiting. Unlike polling [`read_into`](Self::read_into) once, this reaches the transport, so it observes input, EOF and errors already queued there. `WouldBlock` means no input is available now; zero with nonempty storage is EOF, and empty storage returns zero. Other errors are as for [`read_into`](Self::read_into), except that a cancelled transport reports `ConnectionAborted`, since there is no operation cancellation to report and the input cannot resume. Wrappers must forward this rather than polling a read.
    fn try_read_into(&mut self, buffer: &mut [u8]) -> io::Result<usize>;

    /// Retire this direction. Transport ownership and final closure are separate.
    fn close(&mut self) -> impl Future<Output = io::Result<()>> + Send;
}

/// An ordered output direction. Accepted bytes must not be replayed after a
/// cancelled flush; flush establishes delivery to the underlying transport.
pub trait WritableStream: Send {
    /// Accept a possibly short prefix, reporting progress before another wait.
    fn write(
        &mut self,
        buffer: &[u8],
        cancellation: &Cancellation,
    ) -> impl Future<Output = OperationResult<usize>> + Send;

    /// Deliver all previously accepted bytes, without closing the direction.
    fn flush(
        &mut self,
        cancellation: &Cancellation,
    ) -> impl Future<Output = OperationResult<()>> + Send;

    /// Retire this direction. Flush first when delivery is required.
    fn close(&mut self) -> impl Future<Output = io::Result<()>> + Send;
}

impl<R: SocketStreamReader> ReadableStream for R {
    async fn read_into(
        &mut self,
        buffer: &mut [u8],
        cancellation: &Cancellation,
    ) -> OperationResult<usize> {
        SocketStreamReader::read_into(self, buffer, cancellation).await
    }
    fn try_read_into(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
        SocketStreamReader::try_read_into(self, buffer)
    }
    async fn close(&mut self) -> io::Result<()> {
        SocketStreamReader::close(self).await
    }
}

impl<W: SocketStreamWriter> WritableStream for W {
    async fn write(
        &mut self,
        buffer: &[u8],
        cancellation: &Cancellation,
    ) -> OperationResult<usize> {
        SocketStreamWriter::write(self, buffer, cancellation).await
    }
    async fn flush(&mut self, cancellation: &Cancellation) -> OperationResult<()> {
        SocketStreamWriter::flush(self, cancellation).await
    }
    async fn close(&mut self) -> io::Result<()> {
        SocketStreamWriter::close(self).await
    }
}