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

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

Запрос и ответ

Это руководство отправляет запрос другому сервису через NATS и ждёт ответа, отвечает на запросы в их инбокс, обрабатывает запрос, на который никто не отвечает, и масштабирует отвечающую сторону.

NATS сам сопоставляет ответ с запросом. Запрос содержит инбокс, уникальный для него субъект, и ответ публикуется в этот инбокс.

Пример - два сервиса в одном приложении. Сервис цен отвечает на вопросы о ценах, а сервис заказов спрашивает у него цену каждого заказа.

Отвечающая сторона​

src/main.rs
/// What the orders service asks the pricing service.
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Ask {
    sku: String,
}

/// The pricing service's answer, sent to the inbox of the request.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Price {
    sku: String,
    cents: u64,
}

/// The catalogue the pricing service answers from.
fn price_of(sku: &str) -> Option<u64> {
    match sku {
        "book" => Some(1250),
        "pen" => Some(150),
        _ => None,
    }
}

#[derive(OutSlot)]
#[publishes(Price)]
struct Answers;

#[subscriber(CoreSubject::new("prices.quote").queue_group("pricing"))]
async fn quote(
    ask: &Ask,
    ctx: &mut Context<'_>,
    Out(answers): Out<impl Publisher, Answers>,
) -> HandlerOutcome {
    let Some(inbox) = ctx.headers().reply_to().map(str::to_owned) else {
        tracing::warn!(ask.sku, "a question without an inbox dropped");
        return HandlerOutcome::drop();
    };
    let Some(cents) = price_of(&ask.sku) else {
        tracing::info!(ask.sku, "no price for the item, no answer");
        return HandlerOutcome::drop();
    };
    let price = Price {
        sku: ask.sku.clone(),
        cents,
    };
    if answers.message(&price).to(inbox).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Запрос приходит с инбоксом в заголовке reply-to, и ctx.headers().reply_to() читает его. quote публикует цену через свой слот Answers с to(inbox), потому что адрес известен только для каждого запроса отдельно. Поэтому Price не объявляет собственного субъекта. На товар без цены ответа нет, и ожидание запрашивающей стороны заканчивается по тайм-ауту.

Отвечающая сторона читает prices.quote в очереди pricing. Каждая реплика сервиса цен входит в очередь, и каждый запрос достаётся одной из них. На тот же запрос ответит любая отвечающая сторона NATS, например nats reply prices.quote '{"sku":"book","cents":1250}' из командной строки.

Запрашивающая сторона​

src/main.rs
/// How long the orders service waits for a price.
const PRICING_TIMEOUT: Duration = Duration::from_secs(2);

/// How long an order waits before it asks for its price again.
const RETRY_DELAY: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders.placed")]
struct Order {
    id: u64,
    sku: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders.priced")]
struct Priced {
    order: u64,
    cents: u64,
}

#[derive(OutSlot)]
struct Pricing;

#[subscriber(
    JetStreamSubject::new("orders.placed", "ORDERS").durable("pricer"),
    reply
)]
async fn price_order(
    order: &Order,
    Out(pricing): Out<impl RequestReply, Pricing>,
) -> Result<Priced, HandlerOutcome> {
    let Ok(ask) = serde_json::to_vec(&Ask {
        sku: order.sku.clone(),
    }) else {
        return Err(HandlerOutcome::drop());
    };
    let request = OutgoingMessage::new("prices.quote", ask.as_slice());
    let reply = match pricing.request(request, PRICING_TIMEOUT).await {
        Ok(reply) => reply,
        Err(err) => {
            tracing::warn!(order.id, %err, "no price, the order asks again later");
            return Err(HandlerOutcome::retry_after(RETRY_DELAY));
        }
    };
    let Ok(price) = serde_json::from_slice::<Price>(reply.payload()) else {
        return Err(HandlerOutcome::drop());
    };
    tracing::info!(order.id, price.cents, "order priced");
    Ok(Priced {
        order: order.id,
        cents: price.cents,
    })
}

price_order читает новые заказы из стрима и спрашивает цену каждого в prices.quote. Слот Pricing ограничен типом Out<impl RequestReply, _>: тело обработчика заявляет, что ему нужен запрос и ответ, и не называет брокер. Приложение связывает слот с обычной политикой Publish, издатель которой в NATS поддерживает запрос и ответ. request публикует вопрос с новым инбоксом и возвращает первое сообщение, пришедшее в этот инбокс. Вопрос - сырое сообщение, поэтому тело обработчика само кодирует его и декодирует ответ, здесь через serde_json. Ответ на заказ, Priced, уходит, как только цена известна.

Запрос может вернуть ошибку двумя способами. Если субъект не читает ни одна подписка, сервер сразу отвечает «no responders». Если отвечающая сторона есть, но молчит, запрос возвращает ошибку NatsError::RequestTimeout по истечении тайм-аута, здесь двух секунд. В обоих случаях заказ ждёт в стриме 30 секунд и спрашивает снова.

Приложение​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(quote).out(Answers, Publish).build();
            b.include(price_order)
                .out_reply(Publish)
                .out(Pricing, Publish)
                .build();
        },
    )
}

Оба слота принимают обычную политику Publish: один публикует ответы, другой задаёт вопросы. Хук запуска спрашивает так же: через b.after_startup(Publish, ..) и request на издателе, который он получает.

Тесты в файле примера запускают обе стороны в одном приложении внутри процесса. NATS внутри процесса направляет запрос отвечающей стороне самого приложения, отвечает «no responders», если субъект не читает ни одна подписка, и завершает по тайм-ауту запрос к молчащей отвечающей стороне.