Live

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

It connects the broker through its ordinary connect, so the client dials the server and every subscription opens a real consumer. 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:

the ack timeout's own redelivery;
partitioned topics, and the partition a key picks;
which consumer of a Shared or KeyShared subscription the server picks, and KeyShared hash ranges;
messages published while a subscription has no consumer;
a second Exclusive consumer waiting, and the OperationRetries that bound the wait;
the token and TLS;
a backlog older than the test, which start_at replays.

A server keeps what earlier runs left in it, and a subscription is a durable cursor on it. So each run reads in a subscription of its own, which starts at the end of the topic, past what earlier runs published.

In the example, stand returns the server's URL from PULSAR_TEST_URL, and the live 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 URL then fails the test instead of skipping it. The repository's stand is docker-compose.test.yml, a Pulsar standalone on 127.0.0.1:6650 that just brokers-up starts.

Testing a Pulsar service runs the same test against a server.

Example

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

Macros
Manual

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

    tb.broker::<PulsarBroker>()
        .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("pulsar://localhost:6650", "payments-svc")).await?)
        .await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = stand() else {
        return Ok(());
    };
    // A subscription of this run's own: it starts at the end of the topic, past what
    // earlier runs published, and its cursor outlives the test on the server.
    let subscription = format!("payments-{}", process::id());
    accepts_a_payment(TestApp::start_live(app(&url, &subscription)).await?).await
}

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

    tb.broker::<PulsarBroker>()
        .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("pulsar://localhost:6650", "payments-svc")).await?)
        .await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn against_the_stand() -> Result<(), Box<dyn Error>> {
    let Some(url) = stand() else {
        return Ok(());
    };
    // A subscription of this run's own: it starts at the end of the topic, past what
    // earlier runs published, and its cursor outlives the test on the server.
    let subscription = format!("manual-payments-{}", process::id());
    accepts_a_payment(TestApp::start_live(app(&url, &subscription)).await?).await
}

Example​

Example