Request/reply

RPC over RabbitMQ direct reply-to is how one service asks another a question through the broker it already has, instead of growing an HTTP sidechannel: an order service checks stock in the inventory service, a gateway fetches a price, a saga step confirms a reservation. The two halves are two ordinary services; the runnable pair is lapin_rpc_server and lapin_rpc_client.

The requester

broker.requester() implements the RequestReply capability: every request goes out with reply-to set to the direct reply-to pseudo-queue and a generated correlation-id, and the matching reply resolves the call. Wrap the raw capability in a small typed client and put it in the application state, like any other shared dependency:

/// A typed RPC client over the raw requester: encode the request, await the correlated reply,
/// decode it. One reusable value, shared through the application state.
#[derive(Clone)]
struct Inventory {
    requester: LapinRequester,
}

impl Inventory {
    async fn check(
        &self,
        sku: &str,
        quantity: u32,
    ) -> Result<Stock, Box<dyn std::error::Error + Send + Sync>> {
        let request = CheckStock {
            sku: sku.to_owned(),
            quantity,
        };
        let payload = JsonCodec.encode(&request)?;
        let reply = self
            .requester
            .request(
                OutgoingMessage::new("inventory.check", payload.as_ref()),
                Duration::from_secs(2),
            )
            .await?;
        Ok(JsonCodec.decode(reply.payload())?)
    }
}

Handlers then request it by type and call the other service in the middle of their own message flow. The RPC timeout is the failure boundary, and it maps straight onto settlement: a business answer settles the order, an unreachable service asks for redelivery:

// The business handler calls the other service synchronously. A business answer settles the
// order either way; only an unreachable inventory service (the RPC timed out or failed) asks
// the broker to redeliver and try again later.
#[subscriber("orders")]
async fn place_order(order: &Order, State(inventory): State<Inventory>) -> HandlerResult {
    match inventory.check(&order.sku, order.quantity).await {
        Ok(stock) if stock.available => {
            println!("order accepted: {} x{}", order.sku, order.quantity);
            HandlerResult::Ack
        }
        Ok(_) => {
            println!(
                "order rejected, out of stock: {} x{}",
                order.sku, order.quantity
            );
            HandlerResult::Ack
        }
        Err(err) => {
            eprintln!("inventory unavailable, retrying later: {err}");
            HandlerResult::retry()
        }
    }
}

The responder

The responder is an ordinary #[subscriber(.., publish(..))] handler: decode the request, return the reply. Err settles without replying, and the requester's timeout is the recovery mechanism:

// A plain publishing handler: decode the request, return the reply. `Err` settles without
// replying, and the requester's timeout is the recovery mechanism.
#[subscriber("inventory.check", publish("inventory.check.unrouted"))]
async fn check(req: &CheckStock) -> Result<Stock, HandlerResult> {
    if req.sku.is_empty() {
        return Err(HandlerResult::drop());
    }
    // Stand-in for a warehouse lookup.
    Ok(Stock {
        sku: req.sku.clone(),
        available: req.quantity <= 10,
    })
}

What makes it an RPC responder is the reply destination. DirectReplyTo is a ready-made publish transform that sends each reply back to the address the requester asked for and echoes its correlation id; compose it onto the reply publisher at mount time:

RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(broker, |b| {
    let replies = TypedPublisher::new(b.broker().publisher()).transform(DirectReplyTo);
    b.include_publishing(check, replies);
})

The handler stays a pure request-to-reply function, testable in-process like any other.

Semantics

• At-most-once. Direct reply-to keeps reply state in the requester's channel on one broker node; nothing is queued durably. A dropped requester channel loses in-flight replies, and the per-request timeout is the recovery mechanism. An unanswered request fails with a timeout error.
• Transient by default. Requests are published with delivery mode 1: a request nobody is waiting for after the timeout gains nothing from surviving a broker restart. Opt into persistence with .persistent(true) on the requester.
• No infrastructure. The pseudo-queue is never declared; the only real entity involved is the request queue the responder consumes. Responders on other stacks interoperate as long as they publish the reply to the received reply-to and echo correlation-id.