Live

TestApp::start_live runs the same test over loopback sockets in every cargo test, for back-pressure and the real frames.

It connects each broker through its ordinary connect, so every subscription binds or dials a real socket. ZeroMQ needs no server, so a live test needs no stand and no switch, and runs wherever cargo test runs. The body is the one an in-process test runs; only the start call differs.

A live test binds tcp://127.0.0.1:0: the system picks a free port, so tests that run in parallel never collide. An ipc:// path unique to the test serves the same purpose. A test's input leaves through the broker's own publisher, so on a bound queue it carries the subscription's own name. The fan-out's publisher reads its own subscription's filter before its first send, so a live fan-out test publishes once, as an in-process one does. A publish, settle and advance wait on the real clock until every subscription has handled what reached it, ten seconds at most, and start_live_within sets another bound.

The live mode is for what the in-process transport leaves out: the read-ahead bound and the back-pressure it puts on a sender, a port another process holds, and the frames a peer in another language sends.

Testing a ZeroMQ service runs the same test in process and over sockets.

Example

The two tests share one body: one in process, one over loopback sockets.

Macros
Manual

examples/zmq_testing_live.rs
/// The body both modes run; only the start call differs.
async fn works_a_job_and_drops_an_empty_one(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    for id in [7, 0] {
        tb.broker::<ZmqQueue>()
            .message(&Job { id })
            .publish()
            .await?;
    }

    let queue = tb.broker::<ZmqQueue>();
    let jobs = queue.subscriber("jobs");
    assert_eq!(jobs.received::<Job>(), [Job { id: 7 }, Job { id: 0 }]);
    assert_eq!(jobs.outcomes(), [Outcome::Ack, Outcome::Drop]);
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    works_a_job_and_drops_an_empty_one(TestApp::start(app()).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn over_loopback_sockets() -> Result<(), Box<dyn Error>> {
    // Port zero: the system picks a free port, so parallel tests never collide.
    let tb = TestApp::start_live(service("tcp://127.0.0.1:0")).await?;
    works_a_job_and_drops_an_empty_one(tb).await
}

examples/manual/zmq_testing_live.rs
/// The body both modes run; only the start call differs.
async fn works_a_job_and_drops_an_empty_one(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    for id in [7, 0] {
        tb.broker::<ZmqQueue>()
            .message(&Job { id })
            .publish()
            .await?;
    }

    let queue = tb.broker::<ZmqQueue>();
    let jobs = queue.subscriber("jobs");
    assert_eq!(jobs.received::<Job>(), [Job { id: 7 }, Job { id: 0 }]);
    assert_eq!(jobs.outcomes(), [Outcome::Ack, Outcome::Drop]);
    tb.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn in_process() -> Result<(), Box<dyn Error>> {
    works_a_job_and_drops_an_empty_one(TestApp::start(app()).await?).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn over_loopback_sockets() -> Result<(), Box<dyn Error>> {
    // Port zero: the system picks a free port, so parallel tests never collide.
    let tb = TestApp::start_live(service("tcp://127.0.0.1:0")).await?;
    works_a_job_and_drops_an_empty_one(tb).await
}

Example​

Example