Request-reply

request attaches a dynamic reply link, sends with reply-to and correlation-id, and resolves with the matching reply or RequestTimeout.

Request-reply is part of AMQP 1.0, and AmqpPublisher implements the RequestReply capability. A request attaches a receiver link with a dynamic source, and the broker names a private address for that link. The message goes out with that address in reply-to and a fresh id in correlation-id. The call resolves with the first message on the link whose correlation-id matches, so a late answer to an earlier request is ignored. With no match inside the timeout the call returns RequestTimeout. The reply link is detached either way, so each request costs a link attach and a detach on top of the two messages.

A handler asks through a slot it binds as Out<impl RequestReply>. The mount site gives the slot the plain policy: .out(DefaultSlot, Publish).build(). The reply arrives already settled; the handler reads its payload and headers. A request that opens a conversation has no delivery to answer, so it runs from after_startup, where the publisher is already live. The other side of the exchange is a responder.

Request-reply asks a pricing service for the price of each order.

Example

Macros
Manual

examples/amqp_request_reply.rs
#[subscriber(AmqpAddress::queue("visitors"))]
async fn welcome(visitor: &Visitor, Out(greeter): Out<impl RequestReply>) -> HandlerOutcome {
    let request = OutgoingMessage::new("greeter", visitor.name.as_bytes());
    match greeter.request(request, Duration::from_secs(5)).await {
        Ok(reply) => {
            let greeting = String::from_utf8_lossy(reply.payload());
            tracing::info!(%greeting, "visitor welcomed");
            HandlerOutcome::ack()
        }
        // `RequestTimeout` when nothing answered in time; the visitor comes back for another try.
        Err(error) => {
            tracing::warn!(%error, visitor = %visitor.name, "the greeter did not answer");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/amqp_request_reply.rs
struct Welcome;

impl<Entry> Handle<Visitor, (), Outs<(Entry,)>> for Welcome
where
    Entry: OutEntry<DefaultSlot, Wire: RequestReply>,
{
    async fn handle(
        &self,
        visitor: &Visitor,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let request = OutgoingMessage::new("greeter", visitor.name.as_bytes());
        match outs
            .get(DefaultSlot)
            .request(request, Duration::from_secs(5))
            .await
        {
            Ok(reply) => {
                let greeting = String::from_utf8_lossy(reply.payload());
                tracing::info!(%greeting, "visitor welcomed");
                Ok(())
            }
            // `RequestTimeout` when nothing answered in time; the visitor comes back for another
            // try.
            Err(error) => {
                tracing::warn!(%error, visitor = %visitor.name, "the greeter did not answer");
                Err(HandlerOutcome::retry())
            }
        }
    }
}

The app mounts the responder and the requesting handler, each with its slot, and asks once at startup.

Macros
Manual

examples/amqp_request_reply.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("request-reply", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(greet).out(DefaultSlot, Publish).build();
            b.include(welcome).out(DefaultSlot, Publish).build();
            // A request that opens a conversation has no delivery to answer, so it runs once the
            // subscriptions are open.
            b.after_startup(Publish, async move |publisher| -> io::Result<()> {
                let request = OutgoingMessage::new("greeter", b"world".as_slice());
                let reply = publisher
                    .request(request, Duration::from_secs(5))
                    .await
                    .map_err(io::Error::other)?;
                let greeting = String::from_utf8_lossy(reply.payload());
                tracing::info!(%greeting, "the greeter is up");
                Ok(())
            });
        },
    )
}

examples/manual/amqp_request_reply.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("request-reply", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            let greet = subscriber(AmqpAddress::queue("greeter"), Greet);
            b.include(greet.build()).out(DefaultSlot, Publish).build();
            let welcome = subscriber(AmqpAddress::queue("visitors"), Welcome);
            b.include(welcome.build()).out(DefaultSlot, Publish).build();
            // A request that opens a conversation has no delivery to answer, so it runs once the
            // subscriptions are open.
            b.after_startup(Publish, async move |publisher| -> io::Result<()> {
                let request = OutgoingMessage::new("greeter", b"world".as_slice());
                let reply = publisher
                    .request(request, Duration::from_secs(5))
                    .await
                    .map_err(io::Error::other)?;
                let greeting = String::from_utf8_lossy(reply.payload());
                tracing::info!(%greeting, "the greeter is up");
                Ok(())
            });
        },
    )
}

Example​

Example