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

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

Публикация и ответы

Это руководство отвечает на сообщения Pulsar, публикует сообщения с ключом и заголовками через слот Out и публикует после запуска сервиса.

Пример - сервис оформления заказов. Он подтверждает заказы квитанциями, отвечает на запросы цен, отправляет на склады листы комплектации и объявляет о себе при запуске.

Ответ в объявленный топик​

src/main.rs
/// The receipt of an order; the type names the topic it goes to.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

/// Confirms each order with a receipt.
#[subscriber(PulsarSubscription::new("orders", "receipts"), reply)]
async fn confirm(order: &Order) -> Receipt {
    tracing::info!(order.id, "order confirmed");
    Receipt { order: order.id }
}

confirm возвращает Receipt, и фреймворк его публикует. #[outgoing(name = "receipts")] закрепляет топик за типом, поэтому подписчику достаточно аргумента reply макроса без имени. В ручной форме - .reply(). Политика публикации брокера, Publish из прелюдии, используется по умолчанию, поэтому простой include отвечает через неё. Публикация возвращает управление, когда брокер сохранил сообщение.

Ответ в топик, который называет точка монтирования​

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct QuoteRequest {
    request: u64,
    items: u32,
}

/// A price quote; the mount site names where it goes.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Quote {
    request: u64,
    total_cents: u64,
}

/// Answers a quote request on `quotes`, the topic the requesting services read.
#[subscriber(PulsarSubscription::new("quote-requests", "quotes"), reply("quotes"))]
async fn quote(request: &QuoteRequest) -> Quote {
    Quote {
        request: request.request,
        total_cents: u64::from(request.items) * 1_250,
    }
}

Quote не называет топик, поэтому его называет reply("quotes") в точке монтирования. В ручной форме - .reply().to("quotes"). Тогда тот же тип может отвечать в другой топик в другом сервисе.

Запрос и ответ на него проходят через два обычных топика. Сервис, который запрашивает цены, публикует в quote-requests и читает quotes через собственную подписку. Он сопоставляет каждый ответ с запросом по общему для двух сообщений номеру request.

Публикация с ключом и заголовками​

src/main.rs
/// A pick list for the warehouse that ships the order.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "warehouse-picks")]
struct Pick {
    order: u64,
}

#[derive(OutSlot)]
#[publishes(Pick)]
struct Picks;

/// Sends each order to its warehouse, keyed by the warehouse so its picks stay in order.
#[subscriber(PulsarSubscription::new("orders", "dispatch"))]
async fn dispatch(
    order: &Order,
    Out(picks): Out<impl Publisher<Options = PulsarPublishOptions>, Picks>,
) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("source", "checkout");
    let sent = picks
        .message(&Pick { order: order.id })
        .with_headers(headers)
        .partition_key(order.warehouse.clone())
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(order.id, %error, "pick list not sent");
            HandlerOutcome::retry()
        }
    }
}

Слот Out публикует из тела обработчика, а Picks объявляет тип, который он передаёт. partition_key задаёт ключом каждого листа комплектации его склад. На партиционированном топике ключ выбирает партицию, а подписка KeyShared сохраняет порядок каждого ключа. Этот шаг берётся из прелюдии брокера, поэтому слот ограничен Publisher<Options = PulsarPublishOptions>, чтобы шаг был доступен.

Каждый заголовок становится одноимённым свойством сообщения. Консьюмер на другом языке читает source как обычное свойство Pulsar, без конверта вокруг полезной нагрузки. Pulsar передаёт свойства текстом, поэтому публикация со значением заголовка не в UTF-8 возвращает ошибку.

Публикация, которая не удалась, возвращает ошибку, и dispatch повторяет заказ. Полезная нагрузка больше max_message_size брокера, по умолчанию пять мебибайт, возвращает ошибку ещё до отправки.

Приложение​

src/main.rs
/// Tells the rest of the system that checkout is up.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "service-events")]
struct Started {
    service: String,
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            b.include(confirm);
            b.include(quote);
            b.include(dispatch).out(Picks, Publish).build();
            b.after_startup(Publish, async move |publisher| {
                let started = Started {
                    service: "checkout".to_owned(),
                };
                publisher.message(&started).publish().await
            });
        },
    )
}

.out(Picks, Publish) связывает слот с политикой публикации. after_startup выполняется один раз, когда подписки открыты, с издателем названной в нём политики, и checkout объявляет о себе в service-events. Хук, который возвращает ошибку, останавливает сервис при запуске.

Издатель открывает по одному продюсеру на топик при первой публикации в него, а shutdown закрывает их. Код вне приложения берёт издателя у брокера: PulsarBroker::publisher до запуска сервиса и ConnectedPulsarBroker::publisher после.

Тестирование публикаций​

Тесты в файле примера запускают приложение внутри процесса. published проверяет, что дошло до топика, вместе с заголовками, а tb.out::<Picks>() проверяет ключ, с которым публиковал слот. Pulsar внутри процесса упаковывает заголовки и ключ так же, как клиент, поэтому тест с заголовком, который отверг бы сервер, тоже не проходит.