Skip to main content

Requests and replies

This guide asks another service over AMQP 1.0 and waits for its answer, answers requests on their reply address, and handles a request nobody answers.

Request-reply is part of the protocol. A request carries a reply address in reply-to and an id in correlation-id. The answer goes to that address and repeats that id.

The example is two services in one app. The pricing service answers questions about prices, and the orders service asks it for the price of each order.

The responder​

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()
}

A request arrives at a responder with its reply address in the reply-to header, and ctx.headers().reply_to() reads it. quote publishes the price through its Answers slot with to(reply_to), because the address is new for every request. Price therefore declares no address of its own. The answer carries the request's correlation-id, and the requester takes only an answer whose id matches. An item without a price gets no answer, and the requester's timeout ends the wait.

The responder reads the queue prices.quote. Each replica of the pricing service is one more consumer of it, and each request goes to one of them.

The requester​

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,
    })
}

price_order reads orders and asks for each price on prices.quote. The Pricing slot is bounded Out<impl RequestReply, _>: the body says it needs request-reply and names no broker. The app pairs the slot with the plain Publish policy, whose publisher, AmqpPublisher, carries request-reply. request attaches a dynamic receiver link, so the broker creates a private reply address for this one request. It publishes the question with reply-to and correlation-id set and resolves with the first matching answer. ActiveMQ Artemis backs a dynamic link with a temporary queue. The question is a raw message, so the body encodes it and decodes the answer itself, here with serde_json. The order's reply, Priced, goes out once the price is known.

A request nobody answers fails after its timeout, two seconds here, with AmqpError::RequestTimeout. The reply link is detached either way. The order then waits 30 seconds on its queue and asks again.

The app​

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();
        },
    )
}

Both slots take the plain Publish policy: one publishes answers, the other asks questions. A request that starts a conversation has no delivery to answer. It runs from a startup hook, b.after_startup(Publish, ..), with request on the publisher the hook is handed.

The tests in the example file run both halves in one app in process. The in-process broker routes a request to the app's own responder, matches the answer by its correlation id, and times a silent responder out.