base_endpoint/
stream.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4//! Plaintext byte-stream capabilities for established transports.
5//!
6//! 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.
7
8use base_scheduler::{Cancellation, OperationResult, SocketStreamReader, SocketStreamWriter};
9use std::{future::Future, io};
10
11/// A sequential input direction. A nonempty read returns zero at EOF.
12/// Independent input and output directions must be able to make progress.
13pub trait ReadableStream: Send {
14    /// Read a possibly short prefix. Empty storage does not establish EOF.
15    fn read_into(
16        &mut self,
17        buffer: &mut [u8],
18        cancellation: &Cancellation,
19    ) -> impl Future<Output = OperationResult<usize>> + Send;
20
21    /// 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.
22    fn try_read_into(&mut self, buffer: &mut [u8]) -> io::Result<usize>;
23
24    /// Retire this direction. Transport ownership and final closure are separate.
25    fn close(&mut self) -> impl Future<Output = io::Result<()>> + Send;
26}
27
28/// An ordered output direction. Accepted bytes must not be replayed after a
29/// cancelled flush; flush establishes delivery to the underlying transport.
30pub trait WritableStream: Send {
31    /// Accept a possibly short prefix, reporting progress before another wait.
32    fn write(
33        &mut self,
34        buffer: &[u8],
35        cancellation: &Cancellation,
36    ) -> impl Future<Output = OperationResult<usize>> + Send;
37
38    /// Deliver all previously accepted bytes, without closing the direction.
39    fn flush(
40        &mut self,
41        cancellation: &Cancellation,
42    ) -> impl Future<Output = OperationResult<()>> + Send;
43
44    /// Retire this direction. Flush first when delivery is required.
45    fn close(&mut self) -> impl Future<Output = io::Result<()>> + Send;
46}
47
48impl<R: SocketStreamReader> ReadableStream for R {
49    async fn read_into(
50        &mut self,
51        buffer: &mut [u8],
52        cancellation: &Cancellation,
53    ) -> OperationResult<usize> {
54        SocketStreamReader::read_into(self, buffer, cancellation).await
55    }
56    fn try_read_into(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
57        SocketStreamReader::try_read_into(self, buffer)
58    }
59    async fn close(&mut self) -> io::Result<()> {
60        SocketStreamReader::close(self).await
61    }
62}
63
64impl<W: SocketStreamWriter> WritableStream for W {
65    async fn write(
66        &mut self,
67        buffer: &[u8],
68        cancellation: &Cancellation,
69    ) -> OperationResult<usize> {
70        SocketStreamWriter::write(self, buffer, cancellation).await
71    }
72    async fn flush(&mut self, cancellation: &Cancellation) -> OperationResult<()> {
73        SocketStreamWriter::flush(self, cancellation).await
74    }
75    async fn close(&mut self) -> io::Result<()> {
76        SocketStreamWriter::close(self).await
77    }
78}
base_endpoint/stream.rs

base_endpoint/
stream.rs
