Struct StdioSource 

pub struct StdioSource<R> { /* private fields */ }

A FrameSource over any AsyncRead.

Examples

A frame split across two reads is reassembled, because the partial Ok(None) from the decoder is a reason to read again rather than an answer:

use std::pin::Pin;
use std::task::{Context, Poll};

use protocol_droid::adapters::primary::stdio::{StdioConfig, StdioError, StdioSource};
use protocol_droid::application::jsonrpc::FrameSource;
use protocol_droid::domain::MaxFrameSize;
use tokio::io::{AsyncRead, ReadBuf};

/// Hands over at most three bytes per read, so a frame cannot arrive whole.
struct Trickle {
    data: Vec<u8>,
    served: usize,
}

impl AsyncRead for Trickle {
    fn poll_read(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let end = (self.served + 3).min(self.data.len());
        let Some(chunk) = self.data.get(self.served..end) else {
            return Poll::Ready(Ok(())); // end of stream
        };
        buf.put_slice(chunk);
        self.served = end;
        Poll::Ready(Ok(()))
    }
}

let stream = Trickle { data: b"{\"jsonrpc\":\"2.0\"}\n".to_vec(), served: 0 };
let source = StdioSource::new(stream, StdioConfig::new(MaxFrameSize::new(1024)?));

assert_eq!(source.next_frame().await?, Some(r#"{"jsonrpc":"2.0"}"#.to_owned()));

Implementations

impl<R> StdioSource<R>

pub fn new(stream: R, config: StdioConfig) -> Self

Wraps a stream, decoding frames at the configured ceiling.

pub const fn config(&self) -> StdioConfig

The settings this source was built with.

pub const fn limit(&self) -> MaxFrameSize

The ceiling this source refuses frames above.

impl<R> StdioSource<R>

where R: AsyncRead + Unpin,

pub async fn next_frame(&self) -> Result<Option<String>, SourceError>

Reads the peer’s next frame, waiting for as many reads as that takes.

This is the read loop, and the reason this type exists. Every exit from it is a decision, and the decoder’s None is not one of them.

Errors

SourceError::Transport when a read fails, which is latched, and SourceError::Framing when the peer’s bytes are not a frame — most often crate::domain::FramingError::FrameTooLarge, or bytes that are not UTF-8.

Ok(None) is the peer closing and only that.

The lock is held for the whole loop, which is the point of holding one: two callers must not interleave a read with the decoder that consumes what it returned. Every exit from the loop is still holding it, so the guard has no earlier last use to be tightened to.

Trait Implementations

impl<R: Debug> Debug for StdioSource<R>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<R> FrameSource for StdioSource<R>

where R: AsyncRead + Unpin + Send,

fn next_frame( &self, ) -> impl Future<Output = Result<Option<String>, SourceError>> + Send

Waits for the peer’s next frame.

The port’s Ok(None) is the peer closing, so this never returns the decoder’s “no terminator yet” — see the module docs for why the two Nones are opposites and what inverting them looks like from the outside.

A framing error is not latched: the decoder refused the bytes and kept its own buffer, so the next call is refused again by the same check rather than by a flag, and the stream stays open for the caller to shut down.

Errors

As StdioSource::next_frame.