Live

TestApp::start_live runs the same test body against the Pub/Sub emulator, for leases, flow control and ordered delivery.

It connects the broker through its ordinary connect, so every subscription opens a real streaming pull. The body is the one an in-process test runs; only the start call differs. A publish, settle and advance wait on the real clock until every subscription has handled what was published to it. The wait lasts ten seconds at most, and start_live_within sets another bound.

The live app is the production one on a broker pointed at the emulator. The emulator starts empty, so its subscriptions create themselves with create_with_topic. It keeps its subscriptions until it stops, so a run that failed can leave messages behind for the next one. The examples delete their subscriptions before each live run, which live::fresh does with the client library's admin API.

In the example, live::host returns the emulator's address from PUBSUB_TEST_HOST, and the test returns early without one, so cargo test passes on a machine with no emulator. A job that stood an emulator up sets RUSTSTREAM_REQUIRE_LIVE=1, and a missing address then fails the test instead of skipping it. The repository's stand is docker-compose.test.yml, which just brokers-up starts.

Example

The two tests share one body: one in process, one against the emulator at PUBSUB_TEST_HOST.

Macros
Manual

examples/pubsub_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<PubSubBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<PubSubBroker>()
        .subscriber("testing-live-payments-workers")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app()).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_emulator() -> Result<(), Box<dyn Error>> {
    let Some(host) = live::host() else {
        return Ok(());
    };
    live::fresh(&host, &["testing-live-payments-workers"]).await?;
    let broker = PubSubBroker::new("my-project").emulator(host);
    accepts_a_payment(TestApp::start_live(service(broker)).await?).await
}

examples/manual/pubsub_testing_live.rs
/// The body both modes run; only the start call differs.
async fn accepts_a_payment(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    tb.broker::<PubSubBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<PubSubBroker>()
        .subscriber("testing-live-payments-workers")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    accepts_a_payment(TestApp::start(app()).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_emulator() -> Result<(), Box<dyn Error>> {
    let Some(host) = live::host() else {
        return Ok(());
    };
    live::fresh(&host, &["testing-live-payments-workers"]).await?;
    let broker = PubSubBroker::new("my-project").emulator(host);
    accepts_a_payment(TestApp::start_live(service(broker)).await?).await
}

Example​

Example