queue

AmqpAddress::queue asks for the queue terminus: subscriptions on the address compete, and each message goes to one of them.

The descriptor attaches its receiver with the terminus capability queue on the link's source. ActiveMQ Artemis and the other brokers that read the capability treat the address as anycast. That is a work queue: two replicas of a service share the orders, and no order is handled twice. A delivery handed back with a retry goes to whichever receiver the broker picks next. Once a queue subscription has attached, the queue keeps what is published while no receiver is attached, so a replica that restarts finds its backlog. A delivery the process held unsettled when it stopped returns to the queue as well. In process, TestApp hands the messages of a queue address to its subscriptions in turn, as the example's test shows with two replicas.

Example

Macros
Manual

examples/amqp_subscriptions_queue.rs
/// Every replica of the service competes for `orders`: each order is fulfilled once.
#[subscriber(AmqpAddress::queue("orders"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order fulfilled");
    HandlerOutcome::ack()
}

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

examples/manual/amqp_subscriptions_queue.rs
/// Every replica of the service competes for `orders`: each order is fulfilled once.
struct Fulfil;

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

/// The queue terminus: the subscriptions on `orders` compete for its messages.
fn orders() -> AmqpAddress {
    AmqpAddress::queue("orders")
}

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

Example​

Example