Skip to main content

Queues and topics

This guide reads a work queue, broadcasts on a topic, keeps a broadcast for a reader that is away, and publishes to an address.

AMQP 1.0 fixes the wire and leaves the meaning of an address to the broker. A subscription therefore says what it wants: AmqpAddress has a constructor for each intent, and the link tells the broker which one it is. What this page states is proved against ActiveMQ Artemis, the broker the crate's tests run on.

The example is an orders service. It takes orders from a queue and announces each taken order on a topic, which billing, an audit log and a ledger follow.

A work queue​

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

/// What every service that follows orders hears about a taken order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "order-events")]
struct OrderTaken {
    order: u64,
    amount_cents: u64,
}

#[subscriber(AmqpAddress::queue("orders"), reply)]
async fn take_order(order: &Order) -> OrderTaken {
    tracing::info!(order.id, "order taken");
    OrderTaken {
        order: order.id,
        amount_cents: order.amount_cents,
    }
}

AmqpAddress::queue asks for a queue: the link carries the terminus capability queue. Artemis then creates an anycast address orders with one queue behind it. Consumers on a queue compete, and each order goes to one of them. To take orders faster, run more replicas of the service: each replica is one more consumer of orders.

take_order answers each order with an OrderTaken. OrderTaken names its address with #[outgoing(name = "order-events")], so the bare reply clause sends it there.

A broadcast​

src/main.rs
#[subscriber(AmqpAddress::topic("order-events"))]
async fn bill(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, event.amount_cents, "invoice issued");
    HandlerOutcome::ack()
}

#[subscriber(AmqpAddress::topic("order-events"))]
async fn audit(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, "order recorded");
    HandlerOutcome::ack()
}

AmqpAddress::topic asks for a topic: the link carries the capability topic. Artemis makes order-events a multicast address and gives each subscription a queue of its own. So bill and audit each get a copy of every OrderTaken.

A topic subscription lives as long as its link. What is published to order-events while the service is down is not kept for it. A reader that must not miss a message reads a queue.

A broadcast that waits for its reader​

src/main.rs
/// Reads the queue `ledger` that the operator created on the `order-events` address.
#[subscriber(AmqpAddress::raw("order-events::ledger"))]
async fn post_to_ledger(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, "order posted to the ledger");
    HandlerOutcome::ack()
}

The ledger must see every order, also the ones taken while it was stopped. The operator creates a durable queue ledger on the multicast address order-events:

docker compose exec artemis /opt/activemq-artemis/bin/artemis queue create \
  --user artemis --password artemis --silent \
  --name ledger --address order-events --auto-create-address \
  --multicast --durable --preserve-on-no-consumers

Every message published to order-events now lands in ledger too, whether post_to_ledger runs or not. order-events::ledger is the queue's fully qualified name in Artemis: the address, two colons, the queue. AmqpAddress::raw sends an address as written, with no capability, so the link attaches to that queue. A bare #[subscriber("order-events::ledger")] does the same.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(take_order).out_reply(Publish);
            b.include(bill);
            b.include(audit);
            b.include(post_to_ledger);
        },
    )
}

.out_reply(Publish) names the publisher of the replies, AmqpPublish under its prelude name. A publish attaches a sender link to the reply's address on first use and keeps it. A message the broker refuses fails the publish with AmqpError::PublishNotAccepted.

Headers and other clients​

The well-known headers travel in the AMQP properties section: content-type, correlation-id, reply-to, message-id. Every other header travels in application-properties. A message from this crate is a plain AMQP message, so a client in another language reads it without RustStream.

The partition key travels as the PARTITION_KEY_HEADER header and becomes the message's group-id. Artemis uses group-id for message grouping: every message of one group goes to the same consumer of a queue. So orders keyed by customer keep their order across replicas.

Addresses on other brokers​

Other brokers name their nodes their own way, and raw passes their spelling as it is. Their own documentation describes it: RabbitMQ 4.x, Azure Service Bus, ActiveMQ Classic, which Amazon MQ runs.

Testing the routing​

The tests in the example file run the app in process. One order publishes one OrderTaken, orders is called once, and order-events is called twice, once for each topic subscription. The in-process broker delivers a raw address exactly as written, so its test publishes to order-events::ledger directly. Testing an AMQP service covers the harness.