Live

TestApp::start_live runs the same test body against a running NATS server, for what only a server has.

It connects the broker through its ordinary connect, so startup dials the server and every JetStream subscription looks up its stream there. 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 mode is for what the in-process mode leaves to the server: a stream's own subject list and limits, a stream the server lacks failing startup, max_ack_pending, authentication and TLS, and the timing of the connect. A stream is the server's state, and the server keeps what earlier runs left in it. So a test creates the stream through an admin connection's jetstream before the app starts, replacing a leftover, and deletes it afterwards.

In the example, live::url returns the server's address from NATS_TEST_URL, and the test returns early without one, so cargo test passes on a machine with no server. A job that stood a server 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, a NATS server with JetStream that just brokers-up starts.

Example

The two tests share one body: one in process, one against the server at NATS_TEST_URL.

Macros
Manual

examples/nats_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::<NatsBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<NatsBroker>()
        .subscriber("testing-live.payments")
        .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("nats://localhost:4222")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("NATS_TEST_URL") else {
        return Ok(());
    };
    // The stream is the server's state: created before the app starts, replacing what an
    // interrupted run left, and deleted afterwards.
    let admin = NatsBroker::new(url.as_str()).connect().await?;
    let jetstream = admin.jetstream();
    let _ = jetstream.delete_stream(STREAM).await;
    jetstream
        .create_stream(StreamConfig {
            name: STREAM.to_owned(),
            subjects: vec!["testing-live.payments".to_owned()],
            ..StreamConfig::default()
        })
        .await?;

    let outcome =
        async { accepts_a_payment(TestApp::start_live(app(&url)).await?).await }.await;

    jetstream.delete_stream(STREAM).await?;
    admin.shutdown().await?;
    outcome
}

examples/manual/nats_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::<NatsBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<NatsBroker>()
        .subscriber("testing-live.payments")
        .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("nats://localhost:4222")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("NATS_TEST_URL") else {
        return Ok(());
    };
    // The stream is the server's state: created before the app starts, replacing what an
    // interrupted run left, and deleted afterwards.
    let admin = NatsBroker::new(url.as_str()).connect().await?;
    let jetstream = admin.jetstream();
    let _ = jetstream.delete_stream(STREAM).await;
    jetstream
        .create_stream(StreamConfig {
            name: STREAM.to_owned(),
            subjects: vec!["testing-live.payments".to_owned()],
            ..StreamConfig::default()
        })
        .await?;

    let outcome =
        async { accepts_a_payment(TestApp::start_live(app(&url)).await?).await }.await;

    jetstream.delete_stream(STREAM).await?;
    admin.shutdown().await?;
    outcome
}

Example​

Example