Subscriptions

AmqpAddress describes a subscription: the address, the terminus it asks for and its link settings; a bare name sends the address as written.

#[subscriber("orders")] opens a receiver on the address orders. The name reaches the broker as written, with no terminus capability, which makes it the same subscription as AmqpAddress::raw. AMQP 1.0 standardises the wire, not what an address means, so whether the subscriptions on one address compete or each get a copy is a question the broker answers. A descriptor asks the question explicitly: AmqpAddress::queue asks for competing consumers, AmqpAddress::topic for a copy each. A descriptor goes where the name goes, as the attribute's first argument. It also carries the link settings: the credit, the settle mode and the deadline of a batch. The manual form passes either one to subscriber(source, body) as its first argument. The subscription's name is its address: a test reads its deliveries under that name, and a deferred retry is published back to it. An empty address stops the service at startup with AmqpError::InvalidAddress.

Queues and topics builds a service on a work queue, a broadcast and a raw address.

Example

The address orders, read as written.

Macros

examples/amqp_subscriptions_name.rs

/// Reads the address `orders`; the name reaches the broker as written.
#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
    })
}

Manual

examples/manual/amqp_subscriptions_name.rs

/// Reads the address `orders`; the name reaches the broker as written.
struct Place;

impl Handle<Order> for Place {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order placed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Place).build());
    })
}