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