Skip to main content

Request and reply

This guide makes one service call another over RabbitMQ and wait for the answer.

The orders service asks the inventory service whether an item is in stock before it accepts an order. The call goes over RabbitMQ's direct reply-to: no reply queue to declare, and no table of pending calls to keep by hand.

The responder​

src/main.rs
/// The stock of one SKU. An RPC reply goes where the request asked, so the type names no
/// destination.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Stock {
    sku: String,
    available: bool,
}

#[subscriber("inventory.check", reply("inventory.check.unanswered"))]
async fn check_stock(request: &CheckStock) -> Stock {
    tracing::info!(request.sku, request.quantity, "stock checked");
    Stock {
        sku: request.sku.clone(),
        available: request.quantity <= 10,
    }
}

/// The inventory service's routes: each reply goes to the requester's own address.
fn inventory() -> Router<LapinBroker, impl RouterDef<LapinBroker>> {
    Router::new()
        .include(check_stock)
        .out_reply(Publish::default())
        .transform(DirectReplyTo)
        .build()
}

The inventory service answers with an ordinary handler that returns its reply. DirectReplyTo on the reply publisher sends each reply to the address the request named in its reply-to property, and copies the request's correlation-id onto it. Stock names no destination, because each reply goes where its request asked. The name in reply("inventory.check.unanswered") is where a request without a reply-to is answered, and what the service's AsyncAPI document reports.

The routes are a Router, so the inventory service mounts them with b.include_router(inventory()) in its own app. The example runs both services in one process and mounts both routers.

The requester​

src/main.rs
/// How long an order waits for the inventory service.
const RPC_TIMEOUT: Duration = Duration::from_secs(2);
/// How long an order waits before it asks again when the inventory service did not answer.
const RETRY: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
    sku: String,
    quantity: u32,
}

/// Asks the inventory service and waits for its answer.
async fn ask_stock(
    inventory: &impl RequestReply,
    order: &Order,
) -> Result<Stock, Box<dyn Error + Send + Sync>> {
    let request = JsonCodec.encode(&CheckStock {
        sku: order.sku.clone(),
        quantity: order.quantity,
    })?;
    let reply = inventory
        .request(
            OutgoingMessage::new("inventory.check", request.as_ref()),
            RPC_TIMEOUT,
        )
        .await?;
    Ok(JsonCodec.decode(reply.payload())?)
}

#[subscriber("orders")]
async fn place_order(order: &Order, Out(inventory): Out<impl RequestReply>) -> HandlerOutcome {
    match ask_stock(inventory, order).await {
        Ok(stock) if stock.available => {
            tracing::info!(order.id, "order accepted");
            HandlerOutcome::ack()
        }
        Ok(_) => {
            tracing::warn!(order.id, "order rejected, out of stock");
            HandlerOutcome::drop()
        }
        Err(err) => {
            tracing::warn!(order.id, %err, "inventory unavailable, asking again later");
            HandlerOutcome::retry_after(RETRY)
        }
    }
}

/// The orders service's routes: the requester is a policy, paired with the connection at startup.
fn orders() -> Router<LapinBroker, impl RouterDef<LapinBroker>> {
    Router::new()
        .include(place_order)
        .out(DefaultSlot, Request::default())
        .build()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    // The example runs both services in one process; each service mounts its own router.
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        b.include_router(orders());
        b.include_router(inventory());
    })
}

place_order bounds its slot Out<impl RequestReply>, and the mount site pairs it with the Request policy, LapinRequest. Each request goes to the default exchange with the routing key inventory.check, so it reaches the queue of that name. It carries reply-to set to RabbitMQ's amq.rabbitmq.reply-to and a fresh correlation-id. The reply with the same id resolves the call; a reply with another id is dropped. The request and the reply are bytes here, so ask_stock encodes and decodes them with JsonCodec.

RPC_TIMEOUT bounds the wait. When nobody answers in time, the call fails with AmqpError::RequestTimeout. place_order then asks for the order again in 30 seconds with retry_after, instead of guessing the stock. An item out of stock is a business answer, so the order is dropped.

The policy takes the settings of a publish: exchange, priority and expiration. A request is transient by default: nobody waits for it after the timeout, so it gains nothing from surviving a restart. persistent(true) makes it persistent.

Delivery guarantees​

Direct reply-to is at-most-once. The reply address lives in the requester's channel on one RabbitMQ node, and nothing is queued durably. A reply in flight when that channel closes is lost, and the caller sees a timeout. The request itself is an ordinary message in an ordinary queue: when the inventory service is down, it waits in inventory.check and may be answered after the caller has given up. So the inventory call is safe to repeat: answering it twice changes nothing.

The requester reads its replies on a channel of its own, opened by its first request.

Another stack on the other side​

Both sides speak plain AMQP 0.9.1. A responder written with another client consumes the request queue, publishes its reply to the default exchange with the routing key from reply-to, and copies the correlation-id. A RustStream responder with DirectReplyTo answers a requester written with another client the same way.

Testing​

An in-process RabbitMQ belongs to the app that connected it, so the requester and the responder must run in one app to talk. The tests start the example's app, which mounts both routers, and check that an order in stock is accepted and one out of stock is dropped. The third test mounts the orders router alone, with a paused clock: nobody answers, the request times out, and the order asks again later. The in-process RabbitMQ gives every request its own reply address and correlates the reply, as the server does. Testing a RabbitMQ service covers the harness.