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

ruststream_lapin/
queue.rs
