ruststream_lapin/queue.rs
1//! The queue descriptor: what a subscription binds to and, optionally, expects to exist.
2
3use lapin::types::{AMQPValue, FieldTable, ShortString};
4use ruststream::SubscriptionSource;
5
6use crate::broker::LapinBroker;
7use crate::delay::Delay;
8use crate::error::AmqpError;
9use crate::exchange::RabbitExchange;
10use crate::subscriber::LapinSubscriber;
11
12/// The queue implementation selected at declaration time.
13///
14/// Only used when the broker declares topology; an existing queue keeps whatever type it was
15/// created with (`x-queue-type` cannot change after creation).
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
17#[non_exhaustive]
18pub enum QueueType {
19 /// The classic single-node queue implementation.
20 Classic,
21 /// The Raft-replicated quorum queue implementation; requires a durable queue.
22 Quorum,
23}
24
25impl QueueType {
26 pub(crate) fn as_str(self) -> &'static str {
27 match self {
28 Self::Classic => "classic",
29 Self::Quorum => "quorum",
30 }
31 }
32}
33
34/// Describes one queue subscription: the queue, its expected settings, and its bindings.
35///
36/// Descriptors describe the EXPECTED topology for routing; by default nothing is created on the
37/// broker (managing infrastructure is the user's job). Opt in to declaration per broker with
38/// [`declare_topology(true)`](LapinBroker::declare_topology).
39///
40/// # Examples
41///
42/// ```
43/// use ruststream_lapin::{QueueType, RabbitExchange, RabbitQueue};
44///
45/// let orders = RabbitQueue::new("orders")
46/// .queue_type(QueueType::Quorum)
47/// .bind(RabbitExchange::topic("events"), "order.*")
48/// .dead_letter_exchange("dead-letters");
49/// assert_eq!(orders.name(), "orders");
50/// ```
51#[derive(Debug, Clone, PartialEq)]
52pub struct RabbitQueue {
53 name: String,
54 durable: bool,
55 exclusive: bool,
56 auto_delete: bool,
57 queue_type: Option<QueueType>,
58 bindings: Vec<(RabbitExchange, String)>,
59 arguments: FieldTable,
60 prefetch: Option<u16>,
61 delay: Option<Delay>,
62}
63
64impl RabbitQueue {
65 /// Describes the queue `name` with the defaults: durable, shared, not auto-deleted.
66 #[must_use]
67 pub fn new(name: impl Into<String>) -> Self {
68 Self {
69 name: name.into(),
70 durable: true,
71 exclusive: false,
72 auto_delete: false,
73 queue_type: None,
74 bindings: Vec::new(),
75 arguments: FieldTable::default(),
76 prefetch: None,
77 delay: None,
78 }
79 }
80
81 /// Whether the queue survives a broker restart. Defaults to `true`.
82 #[must_use]
83 pub fn durable(mut self, durable: bool) -> Self {
84 self.durable = durable;
85 self
86 }
87
88 /// Whether the queue is exclusive to this connection. Defaults to `false`.
89 #[must_use]
90 pub fn exclusive(mut self, exclusive: bool) -> Self {
91 self.exclusive = exclusive;
92 self
93 }
94
95 /// Whether the queue is deleted when its last consumer disconnects. Defaults to `false`.
96 #[must_use]
97 pub fn auto_delete(mut self, auto_delete: bool) -> Self {
98 self.auto_delete = auto_delete;
99 self
100 }
101
102 /// The queue type to declare, overriding the broker-wide
103 /// [`default_queue_type`](LapinBroker::default_queue_type).
104 ///
105 /// When neither is set no `x-queue-type` argument is sent and the server default applies.
106 #[must_use]
107 pub fn queue_type(mut self, queue_type: QueueType) -> Self {
108 self.queue_type = Some(queue_type);
109 self
110 }
111
112 /// Binds the queue to `exchange` under `routing_key`.
113 ///
114 /// Call repeatedly for multiple bindings. Without any binding the queue only receives
115 /// messages published to the default exchange under the queue name.
116 #[must_use]
117 pub fn bind(mut self, exchange: RabbitExchange, routing_key: impl Into<String>) -> Self {
118 self.bindings.push((exchange, routing_key.into()));
119 self
120 }
121
122 /// Dead-letters rejected messages to `exchange` (the `x-dead-letter-exchange` argument).
123 ///
124 /// A handler returning drop settles with `basic.reject(requeue = false)`, which routes the
125 /// message there.
126 #[must_use]
127 pub fn dead_letter_exchange(mut self, exchange: impl Into<String>) -> Self {
128 self.arguments.insert(
129 ShortString::from("x-dead-letter-exchange"),
130 AMQPValue::LongString(exchange.into().into()),
131 );
132 self
133 }
134
135 /// Overrides the routing key dead-lettered messages carry (`x-dead-letter-routing-key`).
136 #[must_use]
137 pub fn dead_letter_routing_key(mut self, routing_key: impl Into<String>) -> Self {
138 self.arguments.insert(
139 ShortString::from("x-dead-letter-routing-key"),
140 AMQPValue::LongString(routing_key.into().into()),
141 );
142 self
143 }
144
145 /// Sets one raw declaration argument (`x-...`), passed through verbatim.
146 ///
147 /// # Panics
148 ///
149 /// Panics if `name` exceeds 255 bytes (the AMQP short-string limit); argument names are
150 /// compile-time constants in practice.
151 #[must_use]
152 pub fn argument(mut self, name: impl Into<String>, value: AMQPValue) -> Self {
153 self.arguments.insert(ShortString::from(name.into()), value);
154 self
155 }
156
157 /// Replaces the whole raw declaration argument table (`x-...` passthrough).
158 #[must_use]
159 pub fn arguments(mut self, arguments: FieldTable) -> Self {
160 self.arguments = arguments;
161 self
162 }
163
164 /// Caps unacknowledged deliveries in flight for this subscription (`basic.qos`),
165 /// overriding the broker-wide [`prefetch`](LapinBroker::prefetch).
166 ///
167 /// This is the back-pressure window for the subscriber stream. When neither is set the
168 /// server imposes no prefetch limit.
169 #[must_use]
170 pub fn prefetch(mut self, prefetch: u16) -> Self {
171 self.prefetch = Some(prefetch);
172 self
173 }
174
175 /// Makes `retry_after` / `nack_after` native, routing delayed redeliveries through a broker
176 /// waiting queue instead of the core in-process fallback.
177 ///
178 /// Without this the runtime handles a delay with its broker-agnostic deferred re-publish
179 /// (at-most-once over the delay window, held in the service). With it, a delayed message is
180 /// re-published to the [`Delay`] waiting queue with a per-message TTL and dead-lettered back
181 /// to this queue when it fires, so the delayed copy lives on the broker.
182 ///
183 /// The waiting queue is infrastructure: it is declared only when the broker opts into
184 /// [`declare_topology`](LapinBroker::declare_topology); otherwise provision it yourself.
185 #[must_use]
186 pub fn delay(mut self, delay: Delay) -> Self {
187 self.delay = Some(delay);
188 self
189 }
190
191 /// The queue name.
192 #[must_use]
193 pub fn name(&self) -> &str {
194 &self.name
195 }
196
197 pub(crate) fn is_durable(&self) -> bool {
198 self.durable
199 }
200
201 pub(crate) fn is_exclusive(&self) -> bool {
202 self.exclusive
203 }
204
205 pub(crate) fn is_auto_delete(&self) -> bool {
206 self.auto_delete
207 }
208
209 pub(crate) fn queue_type_or(&self, broker_default: Option<QueueType>) -> Option<QueueType> {
210 self.queue_type.or(broker_default)
211 }
212
213 pub(crate) fn bindings(&self) -> &[(RabbitExchange, String)] {
214 &self.bindings
215 }
216
217 pub(crate) fn declare_arguments(&self) -> &FieldTable {
218 &self.arguments
219 }
220
221 pub(crate) fn prefetch_or(&self, broker_default: Option<u16>) -> Option<u16> {
222 self.prefetch.or(broker_default)
223 }
224
225 pub(crate) fn delay_config(&self) -> Option<&Delay> {
226 self.delay.as_ref()
227 }
228}
229
230impl SubscriptionSource<LapinBroker> for RabbitQueue {
231 type Subscriber = LapinSubscriber;
232
233 fn name(&self) -> &str {
234 &self.name
235 }
236
237 async fn subscribe(self, broker: &LapinBroker) -> Result<Self::Subscriber, AmqpError> {
238 broker.subscribe(self).await
239 }
240}
241
242#[cfg(feature = "testing")]
243impl SubscriptionSource<crate::testing::LapinTestBroker> for RabbitQueue {
244 type Subscriber = crate::testing::LapinTestSubscriber;
245
246 fn name(&self) -> &str {
247 &self.name
248 }
249
250 async fn subscribe(
251 self,
252 broker: &crate::testing::LapinTestBroker,
253 ) -> Result<Self::Subscriber, AmqpError> {
254 broker.subscribe(self.name).await
255 }
256}