Skip to main content

Testing a ZeroMQ service

A test runs the service's own app with TestApp, in process or over real sockets on the loopback.

The service under test is a worker. It takes jobs on a queue it binds and announces each finished job as jobs.done on a fan-out it binds. A job whose input has not been uploaded yet is held: the handler asks for it again in a minute with retry_after. After three deliveries a held job is dropped. The app is built by service(jobs, events), so a test chooses the addresses and main uses the production ones. The tests need the testing feature of ruststream-zeromq in [dev-dependencies].

A job and its announcement​

src/main.rs
/// The body both modes run: a job arrives, is worked, and its announcement leaves.
async fn a_job_is_worked_and_announced(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    let job = Job {
        id: 7,
        uploaded: true,
    };
    tb.broker::<ZmqQueue>().message(&job).publish().await?;

    tb.broker::<ZmqQueue>()
        .subscriber("jobs")
        .assert_called_once()
        .with(&job)
        .settled(HandlerOutcome::ack());
    tb.broker::<ZmqFanout>()
        .published::<JobDone>("jobs.done")
        .assert_called_once()
        .with(&JobDone { id: 7 });
    tb.shutdown().await?;
    Ok(())
}

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

TestApp::start connects each broker of the app in process. No socket is opened, and the addresses are never bound. The test addresses a broker by its production type: tb.broker::<ZmqQueue>() is the queue this service binds, and tb.broker::<ZmqFanout>() the fan-out. A broker this service dials is tb.broker::<ZmqQueue<Connect>>(), and two brokers of one type are told apart by the labels of with_broker_labeled, through tb.broker_named(..).

The test's job arrives the way a peer in another process pushes it. publish returns once the handler and its announcement have settled. So the next lines assert on finished state: how the subscriber settled, and what was published on jobs.done. The body takes the started app as an argument, so the test over sockets below runs it too.

A delay, without waiting for it​

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_job_without_its_input_is_held() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let job = Job {
        id: 8,
        uploaded: false,
    };
    tb.broker::<ZmqQueue>().message(&job).publish().await?;
    tb.broker::<ZmqQueue>()
        .subscriber("jobs")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // The clock is paused: a minute of HOLD passes at once.
    tb.advance(HOLD).await?;
    tb.advance(HOLD).await?;

    let queue = tb.broker::<ZmqQueue>();
    assert_eq!(
        queue.subscriber("jobs").outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack]
    );
    queue
        .published::<Job>("jobs")
        .assert_called(3)
        .with_header(RETRY_COUNT_HEADER, "2");
    tb.broker::<ZmqFanout>()
        .published::<JobDone>("jobs.done")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

The test starts on a paused clock, and advance moves it forward. A minute of HOLD costs nothing. ZeroMQ stores nothing, so retry_after publishes a copy of the job to its queue when the delay ends. The copy carries the retry count in a header, and published shows it on jobs next to the test's own job. The third delivery is the last one the mount site allows, so nothing is published after it. Each held delivery is recorded as Outcome::Nack.

What the in-process mode models​

The in-process mode carries the frames a socket carries, and routes them by the pattern's rule:

  • A queue that binds takes whatever a peer pushes to it; subscriptions that dial take a peer's messages in turn.
  • A fan-out delivers to every subscription whose name is a prefix of the message's name.
  • A request reaches the responder, and its answer reaches the peer that asked.
  • A publish to an endpoint the service dials, or binds with no subscription, leaves for another process: the test reads it with published.

The socket's refusals hold in process too, in the socket's words. A publish a bound queue would hand back to its own subscription is refused, and so is a second subscription on a bound endpoint. ack and nack report AckError::Unsupported, as on a socket, so a handler that settles by a plain retry fails its test.

Delivery channels have no bound in process. Back-pressure, the read_ahead bound, late fan-out subscribers, the five-second wait for a peer and reconnection exist only on real sockets.

The same test over sockets​

src/main.rs
#[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", "tcp://127.0.0.1:0")).await?;
    a_job_is_worked_and_announced(tb).await
}

TestApp::start_live starts the same app over real ZMTP sockets. ZeroMQ needs no server, so the test binds loopback ports and runs in every cargo test. Port zero lets the system pick a free port, so tests that run in parallel never collide. The test's job leaves through the broker's own publisher and crosses a real socket. The assertions are the in-process ones, unchanged.

cargo test