AmqpContext

Ctx keys give a handler the exchange, routing key, redelivered flag and delivery tag; Context<'_, AmqpContext> reads them all.

AmqpContext carries what a delivery has beside its payload and its headers. The subscription builds it once per delivery. A handler binds one field with Ctx<K>, where K is a key of context::keys:

Key	Value	What it is
`Exchange`	`String`	the exchange the message was published to, empty for the default exchange
`RoutingKey`	`String`	the routing key the message was published with
`Redelivered`	`bool`	whether the server handed the message out before
`DeliveryTag`	`u64`	the delivery's number on its channel

The four keys are in the prelude. A body that reads several fields takes ctx: &mut Context<'_, AmqpContext> instead and calls ctx.context(key) per field, as the manual form does; there the exchange and the routing key are borrowed &str. AmqpContext itself comes from ruststream_lapin::context.

The routing key tells apart the messages a topic binding takes. Redelivered is set on a delivery a classic queue hands out again after a retry; a quorum queue counts deliveries instead. The delivery tag numbers the deliveries of one channel, so it identifies a delivery only within its subscription.

Example

The handler replies with where each order came from.

Macros
Manual

examples/lapin_context.rs
use ruststream_lapin::prelude::*;

/// Records where each order came from: its exchange and routing key, and whether the server had
/// handed it out before.
#[subscriber(RabbitQueue::new("orders"), reply)]
async fn audit(
    order: &Order,
    Ctx(exchange): Ctx<Exchange>,
    Ctx(routing_key): Ctx<RoutingKey>,
    Ctx(redelivered): Ctx<Redelivered>,
    Ctx(delivery_tag): Ctx<DeliveryTag>,
) -> AuditRecord {
    tracing::info!(order.id, %exchange, %routing_key, redelivered, delivery_tag, "order audited");
    AuditRecord {
        order_id: order.id,
        exchange,
        routing_key,
        redelivered,
    }
}

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

examples/manual/lapin_context.rs
use ruststream_lapin::prelude::*;

/// Records where each order came from: its exchange and routing key, and whether the server had
/// handed it out before.
struct Audit;

impl Handle<Order, AuditRecord, (), AmqpContext> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, AmqpContext>,
    ) -> impl Future<Output = Result<AuditRecord, HandlerOutcome>> {
        let exchange = ctx.context(Exchange);
        let routing_key = ctx.context(RoutingKey);
        let redelivered = ctx.context(Redelivered);
        let delivery_tag = ctx.context(DeliveryTag);
        tracing::info!(order.id, %exchange, %routing_key, redelivered, delivery_tag, "order audited");
        ready(Ok(AuditRecord {
            order_id: order.id,
            exchange: exchange.to_owned(),
            routing_key: routing_key.to_owned(),
            redelivered,
        }))
    }
}

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

Example​

Example