Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Тестирование сервиса на ZeroMQ

Тест запускает собственное приложение сервиса через TestApp внутри процесса или на настоящих сокетах на петлевом интерфейсе.

Тестируемый сервис - воркер. Он берёт задания из очереди, к адресу которой привязывается сам, и оповещает о каждом выполненном задании как jobs.done через широковещательную конечную точку, которую тоже привязывает сам. Задание, входные данные которого ещё не загружены, откладывается: обработчик просит доставить его снова через минуту с retry_after. После трёх доставок отложенное задание отбрасывается. Приложение строит service(jobs, events), поэтому тест выбирает адреса сам, а main берёт рабочие. Тестам нужна фича testing крейта ruststream-zeromq в [dev-dependencies].

Задание и оповещение о нём​

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 подключает каждый брокер приложения внутри процесса. Сокеты не открываются, и к адресам никто не привязывается. Тест обращается к брокеру по его рабочему типу: tb.broker::<ZmqQueue>() - очередь, которую привязывает этот сервис, а tb.broker::<ZmqFanout>() - широковещательная конечная точка. Брокер, который подключается, - это tb.broker::<ZmqQueue<Connect>>(). Два брокера одного типа различаются по меткам из with_broker_labeled через tb.broker_named(..).

Задание теста приходит так же, как его отправил бы узел из другого процесса. publish возвращает управление, когда обработчик и его оповещение завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписчик завершил доставку и что опубликовано в jobs.done. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на сокетах ниже.

Задержка без ожидания​

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(())
}

Тест начинается с остановленными часами, и advance переводит их вперёд. Минута HOLD проходит мгновенно. ZeroMQ ничего не хранит, поэтому по окончании задержки retry_after публикует копию задания в его очередь. Копия хранит число повторов в заголовке, и published показывает её в jobs рядом с заданием самого теста. Третья доставка - последняя, которую разрешает точка монтирования, поэтому после неё ничего не публикуется. Каждая отложенная доставка записывается как Outcome::Nack.

Что моделирует внутрипроцессный режим​

Внутрипроцессный режим передаёт те же кадры, что и сокет, и направляет их по правилу паттерна:

  • Очередь, которая привязывается к адресу, принимает всё, что ей отправляет узел. Подписки, которые подключаются, получают сообщения узла по очереди.
  • Широковещательная конечная точка доставляет сообщение каждой подписке, имя которой - префикс имени сообщения.
  • Запрос доходит до отвечающей стороны, а ответ - до узла, который спросил.
  • Публикация на конечную точку, к которой сервис подключается или которую привязывает без подписки, уходит в другой процесс: тест читает её через published.

Отказы сокета действуют и внутри процесса, с теми же формулировками. Отвергается публикация, которую привязанная очередь вернула бы своей же подписке, и вторая подписка на привязанной конечной точке. ack и nack отвечают AckError::Unsupported, как на сокете, поэтому обработчик, который завершает доставку обычным повтором, не проходит свой тест.

Каналы доставки внутри процесса не ограничены. Обратное давление, предел read_ahead, опоздавшие подписчики широковещательной конечной точки, пятисекундное ожидание узла и переподключение есть только на настоящих сокетах.

Тот же тест на сокетах​

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 запускает то же приложение на настоящих сокетах ZMTP. ZeroMQ не нужен сервер, поэтому тест привязывается к портам петлевого интерфейса и выполняется при каждом cargo test. Порт 0 поручает выбор свободного порта системе, поэтому параллельные тесты не конфликтуют. Задание теста уходит через собственного издателя брокера и проходит через настоящий сокет. Проверки те же, что и внутри процесса, без изменений.

cargo test