Live

TestApp::start_live runs the same test body against a running broker, for what only a server has: credit, dead-letter policy, grouping.

It connects the broker through its ordinary connect, so the socket, the login and every subscription's link are real. 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 covers what the in-process transport leaves to the server: link credit, the broker's dead-letter policy behind a rejection and its own delivery limit, message groups by partition key, the fan-out a broker gives a raw address, how a broker creates an address a publish reaches first, and the login. A broker keeps what earlier runs left on it. Each run reads addresses of its own, named with the process id, so it never takes a message an earlier run left behind.

In the example, live::url returns the broker's URL from AMQP_TEST_URL, and the test returns early without one, so cargo test passes on a machine with no broker. A job that stood a broker up sets RUSTSTREAM_REQUIRE_LIVE=1, and a missing URL then fails the test instead of skipping it. The repository's stand is ActiveMQ Artemis in docker-compose.test.yml, which just brokers-up starts on localhost:5672 with the user artemis.

Example

The two tests share one body: one in process, one against the broker at AMQP_TEST_URL.

Macros
Manual

examples/amqp_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::<AmqpBroker>()
        .message(&Payment { amount: 100 })
        .to(payments().as_str())
        .publish()
        .await?;

    tb.broker::<AmqpBroker>()
        .subscriber(&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("amqp://broker.internal:5672")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("AMQP_TEST_URL") else {
        return Ok(());
    };
    accepts_a_payment(TestApp::start_live(app(&url)).await?).await
}

examples/manual/amqp_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::<AmqpBroker>()
        .message(&Payment { amount: 100 })
        .to(payments().as_str())
        .publish()
        .await?;

    tb.broker::<AmqpBroker>()
        .subscriber(&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("amqp://broker.internal:5672")).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("AMQP_TEST_URL") else {
        return Ok(());
    };
    accepts_a_payment(TestApp::start_live(app(&url)).await?).await
}

Example​

Example