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

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

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

Это руководство отвечает на запросы на DEALER/ROUTER и спрашивает другой сервис изнутри обработчика.

В примере два сервиса. Сервис оценки отвечает, одобрена ли сумма заказа для клиента. Сервис оформления заказов спрашивает его о каждом принятом заказе.

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

src/main.rs
use ruststream_zeromq::rpc::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize)]
struct ScoreRequest {
    customer: u64,
    amount_cents: u64,
}

// No destination of its own: the ROUTER addresses each answer per request.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
struct Score {
    customer: u64,
    approved: bool,
}

// The answer goes to the peer that asked, so the clause's name is only the fallback.
#[subscriber("score", reply("reply"))]
async fn score(request: &ScoreRequest) -> Result<Score, HandlerOutcome> {
    if request.amount_cents == 0 {
        tracing::warn!(request.customer, "a request without an amount, dropped");
        return Err(HandlerOutcome::drop());
    }
    tracing::info!(request.customer, request.amount_cents, "scoring");
    Ok(Score {
        customer: request.customer,
        approved: request.amount_cents <= 50_000,
    })
}

#[ruststream::app]
fn app() -> impl App {
    // The responder is the stable side: it binds, and every requester dials it.
    RustStream::new(AppInfo::new("scoring", "0.1.0")).with_broker(
        ZmqRpc::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")),
        |b| {
            // A responder addresses no retry copies, so the mount names where they go.
            b.include(score)
                .out_reply(Publish)
                .transform(ReplyToRequester)
                .out_retry(Publish)
                .to("score.retry");
        },
    )
}

ZmqRpc - паттерн DEALER/ROUTER. Отвечающая сторона устойчива, поэтому она привязывается к адресу, а каждая запрашивающая сторона подключается к ней. score - обычный обработчик в reply-форме: он возвращает ответ или отбрасывает запрос без суммы.

Сокет ROUTER знает, какой узел отправил каждый запрос. Он записывает этот узел в заголовок запроса reply-to, и ответ должен уйти туда. Поэтому Score не объявляет своего адреса назначения, а reply("reply") называет только запасной адрес. Преобразование ниже заменяет его для каждого запроса.

Отвечающая сторона называет, куда уходят её копии для повтора: .out_retry(Publish).to("score.retry"). У копии запроса нет своего адреса, поэтому регистрацию без этого шага старт отвергает. Пакеты отвечающая сторона тоже не принимает: .batch(..) на ней не компилируется, потому что каждый ответ уходит своему узлу.

Преобразование​

src/main.rs
/// Answers at the `reply-to` address the ROUTER stamped on the request, under the request's
/// `correlation-id`.
struct ReplyToRequester;

impl<C, Options> PublishTransform<ForReply<C>, Options> for ReplyToRequester {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(reply_to) = cx.headers().get_shared("reply-to")
            && let Ok(reply_to) = Str::try_from(reply_to)
        {
            out.set_name(reply_to);
        }
        if let Some(correlation) = cx.headers().get_shared("correlation-id") {
            out.headers_mut()
                .insert(Str::from_static("correlation-id"), correlation);
        }
    }
}

ReplyToRequester - преобразование публикации на ответе. Оно читает заголовок reply-to запроса и отправляет ответ туда. Ещё оно копирует correlation-id запроса: по нему запрашивающая сторона сопоставляет ответ с запросом. Оно объявляет Destination = Names, потому что само выбирает адрес назначения. Так преобразование может поступать, только если тип ответа не объявляет адреса, поэтому тип ответа с именем и это преобразование вместе не компилируются.

Ответ узлу, который уже отключился, отвергается с ошибкой, в которой назван его адрес. Так отвечающая сторона узнаёт, что ответ никуда не дошёл. На широковещательной конечной точке сообщение без совпавшего фильтра отбрасывается молча.

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

src/main.rs
use std::time::Duration;

use ruststream::OutgoingMessage;
use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

/// How long one request waits for its answer.
const ASK_TIMEOUT: Duration = Duration::from_secs(2);
/// When the order is asked about again after the scoring service did not answer.
const ASK_AGAIN: Duration = Duration::from_secs(10);

#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    customer: u64,
    amount_cents: u64,
}

#[derive(Serialize)]
struct ScoreRequest {
    customer: u64,
    amount_cents: u64,
}

#[derive(Deserialize)]
struct Score {
    approved: bool,
}

#[subscriber("orders")]
async fn check(order: &Order, Out(scoring): Out<impl RequestReply>) -> HandlerOutcome {
    let Ok(request) = serde_json::to_vec(&ScoreRequest {
        customer: order.customer,
        amount_cents: order.amount_cents,
    }) else {
        return HandlerOutcome::drop();
    };
    let request = OutgoingMessage::new("score", request.as_slice());
    let answer = match scoring.request(request, ASK_TIMEOUT).await {
        Ok(answer) => answer,
        Err(err) => {
            tracing::warn!(order.id, %err, "no score, asking again later");
            return HandlerOutcome::retry_after(ASK_AGAIN);
        }
    };
    let Ok(score) = serde_json::from_slice::<Score>(answer.payload()) else {
        return HandlerOutcome::drop();
    };
    tracing::info!(order.id, score.approved, "order checked");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // The scoring service binds; every requester dials it.
    let scoring = ZmqRpc::new(ZmqEndpoint::connect("tcp://scoring:5557")).bindable();
    let ask = scoring.bind(ZmqRpcPublish);
    RustStream::new(AppInfo::new("checkout", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(check)
                    .max_attempts(nonzero!(3u32))
                    .out(DefaultSlot, ask)
                    .build();
            },
        )
        .with_broker(scoring, |_b| {})
}

check принимает параметр Out<impl RequestReply>, а точка монтирования связывает его с политикой ZmqRpc через .out(DefaultSlot, ask). Брокер scoring подключается к сервису оценки и подписки не имеет: он только спрашивает. request отправляет запрос через собственный сокет DEALER и возвращает ответ с тем же correlation-id. check сам кодирует запрос и декодирует ответ в JSON, который читает и пишет сервис оценки.

Таймауты​

Если ответ не пришёл за ASK_TIMEOUT, request возвращает ошибку ZmqError::RequestTimeout. Тогда check возвращает retry_after(ASK_AGAIN), и заказ возвращается через десять секунд. max_attempts ограничивает число попыток тремя. Это паттерн Lazy Pirate из руководства ZeroMQ, только попытки считает фреймворк. Запрос с истёкшим таймаутом мог всё же дойти до отвечающей стороны, поэтому повторная отправка запроса должна быть безопасной.

Ни request, ни публикацию ответа нельзя безопасно отменить. Обработчик прекращает ожидание по таймауту запроса, а не через ветку select! (отмена).

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

Тесты сервиса оценки отправляют запросы так, как это сделал бы DEALER запрашивающей стороны, и проверяют, как score завершил их доставку. Тест оформления заказа работает во внутрипроцессном режиме без сервиса оценки, поэтому его запрос истекает по таймауту. Он проверяет, что check попросил повтора и что копия вернулась через ASK_AGAIN со счётчиком повторов, равным единице. Часы остановлены, поэтому таймаут и задержка проходят мгновенно. Обвязку разбирает руководство Тестирование сервиса на ZeroMQ.