queuey_core/
queue.rs

1//! Queue declarations: [`QueueSet`] and [`QueueConfig`].
2
3use std::time::Duration;
4
5use crate::retry::RetryPolicy;
6
7/// Static configuration of a single queue.
8#[derive(Debug, Clone, PartialEq)]
9pub struct QueueConfig {
10    /// Fully-qualified broker queue name (prefix already applied).
11    pub name: String,
12    /// Max unacknowledged messages per consumer.
13    pub prefetch: u16,
14    /// Default retry policy for jobs on this queue (a job may override).
15    pub retry: RetryPolicy,
16    /// Whether the queue survives broker restarts.
17    pub durable: bool,
18    /// Optional per-message TTL applied on publish.
19    pub message_ttl: Option<Duration>,
20    /// Number of priority levels the main queue supports (`x-max-priority` on RabbitMQ).
21    ///
22    /// `None` means the queue is not a priority queue and message priorities are
23    /// ignored by the broker. Deferred jobs (see [`crate::JobError::Deferred`]) are
24    /// republished with the highest level so they run ahead of the backlog.
25    pub max_priority: Option<u8>,
26}
27
28/// Default for [`QueueConfig::max_priority`]: RabbitMQ recommends at most 10 levels.
29pub const DEFAULT_MAX_PRIORITY: u8 = 10;
30
31impl QueueConfig {
32    /// Config for `name` with defaults: prefetch 16, durable, no retries, 10 priority levels.
33    pub fn new(name: impl Into<String>) -> Self {
34        Self {
35            name: name.into(),
36            prefetch: 16,
37            retry: RetryPolicy::default(),
38            durable: true,
39            message_ttl: None,
40            max_priority: Some(DEFAULT_MAX_PRIORITY),
41        }
42    }
43    /// Set the maximum number of unacknowledged messages per consumer.
44    pub fn prefetch(mut self, prefetch: u16) -> Self {
45        self.prefetch = prefetch;
46        self
47    }
48    /// Set the default retry policy for this queue.
49    pub fn retry(mut self, retry: RetryPolicy) -> Self {
50        self.retry = retry;
51        self
52    }
53    /// Set whether the queue survives broker restarts.
54    pub fn durable(mut self, durable: bool) -> Self {
55        self.durable = durable;
56        self
57    }
58    /// Set a per-message TTL applied on publish.
59    pub fn message_ttl(mut self, ttl: Duration) -> Self {
60        self.message_ttl = Some(ttl);
61        self
62    }
63    /// Set the number of priority levels; `0` turns priorities off (`max_priority = None`).
64    ///
65    /// Changing this for a queue that already exists on the broker is refused by
66    /// RabbitMQ (`PRECONDITION_FAILED`): delete the queue first.
67    pub fn max_priority(mut self, levels: u8) -> Self {
68        self.max_priority = (levels > 0).then_some(levels);
69        self
70    }
71}
72
73/// A closed set of queues, normally an enum with `#[derive(Queues)]` from the
74/// macros crate; the hand-written equivalent is:
75///
76/// ```
77/// use queuey_core::{QueueConfig, QueueSet, RetryPolicy};
78///
79/// #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
80/// enum AppQueues { Emails, Images }
81///
82/// impl QueueSet for AppQueues {
83///     fn all() -> &'static [Self] { &[AppQueues::Emails, AppQueues::Images] }
84///     fn name(&self) -> &'static str {
85///         match self { AppQueues::Emails => "myapp.emails", AppQueues::Images => "myapp.img" }
86///     }
87///     fn config(&self) -> QueueConfig {
88///         match self {
89///             AppQueues::Emails => QueueConfig::new(self.name()).prefetch(10),
90///             AppQueues::Images => QueueConfig::new(self.name()).retry(RetryPolicy::exponential(3)),
91///         }
92///     }
93/// }
94///
95/// assert_eq!(AppQueues::from_name("myapp.img"), Some(AppQueues::Images));
96/// assert_eq!(AppQueues::Emails.config().prefetch, 10);
97/// ```
98pub trait QueueSet:
99    Copy + Clone + Eq + std::hash::Hash + std::fmt::Debug + Send + Sync + 'static
100{
101    /// Every variant of the set, in declaration order.
102    fn all() -> &'static [Self];
103
104    /// Fully-qualified broker queue name for this variant.
105    fn name(&self) -> &'static str;
106
107    /// Full configuration for this variant.
108    fn config(&self) -> QueueConfig;
109
110    /// Look up a variant by its broker name.
111    fn from_name(name: &str) -> Option<Self> {
112        Self::all().iter().copied().find(|q| q.name() == name)
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use super::*;
119
120    #[test]
121    fn max_priority_defaults_to_ten_levels() {
122        assert_eq!(DEFAULT_MAX_PRIORITY, 10);
123        assert_eq!(
124            QueueConfig::new("q").max_priority,
125            Some(DEFAULT_MAX_PRIORITY)
126        );
127    }
128
129    #[test]
130    fn max_priority_zero_turns_priorities_off() {
131        assert_eq!(QueueConfig::new("q").max_priority(0).max_priority, None);
132    }
133
134    #[test]
135    fn max_priority_stores_the_number_of_levels() {
136        assert_eq!(QueueConfig::new("q").max_priority(5).max_priority, Some(5));
137        assert_eq!(
138            QueueConfig::new("q").max_priority(255).max_priority,
139            Some(255)
140        );
141        // Last call wins, including back to off.
142        assert_eq!(
143            QueueConfig::new("q").max_priority(5).max_priority(0),
144            QueueConfig::new("q").max_priority(0)
145        );
146    }
147
148    #[test]
149    fn the_other_builders_leave_max_priority_alone() {
150        let config = QueueConfig::new("q")
151            .prefetch(3)
152            .durable(false)
153            .message_ttl(Duration::from_secs(1));
154        assert_eq!(config.max_priority, Some(DEFAULT_MAX_PRIORITY));
155    }
156}
queuey_core/queue.rs

queuey_core/
queue.rs
