queuey_rabbitmq/options.rs
1//! Tunables for [`RabbitMqBackend`](crate::RabbitMqBackend).
2
3use std::time::Duration;
4
5use lapin::ConnectionProperties;
6
7use crate::topology::{DEFAULT_DEAD_SUFFIX, DEFAULT_DEFERRED_SUFFIX, DEFAULT_RETRY_SUFFIX};
8
9/// Default for [`RabbitMqOptions::deferred_granularity`].
10const DEFAULT_DEFERRED_GRANULARITY: Duration = Duration::from_secs(1);
11
12/// Configuration for [`RabbitMqBackend::with_options`](crate::RabbitMqBackend::with_options).
13///
14/// ```
15/// use queuey_rabbitmq::RabbitMqOptions;
16///
17/// let options = RabbitMqOptions::default()
18/// .retry_suffix("-wait")
19/// .dead_suffix("-dlq")
20/// .declare_dead_letter_queues(false);
21/// assert_eq!(options.retry_suffix, "-wait");
22/// ```
23#[derive(Clone, Debug)]
24pub struct RabbitMqOptions {
25 /// Handshake properties passed to `lapin::Connection::connect`.
26 pub connection_properties: ConnectionProperties,
27
28 /// Suffix appended to a queue name to name its retry (wait) queue.
29 ///
30 /// Defaults to [`DEFAULT_RETRY_SUFFIX`] (`".retry"`).
31 pub retry_suffix: String,
32
33 /// Suffix appended to a queue name to name its dead-letter queue.
34 ///
35 /// Defaults to [`DEFAULT_DEAD_SUFFIX`] (`".dead"`).
36 pub dead_suffix: String,
37
38 /// Whether [`declare`](queuey_core::Backend::declare) also declares
39 /// the `q.dead` queues.
40 ///
41 /// Defaults to `true`. Set to `false` when dead-letter queues are managed
42 /// out of band (policies, an operator-owned topology, a different broker
43 /// vhost).
44 ///
45 /// This flag also selects *how* a message is dead-lettered, because this
46 /// backend never publishes to a queue it does not own:
47 ///
48 /// * `true`: [`Delivery::dead_letter`](queuey_core::Delivery::dead_letter)
49 /// publishes the envelope to `q.dead` with the `x-death-*` headers and
50 /// then acks the original.
51 /// * `false`: nothing is published. The original is rejected with
52 /// `requeue = false`, so the broker applies whatever
53 /// `x-dead-letter-exchange` policy the operator put on `q`, and drops the
54 /// message if there is none. The reason is logged at `WARN`, since it is
55 /// not recorded anywhere else.
56 ///
57 /// The same choice governs a message whose body is not a valid envelope.
58 pub declare_dead_letter_queues: bool,
59
60 /// Infix between a queue name and a hold queue's TTL.
61 ///
62 /// Defaults to [`DEFAULT_DEFERRED_SUFFIX`] (`".deferred"`), so a 30-second
63 /// deferral of `myapp.emails` waits in `myapp.emails.deferred.30000`. See
64 /// [`crate::topology`] for why the TTL is part of the name.
65 pub deferred_suffix: String,
66
67 /// Step that deferral delays are rounded **up** to.
68 ///
69 /// Defaults to one second. Every distinct rounded delay gets its own hold
70 /// queue, so this is the knob that trades precision for the number of queues
71 /// on the broker: with the default, `Retry-After: 30` and a computed `29.2s`
72 /// delay share `q.deferred.30000`, and no deferral can create more than
73 /// `MAX_TTL_MS / 1000` queues per work queue.
74 ///
75 /// A deferral is never released *early*: rounding is always up, a delay
76 /// shorter than the granularity still waits one full step, and a delay that
77 /// rounds up past
78 /// [`MAX_DEFERRAL_MS`](crate::topology::MAX_DEFERRAL_MS) (~24.8 days) is
79 /// refused instead of being shortened.
80 ///
81 /// A zero (or sub-millisecond) value is clamped to one millisecond by
82 /// [`deferred_ttl_ms`](crate::topology::deferred_ttl_ms) rather than
83 /// rejected, because a backend constructor must not panic on a config value, but
84 /// one millisecond of granularity means up to one hold queue per distinct
85 /// millisecond, which is almost never what you want.
86 ///
87 /// Note what is *not* here: nothing tunes a hold queue's `x-expires`. Its
88 /// arguments are a pure function of its name (`x-expires = 2 * ttl`), so two
89 /// processes configured differently still agree on `q.deferred.30000`
90 /// instead of locking each other out with `PRECONDITION_FAILED`. The
91 /// granularity is safe to tune because it only changes *which* hold queue a
92 /// delay lands in, never that queue's arguments.
93 pub deferred_granularity: Duration,
94}
95
96impl Default for RabbitMqOptions {
97 fn default() -> Self {
98 Self {
99 connection_properties: ConnectionProperties::default(),
100 retry_suffix: DEFAULT_RETRY_SUFFIX.to_owned(),
101 dead_suffix: DEFAULT_DEAD_SUFFIX.to_owned(),
102 declare_dead_letter_queues: true,
103 deferred_suffix: DEFAULT_DEFERRED_SUFFIX.to_owned(),
104 deferred_granularity: DEFAULT_DEFERRED_GRANULARITY,
105 }
106 }
107}
108
109impl RabbitMqOptions {
110 /// Replace the connection handshake properties.
111 #[must_use]
112 pub fn connection_properties(mut self, properties: ConnectionProperties) -> Self {
113 self.connection_properties = properties;
114 self
115 }
116
117 /// Replace the retry queue suffix.
118 #[must_use]
119 pub fn retry_suffix(mut self, suffix: impl Into<String>) -> Self {
120 self.retry_suffix = suffix.into();
121 self
122 }
123
124 /// Replace the dead-letter queue suffix.
125 #[must_use]
126 pub fn dead_suffix(mut self, suffix: impl Into<String>) -> Self {
127 self.dead_suffix = suffix.into();
128 self
129 }
130
131 /// Enable or disable declaring `q.dead` queues.
132 #[must_use]
133 pub fn declare_dead_letter_queues(mut self, declare: bool) -> Self {
134 self.declare_dead_letter_queues = declare;
135 self
136 }
137
138 /// Replace the hold queue infix.
139 #[must_use]
140 pub fn deferred_suffix(mut self, suffix: impl Into<String>) -> Self {
141 self.deferred_suffix = suffix.into();
142 self
143 }
144
145 /// Replace the step deferral delays are rounded up to.
146 ///
147 /// A zero or sub-millisecond value is *clamped* to one millisecond when the
148 /// TTL is computed, not rejected here: this is a builder, and library code
149 /// does not panic on configuration.
150 #[must_use]
151 pub fn deferred_granularity(mut self, granularity: Duration) -> Self {
152 self.deferred_granularity = granularity;
153 self
154 }
155}
156
157#[cfg(test)]
158mod tests {
159 use super::*;
160
161 #[test]
162 fn defaults_match_the_documented_topology() {
163 let options = RabbitMqOptions::default();
164 assert_eq!(options.retry_suffix, ".retry");
165 assert_eq!(options.dead_suffix, ".dead");
166 assert!(options.declare_dead_letter_queues);
167 assert_eq!(options.deferred_suffix, ".deferred");
168 assert_eq!(options.deferred_granularity, Duration::from_secs(1));
169 }
170
171 #[test]
172 fn deferral_tunables_can_be_overridden() {
173 let options = RabbitMqOptions::default()
174 .deferred_suffix("-hold")
175 .deferred_granularity(Duration::from_millis(250));
176 assert_eq!(options.deferred_suffix, "-hold");
177 assert_eq!(options.deferred_granularity, Duration::from_millis(250));
178 // And they are independent of the retry / dead-letter tunables.
179 assert_eq!(options.retry_suffix, ".retry");
180 assert_eq!(options.dead_suffix, ".dead");
181 }
182
183 #[test]
184 fn a_zero_granularity_is_accepted_and_clamped_later_not_panicked_on() {
185 let options = RabbitMqOptions::default().deferred_granularity(Duration::ZERO);
186 assert_eq!(options.deferred_granularity, Duration::ZERO);
187 // The clamp lives in `deferred_ttl_ms`, so nothing here can panic.
188 assert_eq!(
189 crate::topology::deferred_ttl_ms(
190 Duration::from_millis(7),
191 options.deferred_granularity
192 ),
193 Some(7)
194 );
195 }
196
197 #[test]
198 fn suffixes_can_be_overridden() {
199 let options = RabbitMqOptions::default()
200 .retry_suffix("-wait")
201 .dead_suffix("-dlq");
202 assert_eq!(options.retry_suffix, "-wait");
203 assert_eq!(options.dead_suffix, "-dlq");
204 assert!(options.declare_dead_letter_queues);
205 }
206
207 #[test]
208 fn dead_letter_declaration_can_be_disabled() {
209 let options = RabbitMqOptions::default().declare_dead_letter_queues(false);
210 assert!(!options.declare_dead_letter_queues);
211 // Turning declaration off must not change the names.
212 assert_eq!(options.dead_suffix, ".dead");
213 }
214
215 #[test]
216 fn connection_properties_can_be_replaced() {
217 let options = RabbitMqOptions::default()
218 .connection_properties(ConnectionProperties::default().with_locale("nl_NL".into()));
219 assert!(format!("{:?}", options.connection_properties).contains("nl_NL"));
220 }
221}