Queues

A queue set is a fieldless enum that derives Queues. One variant is one broker queue, and the attributes on it are the complete configuration of that queue: its name, how much a consumer prefetches, whether it survives a broker restart, its message TTL, its priority levels and its default retry policy.

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Queues)]
#[queues(prefix = "myapp")]
enum AppQueues {
    #[queue(prefetch = 10)]
    Emails,                                   // broker name "myapp.emails"
    #[queue(name = "img", message_ttl = "30s", max_priority = 0)]
    ImageResize,                              // broker name "myapp.img"
}

Naming

The broker name of a variant is its name snake-cased, with the set’s prefix in front of it if there is one:

Variant	`#[queues(prefix = ...)]`	`#[queue(name = ...)]`	Broker queue
`Emails`	none	none	`emails`
`Emails`	`"myapp"`	none	`myapp.emails`
`ImageResize`	`"myapp"`	none	`myapp.image_resize`
`ImageResize`	`"myapp"`	`"img"`	`myapp.img`
`HTTPCalls`	none	none	`http_calls`

Names have to be unique after the prefix is applied. Two variants that resolve to the same string are a compile error that points at the second one and notes where the first was used.

The attributes

#[queues(...)] on the enum, both keys optional:

Key	Meaning
`prefix = "myapp"`	Queue names become `"myapp.<name>"`
`crate = "path"`	Where generated code finds the core crate. Normally unnecessary

#[queue(...)] on each variant, every key optional:

Key	Default	Meaning
`name = "img"`	snake_case of the variant	The queue’s name, before the prefix
`prefetch = 10`	`16`	Unacknowledged messages per consumer, `1..=65535`
`durable = true`	`true`	Whether the queue survives a broker restart
`message_ttl = "30s"`	none	Per-message TTL applied on publish
`max_priority = 10`	`10`	Priority levels the queue is declared with; `0` turns priorities off
`retry(...)`	no retries	Default retry policy for jobs on this queue

The full key-by-key reference, with ranges and the exact compile error for each mistake, is in Queue attributes.

Prefetch

prefetch is how many messages one consumer may hold unacknowledged. On RabbitMQ it becomes basic_qos(prefetch, global = false) on the consuming channel.

It is per queue, not per process. A worker’s own cap on jobs running at once is concurrency, which defaults to the sum of the prefetch of every queue it consumes. See Worker.

prefetch = 0 is rejected at compile time: zero means unlimited in AMQP, and a silently unbounded consumer is never what someone meant to write. Omit the key to get the default of 16.

Priority levels

max_priority is the number of priority levels the queue is declared with. It exists for one reason: a deferred job comes back at the highest level its queue knows, ahead of the backlog that built up while it waited.

Attribute	Meaning
`#[queue(max_priority = 10)]`	Ten levels; deferred jobs return at `10`. This is the default, `DEFAULT_MAX_PRIORITY`
`#[queue(max_priority = 0)]`	Not a priority queue; the broker ignores priorities and deferred jobs come back FIFO

Message TTL

message_ttl is applied to the main queue as x-message-ttl, so the broker discards a message that has been sitting there longer than that. Durations are string literals parsed at compile time; zero is rejected, because a zero TTL discards every message the moment it is published.

Reading the configuration back

The derive generates an ordinary impl QueueSet, so the configuration is available at runtime and in tests:

pub trait QueueSet:
    Copy + Clone + Eq + std::hash::Hash + std::fmt::Debug + Send + Sync + 'static
{
    fn all() -> &'static [Self];
    fn name(&self) -> &'static str;
    fn config(&self) -> QueueConfig;
    fn from_name(name: &str) -> Option<Self>;
}

config() returns a QueueConfig, which is also constructible by hand if you are writing a backend or a test fixture rather than an application:

pub struct QueueConfig {
    pub name: String,
    pub prefetch: u16,
    pub retry: RetryPolicy,
    pub durable: bool,
    pub message_ttl: Option<Duration>,
    pub max_priority: Option<u8>,
}

QueueConfig::new(name) defaults to prefetch 16, RetryPolicy::default() (which is no retries), durable, no TTL, and Some(10) priority levels. The builders prefetch, retry, durable, message_ttl and max_priority each take and return self; max_priority(0) sets the field to None.

Selecting queues per process

A worker consumes every queue in the set by default. To split queues across processes, name the ones this process should take:

let worker = Worker::<AppQueues, _>::builder(backend)
    .handler(EmailHandler)
    .queues(&[AppQueues::Emails])
    .build()
    .await?;

build() declares only the selected queues, and de-duplicates them by resolved name.