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

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

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

Тест запускает собственное приложение сервиса через TestApp на Pulsar внутри процесса или на работающем сервере.

Тестируемый сервис списывает платежи по картам. Платёж приходит в топик card-payments, и его ключ партиционирования - счёт для списания. Обработчик отвечает квитанцией в топик card-receipts. Платёж без ключа отбрасывается. Платёж сверх мгновенного лимита банка откладывается: обработчик просит доставить его снова через минуту с retry_after. После трёх доставок платёж уходит в card-payments-review, топик dead-letter. Приложение строит service(broker), поэтому тест выбирает брокер сам, а main берёт рабочий. Тестам нужна фича testing крейта ruststream-pulsar в [dev-dependencies].

Платёж с ключом​

src/main.rs
#[tokio::test]
async fn a_keyed_payment_gets_a_receipt() -> Result<(), Box<dyn Error>> {
    let broker = PulsarBroker::new("pulsar://localhost:6650").default_subscription("payments");
    let tb = TestApp::start(service(broker)).await?;
    a_receipt_names_the_account_of_the_key(tb).await
}

async fn a_receipt_names_the_account_of_the_key(tb: TestApp<()>) -> Result<(), Box<dyn Error>> {
    let mut key = HeaderMap::new();
    key.insert(PARTITION_KEY_HEADER, "acct-7");
    tb.broker::<PulsarBroker>()
        .message(&Payment {
            id: 1,
            amount_cents: 2_500,
        })
        .with_headers(key)
        .publish()
        .await?;

    tb.broker::<PulsarBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::ack());
    tb.broker::<PulsarBroker>()
        .published::<Receipt>("card-receipts")
        .assert_called_once()
        .with(&Receipt {
            payment: 1,
            account: "acct-7".to_owned(),
        });
    tb.shutdown().await?;
    Ok(())
}

TestApp::start подключает PulsarBroker к Pulsar внутри процесса. Адрес подключения проверяется так, как его разобрал бы клиент, но соединение по нему не устанавливается. Тест обращается к брокеру по его рабочему типу: tb.broker::<PulsarBroker>(). Ключ партиционирования передаётся заголовком PARTITION_KEY_HEADER, поэтому тест задаёт его через with_headers. Брокер превращает этот заголовок в ключ сообщения, а обработчик читает его обратно из того же заголовка. publish возвращает управление, когда обработчик и его ответ завершены. Поэтому следующие строки проверяют уже готовое состояние: как подписчик завершил доставку и что дошло до card-receipts. Тело теста принимает запущенное приложение аргументом, поэтому его же выполняет и тест на сервере ниже.

Платёж без ключа​

src/main.rs
#[tokio::test]
async fn a_payment_without_an_account_is_dropped() -> Result<(), Box<dyn Error>> {
    let broker = PulsarBroker::new("pulsar://localhost:6650").default_subscription("payments");
    let tb = TestApp::start(service(broker)).await?;

    tb.broker::<PulsarBroker>()
        .message(&Payment {
            id: 2,
            amount_cents: 2_500,
        })
        .publish()
        .await?;

    tb.broker::<PulsarBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::drop());
    tb.broker::<PulsarBroker>()
        .published::<Receipt>("card-receipts")
        .assert_not_called();
    tb.shutdown().await?;
    Ok(())
}

settled проверяет исход, который вернул обработчик. assert_not_called проверяет, что квитанция не опубликована.

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

src/main.rs
#[tokio::test(start_paused = true)]
async fn a_payment_the_bank_keeps_holding_goes_to_review() -> Result<(), Box<dyn Error>> {
    let broker = PulsarBroker::new("pulsar://localhost:6650").default_subscription("payments");
    let tb = TestApp::start(service(broker)).await?;
    let held = Payment {
        id: 3,
        amount_cents: 250_000,
    };
    let mut key = HeaderMap::new();
    key.insert(PARTITION_KEY_HEADER, "acct-7");

    tb.broker::<PulsarBroker>()
        .message(&held)
        .with_headers(key)
        .publish()
        .await?;
    tb.broker::<PulsarBroker>()
        .subscriber("card-payments")
        .assert_called_once()
        .settled(HandlerOutcome::retry_after(HOLD));

    // Each delivery waits `HOLD`, then the broker hands the same message over again.
    tb.advance(HOLD).await?;
    tb.broker::<PulsarBroker>()
        .subscriber("card-payments")
        .assert_called(2);

    // The third delivery is the last one the mount site allows: after its wait, the client
    // moves the payment to the review topic.
    tb.advance(HOLD).await?;
    tb.advance(HOLD).await?;
    assert_eq!(
        tb.broker::<PulsarBroker>()
            .subscriber("card-payments")
            .outcomes(),
        [Outcome::Nack, Outcome::Nack, Outcome::Nack],
    );
    tb.broker::<PulsarBroker>()
        .published::<Payment>("card-payments-review")
        .assert_called_once()
        .with(&held);
    tb.shutdown().await?;
    Ok(())
}

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

Что моделирует Pulsar внутри процесса​

Pulsar внутри процесса читает ту же конфигурацию, которую получил бы сервер: настройки брокера, каждый дескриптор и объявление повторов каждой регистрации. Он отвергает то, что отвергает сервер, с той же ошибкой, которую возвращает живой брокер: адрес подключения, который клиент не может разобрать, адрес публикации, который не является именем топика, второго консьюмера подписки Exclusive и дескриптор, использованный после остановки. Он моделирует Pulsar, а не очередь:

  • Топик определяется полным именем, поэтому orders и persistent://public/default/orders - один топик.
  • Каждая подписка получает каждое сообщение своих топиков один раз, а её тип выбирает консьюмера.
  • Отрицательно подтверждённое сообщение возвращается в свою подписку, а политика dead-letter считает его доставки так же, как клиент.
  • Каждый топик хранит свой журнал, поэтому перемотка перечитывает его. Нехранимый топик ничего не перечитывает.
  • Подписка по шаблону читает новый топик со следующего обновления списка у клиента, через 30 секунд по остановленным часам.
  • Заголовки передаются текстовыми свойствами, а публикация с полезной нагрузкой больше max_message_size возвращает ошибку.

Сервер заставляет второго консьюмера Exclusive ждать, а Pulsar внутри процесса сразу ему отказывает, поэтому такой тест не проходит, а не зависает. Для повторной доставки по тайм-ауту подтверждения, партиционированных топиков, диапазонов ключей KeyShared и сообщений, опубликованных, пока у подписки нет консьюмера, нужен сервер.

Тот же тест на сервере​

src/main.rs
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_keyed_payment_gets_a_receipt_on_a_server() -> Result<(), Box<dyn Error>> {
    let Some(url) = stand() else {
        return Ok(());
    };
    let subscription = format!("payments-{}", process::id());
    let broker = PulsarBroker::new(url).default_subscription(subscription);
    let tb = TestApp::start_live(service(broker)).await?;
    a_receipt_names_the_account_of_the_key(tb).await
}

TestApp::start_live подключает то же приложение к работающему Pulsar. Тест читает адрес из PULSAR_TEST_URL через stand, собственную вспомогательную функцию примера, и пропускается, если переменная не задана. RUSTSTREAM_REQUIRE_LIVE=1 превращает этот пропуск в ошибку, поэтому задание CI не может пройти без сервера. Тест подключается к собственной подписке, названной по процессу. Новая подписка начинается с конца топика, поэтому ничего, что оставил прошлый прогон, не читается. Проверки те же, что и внутри процесса, без изменений.

cargo test
PULSAR_TEST_URL=pulsar://localhost:6650 cargo test -- --test-threads=1

Сервер - тот же, что в руководстве Начало работы.