LapinRequest

LapinRequest, Request in the prelude, gives a handler a requester: it publishes a request over RabbitMQ direct reply-to and awaits the correlated reply.

The handler bounds its slot Out<impl RequestReply> and calls request with a message and a timeout. At startup the runtime pairs the policy into a LapinRequester. Each request goes out with reply-to set to the amq.rabbitmq.reply-to pseudo-queue and a generated correlation-id. The reply that carries that id resolves the call, and a reply with an id nobody waits for is dropped. No reply within the timeout fails the call with AmqpError::RequestTimeout. One task per requester reads the replies, started by its first request on the runtime the broker connected on. The request carries bytes, so the example encodes the question and decodes the answer with the core's JSON codec.

The policy takes exchange, priority and expiration like a publish policy. Requests are transient by default, and persistent(true) makes them persistent. Direct reply-to is at-most-once: the reply state lives on the requester's channel on one node, and nothing is queued durably. A channel that drops loses the replies in flight, and the caller sees a timeout. The responder is any service that answers at the received reply-to with the same correlation-id, as DirectReplyTo does.

Example

An order asks the inventory service whether its item is in stock. An order in stock is accepted, one out of stock is dropped, and no answer brings the order back.

Macros
Manual

examples/lapin_request_request.rs
/// Asks the inventory service whether `sku` is in stock, and waits two seconds at most.
async fn in_stock(
    inventory: &impl RequestReply,
    sku: &str,
) -> Result<bool, Box<dyn Error + Send + Sync>> {
    let request = JsonCodec.encode(&CheckStock {
        sku: sku.to_owned(),
    })?;
    let ask = OutgoingMessage::new("inventory.check", request.as_ref());
    let reply = inventory.request(ask, Duration::from_secs(2)).await?;
    let stock: Stock = JsonCodec.decode(reply.payload())?;
    Ok(stock.available)
}

/// An order in stock is accepted and one out of stock is dropped. An inventory service that does
/// not answer brings the order back.
#[subscriber(RabbitQueue::new("orders"))]
async fn place_order(order: &Order, Out(inventory): Out<impl RequestReply>) -> HandlerOutcome {
    match in_stock(inventory, &order.sku).await {
        Ok(true) => HandlerOutcome::ack(),
        Ok(false) => HandlerOutcome::drop(),
        Err(err) => {
            tracing::warn!(order.id, %err, "inventory unavailable");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/lapin_request_request.rs
/// Asks the inventory service whether `sku` is in stock, and waits two seconds at most.
async fn in_stock(
    inventory: &impl RequestReply,
    sku: &str,
) -> Result<bool, Box<dyn Error + Send + Sync>> {
    let request = JsonCodec.encode(&CheckStock {
        sku: sku.to_owned(),
    })?;
    let ask = OutgoingMessage::new("inventory.check", request.as_ref());
    let reply = inventory.request(ask, Duration::from_secs(2)).await?;
    let stock: Stock = JsonCodec.decode(reply.payload())?;
    Ok(stock.available)
}

/// An order in stock is accepted and one out of stock is dropped. An inventory service that does
/// not answer brings the order back.
struct PlaceOrder;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for PlaceOrder
where
    Egress: OutEntry<DefaultSlot, Wire: RequestReply>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        match in_stock(outs.get(DefaultSlot), &order.sku).await {
            Ok(true) => Ok(()),
            Ok(false) => Err(HandlerOutcome::drop()),
            Err(err) => {
                tracing::warn!(order.id, %err, "inventory unavailable");
                Err(HandlerOutcome::retry())
            }
        }
    }
}

The app mounts the requester's policy on the handler's slot. The inventory service runs in the same app here.

Macros
Manual

examples/lapin_request_request.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(place_order)
                .out(DefaultSlot, Request::default())
                .build();
            b.include(check_stock)
                .out_reply(Publish::default())
                .transform(DirectReplyTo);
        },
    )
}

examples/manual/lapin_request_request.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            let check = subscriber(RabbitQueue::new("inventory.check"), CheckStockBody);
            b.include(subscriber(RabbitQueue::new("orders"), PlaceOrder).build())
                .out(DefaultSlot, Request::default())
                .build();
            b.include(check.reply().to("inventory.check.unrouted").build())
                .out_reply(Publish::default())
                .transform(DirectReplyTo);
        },
    )
}

Example​

Example