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¶

LapinRequest is the requester's policy; paired with the connection it produces a LapinRequester, which 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 call:

crates/ruststream-lapin/examples/lapin_rpc_client.rs

/// A typed RPC call over the raw requester: encode the request, await the correlated reply,
/// decode it.
async fn check_stock(
    requester: &LapinRequester,
    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 = requester
        .request(
            OutgoingMessage::new("inventory.check", payload.as_ref()),
            Duration::from_secs(2),
        )
        .await?;
    Ok(JsonCodec.decode(reply.payload())?)
}

Attach the policy at the mount site (b.include(handler).publisher(LapinRequest::default())) and the live requester arrives in the handler as an Out parameter, so a handler calls the other service in the middle of its 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:

crates/ruststream-lapin/examples/lapin_rpc_client.rs

// 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, Out(inventory): Out<LapinRequester>) -> HandlerResult {
    match check_stock(inventory, &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:

crates/ruststream-lapin/examples/lapin_rpc_server.rs

// 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:

crates/ruststream-lapin/examples/lapin_rpc_server.rs

RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(broker, |b| {
    // The reply publisher is a policy: it holds no connection, so it is declared here and
    // paired with the broker by the runtime at startup.
    let replies = TypedPublisher::new(LapinPublish::default()).transform(DirectReplyTo);
    b.include(check).publisher(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 policy.
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.