DirectReplyTo

DirectReplyTo is a reply transform: it sends each reply to the address the request named in reply-to and echoes the request's correlation-id.

It mounts on the reply position: .out_reply(policy).transform(DirectReplyTo). The handler is an ordinary reply handler and knows nothing about the requester. The transform names the destination of every reply, so it mounts only over a reply type that declares none. Over a type with its own #[outgoing(name = "..")] the mount does not compile. The name in the mount site's reply("..") is the fallback: a request without reply-to is answered there, and the generated AsyncAPI document reports it. A LapinRequest requester and a requester on another stack both name the reply address this way.

Example

The inventory service answers every stock check at the address the request named. A request without one is answered at inventory.check.unrouted.

Macros
Manual

examples/lapin_request_direct_reply_to.rs
/// The reply goes wherever the request asks, so its type declares no destination.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Stock {
    sku: String,
    available: bool,
}

#[subscriber(RabbitQueue::new("inventory.check"), reply("inventory.check.unrouted"))]
async fn check(ask: &CheckStock) -> Stock {
    Stock {
        sku: ask.sku.clone(),
        available: ask.sku != "unobtainium",
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(check)
                .out_reply(Publish::default())
                .transform(DirectReplyTo);
        },
    )
}

examples/manual/lapin_request_direct_reply_to.rs
/// The reply goes wherever the request asks, so its type declares no destination.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Stock {
    sku: String,
    available: bool,
}

struct Check;

impl Handle<CheckStock, Stock> for Check {
    fn handle(
        &self,
        ask: &CheckStock,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Stock, HandlerOutcome>> {
        ready(Ok(Stock {
            sku: ask.sku.clone(),
            available: ask.sku != "unobtainium",
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            let check = subscriber(RabbitQueue::new("inventory.check"), Check);
            b.include(check.reply().to("inventory.check.unrouted").build())
                .out_reply(Publish::default())
                .transform(DirectReplyTo);
        },
    )
}

Example​

Example