Module frame_source 

The inbound edge: a stream of whole frames from a peer.

Why this trait exists

FrameSink already covers the outbound half, and it lives in crate::application::session because the session engine is what consumes it. The inbound half has a different consumer — the loop that reads a frame and routes it — and by the project’s rule that a port trait belongs to the consumer, that puts this one here rather than beside its sibling. Two transports facing opposite directions are not one capability, and a shared ports module would have implied they were.

It is deliberately not a Transport trait. A trait that both read and wrote would force every implementor to answer for both, which is exactly the “three adapters wearing one name” objection that FrameSink is documented against. The stdio server writes to a child it spawned and reads from the same pipe; the bridge reads from an HTTP body and writes to stdio; a replay source only reads at all.

What it hides

Chunking. An implementor is given no way to express “I have half a message”, and cannot: whatever a read returns goes into a FrameDecoder, and a frame comes out when it is complete. A caller of this trait never sees a partial line, never has to know that a read boundary can fall inside a UTF-8 character, and never has to know what the ceiling is — although the implementor does, and is the one that must set it.

The distinction this trait has to be careful about

Ok(None) means the peer closed, and nothing else. It does not mean “no frame is ready”. The decoder underneath answers None for exactly that second thing, so the two meanings are opposites and an implementor that forwards the decoder’s answer straight up has inverted them. An implementor awaits more bytes instead; this method resolves only on a complete frame, an end of stream, or a failure, and a caller can therefore treat None as “the session is over” without also having to ask whether anything was pending.

Two traits, because one of them cannot be both

FrameSource is a native async fn, which is a return-position impl trait and therefore not object-safe. A dispatch loop that is generic over its source takes impl FrameSource and the call monomorphises to nothing. The CLI bridge, which holds a source of an unknown concrete type behind one handle, needs Arc<dyn DynFrameSource> and cannot have it from this trait.

DynFrameSource is the object-safe spelling, written out by hand. A blanket impl with async-trait would put a boxed future on the generic path too, and one allocation per inbound frame is what the generic form exists to avoid. The box is confined to the callers that asked for it.

Examples

use std::sync::Arc;

use protocol_droid::application::jsonrpc::{DynFrameSource, FrameSource, SourceError};
use protocol_droid::domain::{FrameDecoder, MaxFrameSize};

/// A source backed by a fixed buffer: everything a read would have produced.
struct Replay {
    decoder: std::sync::Mutex<FrameDecoder>,
}

impl Replay {
    fn new(wire: &[u8]) -> Self {
        let mut decoder = FrameDecoder::new(MaxFrameSize::DEFAULT);
        // A test's fixture is trusted, so a refusal here is the fixture's
        // problem and is ignored rather than unwrapped. A real implementor
        // would have nowhere to put the error either — it has no frame to
        // report it against — and would surface it on the first `next_frame`.
        let _ = decoder.push(wire);
        Self {
            decoder: std::sync::Mutex::new(decoder),
        }
    }
}

impl FrameSource for Replay {
    async fn next_frame(&self) -> Result<Option<String>, SourceError> {
        let mut decoder = self.decoder.lock().map_err(|_| SourceError::Transport)?;
        decoder
            .next_frame()
            .map(|frame| frame.map(str::to_owned))
            .map_err(SourceError::from)
    }
}

let runtime = tokio::runtime::Builder::new_current_thread()
    .build()
    .map_err(|error| error.to_string())?;

// The generic form: the dispatch loop is written once and monomorphises.
let generic = Replay::new(b"{\"jsonrpc\":\"2.0\"}\n");
assert_eq!(
    runtime.block_on(FrameSource::next_frame(&generic)),
    Ok(Some(String::from("{\"jsonrpc\":\"2.0\"}")))
);

// The boxed form: the same implementation, behind a trait object.
let boxed: Arc<dyn DynFrameSource> = Arc::new(Replay::new(b"{\"id\":1}\n"));
assert_eq!(
    runtime.block_on(boxed.next_frame()),
    Ok(Some(String::from("{\"id\":1}")))
);

Enums

SourceError

Why the next inbound frame did not arrive.

Traits

DynFrameSource

A FrameSource that can sit behind a trait object.

FrameSource

An inbound frame stream from a peer.