ruststream_lapin/queue/
mod.rs

1//! The queue descriptors: what a subscription binds to and, optionally, expects to exist.
2//!
3//! `RabbitMQ` has two queue implementations and they answer a retry differently, so each has a
4//! descriptor of its own. [`RabbitQueue`] is the classic queue: it counts nothing, so a
5//! registration's cap is the runtime's to apply with the copies it publishes. [`RabbitQuorumQueue`]
6//! counts the deliveries a message spends and carries a spent one away itself, so the same cap
7//! becomes the queue's own arguments and no copy is published at all.
8
9use ruststream::RetryDeclaration;
10
11use crate::error::AmqpError;
12
13mod classic;
14mod quorum;
15mod spec;
16
17pub use classic::RabbitQueue;
18pub use quorum::RabbitQuorumQueue;
19pub use spec::QueueSpec;
20pub(crate) use spec::{DEAD_LETTER_EXCHANGE, DEAD_LETTER_ROUTING_KEY, declared_arguments};
21// The in-process transport reads a declaration's arguments back the way a server reads them off
22// the queue, and nothing else in the crate needs them by these names.
23#[cfg(feature = "testing")]
24pub(crate) use spec::Binding;
25#[cfg(feature = "testing")]
26pub(crate) use spec::{DELIVERY_LIMIT, QueueKind};
27
28mod sealed {
29    /// Closes the list of queue descriptors: `RabbitMQ` has two queue implementations, and which
30    /// one a subscription consumes decides what a retry does.
31    pub trait Sealed {}
32
33    impl Sealed for super::RabbitQueue {}
34    impl Sealed for super::RabbitQuorumQueue {}
35}
36
37/// The queue a subscription consumes: [`RabbitQueue`] or [`RabbitQuorumQueue`].
38///
39/// Sealed. A queue's implementation is fixed when the queue is created, and what a retry does
40/// follows from it, so the two descriptors are the whole list. The methods are how the broker
41/// reads a descriptor; they are machinery, never called directly.
42pub trait QueueDescriptor: sealed::Sealed {
43    /// The settings the queue is declared and consumed with.
44    #[doc(hidden)]
45    fn spec(&self) -> &QueueSpec;
46
47    /// What the mount site declared about this registration's retries, where the queue is the one
48    /// that applies it. `None` on a queue whose retries the runtime applies.
49    #[doc(hidden)]
50    fn declared_retry(&self) -> Option<&RetryDeclaration>;
51
52    /// Refuses a declaration this queue cannot carry, before the subscription opens.
53    ///
54    /// # Errors
55    ///
56    /// Returns [`AmqpError::InvalidOptions`] when the registration declared half a retry policy,
57    /// or a whole one on a queue this service does not declare.
58    #[doc(hidden)]
59    fn check_retry(&self, declares_topology: bool) -> Result<(), AmqpError>;
60}
61
62impl QueueDescriptor for RabbitQueue {
63    fn spec(&self) -> &QueueSpec {
64        self.spec()
65    }
66
67    /// A classic queue counts nothing and has no delivery limit, so the declaration stays with
68    /// the runtime and reaches no queue argument.
69    fn declared_retry(&self) -> Option<&RetryDeclaration> {
70        None
71    }
72
73    fn check_retry(&self, _declares_topology: bool) -> Result<(), AmqpError> {
74        Ok(())
75    }
76}
77
78impl QueueDescriptor for RabbitQuorumQueue {
79    fn spec(&self) -> &QueueSpec {
80        self.spec()
81    }
82
83    fn declared_retry(&self) -> Option<&RetryDeclaration> {
84        Some(self.retry())
85    }
86
87    fn check_retry(&self, declares_topology: bool) -> Result<(), AmqpError> {
88        self.check_retry(declares_topology)
89    }
90}
ruststream_lapin/queue/mod.rs

ruststream_lapin/queue/
mod.rs
