Channels

A subscription's channel is its queue with durability and exclusivity; a publish names a routing key on an exchange, or a queue.

A subscription's channel binding is is: queue. Its queue object carries the name and the durable, exclusive and autoDelete settings of the RabbitQueue descriptor.

A channel the service publishes to is described from its publish policy. Its name is the destination the mount site resolved: a reply's name, the name of a slot entry, or a dead_letter(..) declaration. On a policy with a named exchange, the name is the routing key: the binding is is: routingKey with the exchange's name. On the default exchange the routing key is the queue a message lands in: the binding is is: queue with that queue's name. A queue's settings belong to the side that declares it, so this binding carries the name alone.

A reply under DirectReplyTo goes to the address each request carries. Its channel therefore has no address, and the receive operation names the header the address is read from.

Example

The channel inventory.check reports a transient, exclusive, auto-delete queue. The reply channel inventory.unrouted reports the exchange answers, and the dead-letter channel inventory.dead reports the queue of that name.

Macros
Manual

examples/lapin_asyncapi.rs
#[subscriber(
    RabbitQueue::new("inventory.check").durable(false).exclusive(true).auto_delete(true),
    reply("inventory.unrouted")
)]
async fn check(ask: &Ask) -> Stock {
    Stock {
        available: !ask.sku.is_empty(),
    }
}

fn service(broker: LapinBroker) -> RustStream {
    let inventory = AppInfo::new("inventory", "1.0.0");
    RustStream::new(inventory).with_broker_labeled("rabbit", broker, |b| {
        b.include(check)
            .max_attempts(nonzero!(3u32))
            .dead_letter("inventory.dead")
            .out_reply(Publish::default().exchange("answers"))
            .transform(DirectReplyTo);
        b.include(on_event)
            .out(
                Restocks,
                Publish::default()
                    .priority(4)
                    .expiration(Duration::from_secs(30)),
            )
            .build();
    })
}

examples/manual/lapin_asyncapi.rs
/// A queue that lives as long as this service's connection.
fn checks() -> RabbitQueue {
    RabbitQueue::new("inventory.check")
        .durable(false)
        .exclusive(true)
        .auto_delete(true)
}

fn service(broker: LapinBroker) -> RustStream {
    let inventory = AppInfo::new("inventory", "1.0.0");
    RustStream::new(inventory).with_broker_labeled("rabbit", broker, |b| {
        let check = subscriber(checks(), Check).reply().to("inventory.unrouted");
        b.include(check.build())
            .max_attempts(nonzero!(3u32))
            .dead_letter("inventory.dead")
            .out_reply(Publish::default().exchange("answers"))
            .transform(DirectReplyTo);
        b.include(subscriber(events(), OnEvent).build())
            .out(
                Restocks,
                Publish::default()
                    .priority(4)
                    .expiration(Duration::from_secs(30)),
            )
            .build();
    })
}

Example​

Example