Function next_matching 

pub async fn next_matching(
    scenario: &str,
    rx: &mut Receiver<Envelope>,
    pred: impl Fn(&Envelope) -> bool,
) -> Envelope

Reads envelopes from rx until pred accepts one, and returns that one.

Envelopes that do not match are read and thrown away, so they are not seen by later calls. Use a precise pred (the event kind and, when several subscriptions run, the subscription id). scenario is only the name shown in the failure message.

The whole wait is limited to EVENT_TIMEOUT, not each envelope.

Panics

With [conformance:<scenario>] ... when the channel is closed before a matching envelope arrives (event channel closed before the expected event), and when none arrives within EVENT_TIMEOUT (no expected event within 8s).

Examples

use brokapi_conformance::next_matching;
use brokapi_core::{Envelope, Event, Timestamps, UnixNanos};
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(4);
let ts = Timestamps::new(UnixNanos::ZERO, std::time::Instant::now());
tx.send(Envelope::new("demo".into(), ts, Event::Balances(vec![]))).await.unwrap();
tx.send(Envelope::new("demo".into(), ts, Event::Positions(vec![]))).await.unwrap();

// The balances envelope is skipped, the positions envelope is returned.
let env = next_matching("my_scenario", &mut rx, |e| matches!(e.payload, Event::Positions(_))).await;
assert!(matches!(env.payload, Event::Positions(_)));

When nothing matching can arrive, the scenario fails with its name in the message:

use brokapi_conformance::next_matching;
use brokapi_core::{Envelope, Event};
use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<Envelope>(1);
drop(tx); // the connector went away
// panics: [conformance:my_scenario] event channel closed before the expected event
next_matching("my_scenario", &mut rx, |_| true).await;