Streaming responses

The Response enum models several reply styles: a single frame, a vector of frames, a streamed response, a channel-backed multi-packet response, or an empty reply. into_stream converts any variant into a boxed FrameStream, ready to install on a connection actor with set_response so streaming output can be interleaved with push traffic. WireframeError distinguishes transport failures from protocol-level errors emitted by streaming responses.[^34][^35][^31]

When constructing imperative streams, the async-stream crate integrates smoothly. The example below yields five frames and converts them into a Response::Stream value:[^36]

use async_stream::try_stream;
use wireframe::response::Response;

#[derive(bincode::Encode, bincode::BorrowDecode, Debug, PartialEq)]
struct Frame(u32);

fn stream_response() -> Response<Frame> {
    let frames = try_stream! {
        for value in 0..5 {
            yield Frame(value);
        }
    };
    Response::Stream(Box::pin(frames))
}

Response::MultiPacket exposes a different surface: callers hand ownership of the receiving half of a tokio::sync::mpsc channel to the connection actor, retain the sender, and push frames whenever back-pressure allows. The Response::with_channel helper constructs the pair and returns the sender alongside a Response::MultiPacket, making the ergonomic tuple pattern documented in ADR 0001 trivial to adopt. The library awaits channel capacity before accepting each frame, so producers can rely on the send().await future to coordinate flow control with the peer.

use tokio::spawn;
use wireframe::response::{Frame, Response};

async fn multi_packet() -> (tokio::sync::mpsc::Sender<Frame>, Response<Frame>) {
    let (sender, response) = Response::with_channel(16);

    spawn({
        let mut producer = sender.clone();
        async move {
            // Back-pressure: `send` awaits whenever the channel is full.
            for chunk in [b"alpha".as_slice(), b"beta".as_slice()] {
                let frame = Frame::from(chunk);
                if producer.send(frame).await.is_err() {
                    return; // connection dropped; stop producing
                }
            }

            // Dropping the last sender releases the channel and triggers the
            // terminator. Explicitly drop when the stream should end.
            drop(producer);
        }
    });

    (sender, response)
}

Multi-packet responders rely on the protocol hook stream_end_frame to emit a terminator when the producer side of the channel closes naturally. The connection actor records why the channel ended (drained, disconnected, or shutdown), stamps the stored correlation_id on the terminator frame, and routes it through the standard before_send instrumentation so telemetry and higher-level lifecycle hooks observe a consistent end-of-stream signal. Dropping all senders closes the channel; the actor logs the termination reason and forwards the terminator through the same hooks used for regular frames so existing observability continues to work.