Switch

A test build leaves the outbox off, so a test needs no outbox table; RUSTSTREAM_SQLX_OUTBOX=on turns it on for the whole test process.

A build with the testing feature leaves the outbox off: the two middlewares pass every message untouched, wrap sends straight to the publisher it wraps, and the republish sends nothing. A test of the handlers then runs on the service's production app with no outbox table and no database for it. A production build always tracks, and the switch is compiled out.

RUSTSTREAM_SQLX_OUTBOX=on in the environment turns the outbox on in a test build, and any other value leaves it off. The variable is read once per process, the first time the outbox runs, and tests share one process, so it is set outside them: in a just recipe, a CI job or a nextest profile. A test of tracking itself runs only with the switch on and returns early without it; a job that sets RUSTSTREAM_REQUIRE_LIVE then fails such a test instead of passing it. The testing feature also reaches cargo run --example and cargo build --all-targets of a crate that enables it, so an example among the outbox's pages runs with the switch set.

Example

The test drives the service of the Transactional outbox page over a database that has no tables, and runs only without the switch: the order reaches its consumer and carries no record id.

Macros

examples/sqlx_outbox_switch.rs

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_test_of_the_handlers_needs_no_outbox_table() -> Result<(), Box<dyn Error>> {
    // The switch is read once per process: with it on, this process tracks every message.
    if env::var("RUSTSTREAM_SQLX_OUTBOX").as_deref() == Ok("on") {
        return Ok(());
    }
    let (pool, _keeper) = database("sqlx_outbox_switch").await?;
    let tb = TestApp::start(service(pool)).await?;
    tb.broker::<MemoryBroker>()
        .message(&PlaceOrder { id: 7 })
        .publish()
        .await?;

    tb.broker::<MemoryBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&OrderPlaced { id: 7 })
        .settled(HandlerOutcome::ack());
    // Nothing was recorded, so the order carries no record id.
    let published = tb.broker::<MemoryBroker>().published::<()>("orders");
    let ids: Vec<_> = published
        .messages()
        .iter()
        .map(|message| message.headers().get_str(OUTBOX_ID_HEADER))
        .collect();
    assert_eq!(ids, [None]);
    Ok(())
}

Manual

examples/manual/sqlx_outbox_switch.rs

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_test_of_the_handlers_needs_no_outbox_table() -> Result<(), Box<dyn Error>> {
    // The switch is read once per process: with it on, this process tracks every message.
    if env::var("RUSTSTREAM_SQLX_OUTBOX").as_deref() == Ok("on") {
        return Ok(());
    }
    let (pool, _keeper) = database("manual_sqlx_outbox_switch").await?;
    let tb = TestApp::start(service(pool)).await?;
    tb.broker::<MemoryBroker>()
        .message(&PlaceOrder { id: 7 })
        .publish()
        .await?;

    tb.broker::<MemoryBroker>()
        .subscriber("orders")
        .assert_called_once()
        .with(&OrderPlaced { id: 7 })
        .settled(HandlerOutcome::ack());
    // Nothing was recorded, so the order carries no record id.
    let published = tb.broker::<MemoryBroker>().published::<()>("orders");
    let ids: Vec<_> = published
        .messages()
        .iter()
        .map(|message| message.headers().get_str(OUTBOX_ID_HEADER))
        .collect();
    assert_eq!(ids, [None]);
    Ok(())
}