assert_batch_sizes

assert_batch_sizes(&[..]) checks how a subscriber's deliveries were cut into handler calls.

Each number is the size of one call, in call order. The mount names the largest size, batch(nonzero!(2)); the broker builds the batches, and this reads back what it built. One batch is one call, so five orders in batches of two make assert_called(3). A single-message handler is called per delivery, so three deliveries read [1, 1, 1]. 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 five_orders_make_three_calls() -> Result<(), Box<dyn Error>> {
    // The log already holds five 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..=5 {
        publisher
            .message(&Order { id })
            .to("orders")
            .publish()
            .await?;
    }
    let tb = TestApp::start(app(broker)).await?;
    tb.settle().await?;

    tb.broker::<MemoryBroker<Retaining>>()
        .subscriber("orders")
        .assert_called(3)
        .assert_batch_sizes(&[2, 2, 1])
        .settled(HandlerOutcome::ack());
    Ok(())
}

Manual

examples/manual/testing_subscriber_payloads.rs

#[tokio::test]
async fn five_orders_make_three_calls() -> Result<(), Box<dyn Error>> {
    // The log already holds five 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..=5 {
        publisher
            .message(&Order { id })
            .to("orders")
            .publish()
            .await?;
    }
    let tb = TestApp::start(app(broker)).await?;
    tb.settle().await?;

    tb.broker::<MemoryBroker<Retaining>>()
        .subscriber("orders")
        .assert_called(3)
        .assert_batch_sizes(&[2, 2, 1])
        .settled(HandlerOutcome::ack());
    Ok(())
}