Live

TestApp::start_live runs the same test body against a running MQTT server, for what only a server has: sessions, the last will, flow control.

It connects the broker through its ordinary connect, so startup waits for the server's CONNACK and each subscription for its SUBACK. 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 server leaves out: the login and TLS, a persistent session, the last will, the receive window, a lost connection, and several connections sharing one server. A server is shared with whatever else runs against it. So each run connects under a client id of its own: a second connection under one id takes the session over.

In the example, live::url returns the server's address from MQTT_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 a Mosquitto container in docker-compose.test.yml, which just brokers-up starts, with a listener that asks for a password and one that speaks TLS.

Example

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

Macros
Manual

examples/mqtt_testing_live.rs
/// The body both modes run; only the start call differs.
async fn records_a_reading(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    let reading = Telemetry {
        device: "dev42".to_owned(),
        temperature: 21.5,
    };
    tb.broker::<MqttBroker>()
        .message(&reading)
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("testing-live/telemetry")
        .assert_called_once()
        .with(&reading)
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    records_a_reading(TestApp::start(app("mqtt://localhost:1883", "telemetry-svc")).await?)
        .await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("MQTT_TEST_URL") else {
        return Ok(());
    };
    // A client id of this run's own: a second connection under one id takes the session over.
    let tb = TestApp::start_live(app(&url, &live::client_id("telemetry"))).await?;
    records_a_reading(tb).await
}

examples/manual/mqtt_testing_live.rs
/// The body both modes run; only the start call differs.
async fn records_a_reading(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    let reading = Telemetry {
        device: "dev42".to_owned(),
        temperature: 21.5,
    };
    tb.broker::<MqttBroker>()
        .message(&reading)
        .publish()
        .await?;

    tb.broker::<MqttBroker>()
        .subscriber("testing-live/telemetry")
        .assert_called_once()
        .with(&reading)
        .settled(HandlerOutcome::ack());
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    records_a_reading(TestApp::start(app("mqtt://localhost:1883", "telemetry-svc")).await?)
        .await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = live::url("MQTT_TEST_URL") else {
        return Ok(());
    };
    // A client id of this run's own: a second connection under one id takes the session over.
    let tb = TestApp::start_live(app(&url, &live::client_id("telemetry"))).await?;
    records_a_reading(tb).await
}

Example​

Example