received_raw

received_raw() returns the bytes of every payload a subscriber received, in delivery order.

It returns one Bytes per delivery; a batch contributes each of its elements. No codec runs, so it reads payloads that are no model: here frames the handler takes as a Deserialized view. The test publishes such bytes through a #[derive(Outgoing, Serialized)] newtype, which keeps every codec off them. with_raw checks the latest payload only, and batches_raw groups the same bytes by call.

Example

Macros

examples/testing_subscriber_payloads.rs

#[tokio::test]
async fn the_frames_arrived_untouched() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let broker = tb.broker::<MemoryBroker>();
    for frame in [b"\x00\x01", b"\xff\xfe"] {
        broker
            .message(&Wire(frame.to_vec()))
            .to("frames")
            .publish()
            .await?;
    }

    assert_eq!(
        broker.subscriber("frames").received_raw(),
        [&b"\x00\x01"[..], &b"\xff\xfe"[..]],
    );
    Ok(())
}

Manual

examples/manual/testing_subscriber_payloads.rs

#[tokio::test]
async fn the_frames_arrived_untouched() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let broker = tb.broker::<MemoryBroker>();
    for frame in [b"\x00\x01", b"\xff\xfe"] {
        broker
            .message(&Wire(frame.to_vec()))
            .to("frames")
            .publish()
            .await?;
    }

    assert_eq!(
        broker.subscriber("frames").received_raw(),
        [&b"\x00\x01"[..], &b"\xff\xfe"[..]],
    );
    Ok(())
}