batches

batches::<T>() returns what a batch handler received, one decoded list per call.

Each inner list is one handler call, in call order, so the batch boundaries that received drops stay. It decodes with DefaultCodec; batches_with(&codec) decodes with another codec. A publish through the harness runs each message to completion on its own, so it reaches a batch handler as a batch of one. Fuller batches come from a log that already holds the messages: the test fills a retaining broker, the subscription opens at the start of its log, and tb.settle() drives the replay. The settle page shows a batch built from a bare publisher. The mount names the size, batch(nonzero!(2)), so three orders arrive as two and one. batches_raw returns the same batches as bytes, and assert_batch_sizes checks only the sizes.

Example

Macros

examples/testing_subscriber_payloads.rs

#[tokio::test]
async fn the_orders_were_billed_two_at_a_time() -> Result<(), Box<dyn Error>> {
    // The log already holds three orders, so the subscription reads them as full batches.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(16)));
    let publisher = broker.publisher();
    for id in 1..=3 {
        publisher
            .message(&Order { id })
            .to("orders")
            .publish()
            .await?;
    }
    let tb = TestApp::start(app(broker)).await?;
    tb.settle().await?;

    assert_eq!(
        tb.broker::<MemoryBroker<Retaining>>()
            .subscriber("orders")
            .batches::<Order>(),
        [
            vec![Order { id: 1 }, Order { id: 2 }],
            vec![Order { id: 3 }]
        ],
    );
    Ok(())
}

Manual

examples/manual/testing_subscriber_payloads.rs

#[tokio::test]
async fn the_orders_were_billed_two_at_a_time() -> Result<(), Box<dyn Error>> {
    // The log already holds three orders, so the subscription reads them as full batches.
    let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(16)));
    let publisher = broker.publisher();
    for id in 1..=3 {
        publisher
            .message(&Order { id })
            .to("orders")
            .publish()
            .await?;
    }
    let tb = TestApp::start(app(broker)).await?;
    tb.settle().await?;

    assert_eq!(
        tb.broker::<MemoryBroker<Retaining>>()
            .subscriber("orders")
            .batches::<Order>(),
        [
            vec![Order { id: 1 }, Order { id: 2 }],
            vec![Order { id: 3 }]
        ],
    );
    Ok(())
}