Consuming streaming responses on the client

The client library supports consuming Response::Stream and Response::MultiPacket server responses through the call_streaming and receive_streaming methods on WireframeClient. These methods return a ResponseStream that yields typed data frames until the protocol's end-of-stream terminator arrives. ResponseStream holds an exclusive mutable borrow of WireframeClient while it exists, so the client cannot perform other I/O operations until the stream is dropped or fully consumed (None is observed).

For protocols where every frame is meaningful, a manual StreamExt::next loop remains the clearest option. For multiplexed protocols that interleave data frames with notices, progress updates, or other control frames, prefer StreamingResponseExt::typed_with. It maps each frame into Result<Option<Item>, ClientError>, yielding Some(item) values in order while silently skipping None.

Protocol implementations must override Packet::is_stream_terminator to teach the client how to recognize the server's end-of-stream marker. The default implementation returns false; protocols override it to match their terminator format:

use wireframe::app::{Packet, PacketParts};
use wireframe::correlation::CorrelatableFrame;

#[derive(bincode::BorrowDecode, bincode::Encode)]
struct MyEnvelope {
    id: u32,
    correlation_id: Option<u64>,
    payload: Vec<u8>,
}

impl CorrelatableFrame for MyEnvelope {
    fn correlation_id(&self) -> Option<u64> { self.correlation_id }
    fn set_correlation_id(&mut self, cid: Option<u64>) {
        self.correlation_id = cid;
    }
}

// Message is auto-implemented via the blanket impl for Encode + BorrowDecode types.

impl Packet for MyEnvelope {
    fn id(&self) -> u32 { self.id }
    fn into_parts(self) -> PacketParts {
        PacketParts::new(self.id, self.correlation_id, self.payload)
    }
    fn from_parts(parts: PacketParts) -> Self {
        Self {
            id: parts.id(),
            correlation_id: parts.correlation_id(),
            payload: parts.into_payload(),
        }
    }
    // An envelope with id == 0 signals end-of-stream.
    fn is_stream_terminator(&self) -> bool { self.id == 0 }
}

This mirrors the server's stream_end_frame hook symmetrically: the server produces terminators; the client detects them.[^49]

call_streaming sends a request, auto-generates a correlation identifier if needed, and returns a ResponseStream. The stream yields data frames as Result<P, ClientError> and terminates with None when the terminator arrives: because the stream borrows the WireframeClient mutably, this call also prevents other client I/O until the stream is dropped or drained.

use futures::StreamExt;
use std::net::SocketAddr;
use wireframe::{app::Envelope, client::{ClientError, WireframeClient}};

# #[tokio::main]
# async fn main() -> Result<(), ClientError> {
let addr: SocketAddr = "127.0.0.1:9000".parse().expect("valid address");
let mut client = WireframeClient::builder().connect(addr).await?;

let request = Envelope::new(1, None, vec![]);
let mut stream = client.call_streaming::<Envelope>(request).await?;

while let Some(result) = stream.next().await {
    let frame = result?;
    println!("received: {:?}", frame);
}
// Stream terminated — all data frames consumed.
# Ok(())
# }

StreamingResponseExt::typed_with adapts a ResponseStream into a stream of domain items. This is useful when only some protocol frames should be exposed to the caller:

use futures::TryStreamExt;
use std::net::SocketAddr;
use wireframe::{
    app::{Packet, PacketParts},
    client::{ClientError, StreamingResponseExt, WireframeClient},
    correlation::CorrelatableFrame,
};

#[derive(bincode::BorrowDecode, bincode::Encode)]
struct MyEnvelope {
    id: u32,
    correlation_id: Option<u64>,
    payload: Vec<u8>,
}

impl CorrelatableFrame for MyEnvelope {
    fn correlation_id(&self) -> Option<u64> { self.correlation_id }
    fn set_correlation_id(&mut self, cid: Option<u64>) {
        self.correlation_id = cid;
    }
}

impl Packet for MyEnvelope {
    fn id(&self) -> u32 { self.id }
    fn into_parts(self) -> PacketParts {
        PacketParts::new(self.id, self.correlation_id, self.payload)
    }
    fn from_parts(parts: PacketParts) -> Self {
        Self {
            id: parts.id(),
            correlation_id: parts.correlation_id(),
            payload: parts.into_payload(),
        }
    }
    fn is_stream_terminator(&self) -> bool { self.id == 0 }
}

#[derive(Debug, PartialEq, Eq)]
struct Row(Vec<u8>);

# #[tokio::main]
# async fn main() -> Result<(), ClientError> {
let addr: SocketAddr = "127.0.0.1:9000".parse().expect("valid address");
let mut client = WireframeClient::builder().connect(addr).await?;

let request = MyEnvelope {
    id: 1,
    correlation_id: None,
    payload: vec![],
};

let rows: Vec<Row> = client
    .call_streaming::<MyEnvelope>(request)
    .await?
    .typed_with(|frame| match frame.id {
        1 => Ok(Some(Row(frame.payload))),
        2 => Ok(None),
        other => Err(ClientError::from(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("unexpected frame id {other}"),
        ))),
    })
    .try_collect()
    .await?;

println!("received {} rows", rows.len());
# Ok(())
# }

Because the helper wraps ResponseStream rather than replacing it, transport failures and correlation mismatches still surface as ClientError values from the underlying stream. The helper also does not change the exclusive &mut WireframeClient borrow: while the typed stream is alive, other client I/O remains blocked.

When the caller has already sent a request via send_envelope, the receive_streaming method accepts the known correlation identifier and returns a ResponseStream without re-sending. As with call_streaming, that stream holds an exclusive mutable borrow of WireframeClient until dropped or fully consumed:

use futures::StreamExt;
use std::net::SocketAddr;
use wireframe::{app::Envelope, client::{ClientError, WireframeClient}};

# #[tokio::main]
# async fn main() -> Result<(), ClientError> {
let addr: SocketAddr = "127.0.0.1:9000".parse().expect("valid address");
let mut client = WireframeClient::builder().connect(addr).await?;

let request = Envelope::new(1, None, vec![]);
let cid = client.send_envelope(request).await?;
let mut stream = client.receive_streaming::<Envelope>(cid);

while let Some(result) = stream.next().await {
    let frame = result?;
    println!("received: {:?}", frame);
}
# Ok(())
# }

Back-pressure propagates naturally through Transmission Control Protocol (TCP) flow control. If the client reads slowly, the client's TCP receive buffer fills. As that receive window shrinks, the server's TCP send buffer fills because bytes cannot be flushed quickly enough. Once the send buffer is full, write operations suspend, and the server stops polling its response stream or channel until the client drains data. No explicit flow-control messages are required.[^50]

Interleaved high- and low-priority push behaviour is validated against this streaming path as part of roadmap item 11.3.2. The parity suite confirms fairness-driven low-priority progress and shared cross-priority rate limiting without changing the public WireframeClient interface.

ResponseStream validates the correlation identifier on every frame. If a frame carries a different identifier, the stream yields ClientError::StreamCorrelationMismatch and terminates. Transport errors and decode failures are surfaced through ClientError::Wireframe.[^51]

• Codec length mismatch: if small requests succeed but larger ones fail once payload size crosses a threshold, suspect a frame-budget mismatch between client and server. The client usually reports ClientError::Wireframe(WireframeError::Io(_)) because the peer closes the connection after rejecting the oversized frame. Check the client ClientCodecConfig::max_frame_length, the server buffer_capacity, and any server-side frame-limit logs together. The fix is to align both ends to the same maximum frame length.
• Preamble timeout: ClientError::PreambleTimeout means the handshake stalled before framed traffic began. Reduce ambiguity by setting an explicit preamble_timeout(Duration) and verify that the server actually reads or replies during the preamble phase.
• Preamble read or decode failure: ClientError::PreambleRead(_) means the success callback could not read or decode the server's acknowledgement bytes. This usually indicates the wrong preamble type, malformed server bytes, or callback logic that reads the response incorrectly. Confirm both sides use the same preamble schema and that on_preamble_success returns any leftover bytes it consumed.
• Preamble encode or write failure: ClientError::PreambleEncode(_) means the client failed before handshake completion while serializing or writing the preamble. In the current client flow, write_preamble(...) wraps write-side I/O in bincode::EncodeError, so there is no separate user-visible PreambleWrite branch to match against.
• TLS or wrong-protocol port mismatch: pointing a plain WireframeClient at a port that expects Transport Layer Security (TLS), Hypertext Transfer Protocol (HTTP), or another protocol usually surfaces as ClientError::Wireframe(WireframeError::Io(_)) after the first request because the bytes returned are not valid Wireframe frames. Verify the host and port, confirm whether a TLS terminator is required, and remember that built-in client TLS configuration is still future work.
• Correlation mismatch errors: ClientError::CorrelationMismatch and ClientError::StreamCorrelationMismatch mean the response did not preserve the request correlation identifier. Verify that the server echoes or stamps the expected correlation_id for call_correlated and streaming responses.
• Streaming API contention: ResponseStream holds &mut WireframeClient; do not issue additional client I/O until the stream is drained or dropped.
• Transport disconnects: treat ClientError::Wireframe(WireframeError::Io(_)) as a network-level or peer-closure failure and apply a reconnect or retry policy at the call site.