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

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

Запрос и ответ входят в сам протокол. В запросе есть адрес для ответа в reply-to и идентификатор в correlation-id. Ответ уходит на этот адрес и повторяет этот идентификатор.

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

Отвечающий сервис

Макросы

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 reply address 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(AmqpAddress::queue("prices.quote"))]
async fn quote(
    ask: &Ask,
    ctx: &mut Context<'_>,
    Out(answers): Out<impl Publisher, Answers>,
) -> HandlerOutcome {
    let Some(reply_to) = ctx.headers().reply_to().map(str::to_owned) else {
        tracing::warn!(ask.sku, "a question without a reply address 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();
    };
    // The requester matches the answer to its question by the correlation id.
    let mut headers = HeaderMap::new();
    if let Some(correlation_id) = ctx.headers().correlation_id() {
        headers.insert("correlation-id", correlation_id.to_owned());
    }
    let price = Price {
        sku: ask.sku.clone(),
        cents,
    };
    if answers
        .message(&price)
        .to(reply_to)
        .with_headers(headers)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// What the orders service asks the pricing service.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Ask {
    sku: String,
}

/// The pricing service's answer, sent to the reply address of the request.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Price {
    sku: String,
    cents: u64,
}

impl OutgoingDestination for Price {
    type Form = CallerName;
}

impl MessageHeaders for Price {
    type Contract = NoHeaders;
}

impl MessageInfo for Price {
    const NAME: &'static str = "Price";
}

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

struct Answers;

impl OutSlot for Answers {
    const NAME: &'static str = "Answers";
    type Destination = Reads;
}

impl PublishedThrough<Answers> for Price {}

struct Quote;

impl<Sent> Handle<Ask, (), Outs<(Sent,)>> for Quote
where
    Sent: OutEntry<Answers, Wire: Publisher>,
{
    async fn handle(
        &self,
        ask: &Ask,
        outs: &Outs<(Sent,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let Some(reply_to) = ctx.headers().reply_to().map(str::to_owned) else {
            tracing::warn!(ask.sku, "a question without a reply address dropped");
            return Err(HandlerOutcome::drop());
        };
        let Some(cents) = price_of(&ask.sku) else {
            tracing::info!(ask.sku, "no price for the item, no answer");
            return Err(HandlerOutcome::drop());
        };
        // The requester matches the answer to its question by the correlation id.
        let mut headers = HeaderMap::new();
        if let Some(correlation_id) = ctx.headers().correlation_id() {
            headers.insert("correlation-id", correlation_id.to_owned());
        }
        let price = Price {
            sku: ask.sku.clone(),
            cents,
        };
        if outs
            .get(Answers)
            .message(&price)
            .to(reply_to)
            .with_headers(headers)
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

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

Отвечающий сервис читает очередь prices.quote. Каждая реплика сервиса цен - ещё один её потребитель, и каждый запрос достаётся одной из них.

Запрашивающий сервис

Макросы

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")]
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(AmqpAddress::queue("orders"), 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,
    })
}

Вручную

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, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    sku: String,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

impl MessageInfo for Order {
    const NAME: &'static str = "Order";
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Priced {
    order: u64,
    cents: u64,
}

impl OutgoingDestination for Priced {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders.priced";
}

impl MessageHeaders for Priced {
    type Contract = NoHeaders;
}

impl MessageInfo for Priced {
    const NAME: &'static str = "Priced";
}

struct Pricing;

impl OutSlot for Pricing {
    const NAME: &'static str = "Pricing";
    type Destination = Reads;
}

struct PriceOrder;

impl<Asks> Handle<Order, Priced, Outs<(Asks,)>> for PriceOrder
where
    Asks: OutEntry<Pricing, Wire: RequestReply>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Asks,)>,
        _ctx: &mut Context<'_>,
    ) -> 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 outs.get(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. Её издатель, AmqpPublisher, поддерживает запрос и ответ. request присоединяет динамическую связь получателя, поэтому брокер создаёт закрытый адрес для ответа на этот единственный запрос. Он публикует вопрос с заданными reply-to и correlation-id и возвращает первый совпавший ответ. ActiveMQ Artemis создаёт для динамической связи временную очередь. Вопрос публикуется как сообщение из байтов, без кодека, поэтому тело обработчика само кодирует его и декодирует ответ, здесь через serde_json. Ответ на заказ, Priced, уходит, как только цена известна.

Запрос, на который никто не ответил, возвращает ошибку AmqpError::RequestTimeout по истечении тайм-аута, здесь двух секунд. Связь для ответа отсоединяется в любом случае. Затем заказ ждёт 30 секунд в своей очереди и спрашивает снова.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(quote).out(Answers, Publish).build();
            b.include(price_order)
                .out_reply(Publish)
                .out(Pricing, Publish)
                .build();
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(subscriber(AmqpAddress::queue("prices.quote"), Quote).build())
                .out(Answers, Publish)
                .build();
            b.include(
                subscriber(AmqpAddress::queue("orders"), PriceOrder)
                    .reply()
                    .build(),
            )
            .out_reply(Publish)
            .out(Pricing, Publish)
            .build();
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

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

Тесты в файле примера запускают обе половины в одном приложении внутри процесса. Брокер внутри процесса направляет запрос собственному отвечающему сервису приложения, сопоставляет ответ по correlation-id и завершает по тайм-ауту запрос к молчащему сервису.