received_values

received_values::<T>() returns every value a subscriber's deliveries lent its handler, batch elements included.

It is the counterpart of received for a broker whose deliveries carry a value (Carries<T>) instead of bytes to decode. The list is flat and in delivery order. It panics when a delivery lent no T. The row subscription here is the in-memory stand-in described on the with_value page.

Example

Macros

examples/testing_subscriber_carried.rs

#[tokio::test]
async fn the_handler_got_every_row() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let broker = tb.broker::<MemoryBroker>();
    for row in ["7:alice", "8:bob"] {
        broker
            .message(&Wire(row.as_bytes().to_vec()))
            .to("users")
            .publish()
            .await?;
    }

    assert_eq!(
        broker.subscriber("users").received_values::<Row>(),
        [Row::new(7, "alice"), Row::new(8, "bob")],
    );
    Ok(())
}

Manual

examples/manual/testing_subscriber_carried.rs

#[tokio::test]
async fn the_handler_got_every_row() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let broker = tb.broker::<MemoryBroker>();
    for row in ["7:alice", "8:bob"] {
        broker
            .message(&Wire(row.as_bytes().to_vec()))
            .to("users")
            .publish()
            .await?;
    }

    assert_eq!(
        broker.subscriber("users").received_values::<Row>(),
        [Row::new(7, "alice"), Row::new(8, "bob")],
    );
    Ok(())
}