batches_raw

batches_raw() returns the bytes a batch handler received, one list per call.

It is received_raw with the batch boundaries kept. A single-message handler yields lists of one. Here the handler takes raw frames, &[Frame<'_>], so no codec runs on the way in or in the check. The batches come from a replayed log, as on the batches page. A batch built from a bare publisher is driven with settle.

Example

Macros

examples/testing_subscriber_payloads.rs

#[tokio::test]
async fn the_frames_were_cut_in_pairs() -> Result<(), Box<dyn Error>> {
    // The log already holds three frames, so the subscription reads them as full batches.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(16)));
    let publisher = broker.publisher();
    for frame in [b"\x01", b"\x02", b"\x03"] {
        publisher
            .message(&Wire(frame.to_vec()))
            .to("frames")
            .publish()
            .await?;
    }
    let tb = TestApp::start(app(broker)).await?;
    tb.settle().await?;

    assert_eq!(
        tb.broker::<MemoryBroker<Retaining>>()
            .subscriber("frames")
            .batches_raw(),
        [vec![&b"\x01"[..], &b"\x02"[..]], vec![&b"\x03"[..]]],
    );
    Ok(())
}

Manual

examples/manual/testing_subscriber_payloads.rs

#[tokio::test]
async fn the_frames_were_cut_in_pairs() -> Result<(), Box<dyn Error>> {
    // The log already holds three frames, so the subscription reads them as full batches.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(16)));
    let publisher = broker.publisher();
    for frame in [b"\x01", b"\x02", b"\x03"] {
        publisher
            .message(&Wire(frame.to_vec()))
            .to("frames")
            .publish()
            .await?;
    }
    let tb = TestApp::start(app(broker)).await?;
    tb.settle().await?;

    assert_eq!(
        tb.broker::<MemoryBroker<Retaining>>()
            .subscriber("frames")
            .batches_raw(),
        [vec![&b"\x01"[..], &b"\x02"[..]], vec![&b"\x03"[..]]],
    );
    Ok(())
}