ruststream_lapin/exchange.rs
1//! The exchange half of a queue binding descriptor.
2
3use lapin::ExchangeKind;
4
5/// Describes the exchange side of a [`RabbitQueue`](crate::RabbitQueue) binding.
6///
7/// Like every descriptor in this crate it only records the EXPECTED topology; nothing is
8/// declared unless the broker was built with
9/// [`declare_topology(true)`](crate::LapinBroker::declare_topology).
10///
11/// # Examples
12///
13/// ```
14/// use ruststream_lapin::prelude::*;
15/// use serde::Deserialize;
16///
17/// #[derive(Deserialize)]
18/// struct OrderPlaced {
19/// id: u64,
20/// }
21///
22/// #[subscriber(RabbitQueue::new("orders")
23/// .bind(RabbitExchange::topic("events").durable(true), "order.*"))]
24/// async fn on_order(event: &OrderPlaced) -> HandlerOutcome {
25/// tracing::info!(event.id, "order placed");
26/// HandlerOutcome::ack()
27/// }
28///
29/// fn app() -> RustStream {
30/// let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
31/// RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
32/// b.include(on_order);
33/// })
34/// }
35/// ```
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct RabbitExchange {
38 name: String,
39 kind: ExchangeKind,
40 durable: bool,
41 auto_delete: bool,
42}
43
44impl RabbitExchange {
45 fn new(name: impl Into<String>, kind: ExchangeKind) -> Self {
46 Self {
47 name: name.into(),
48 kind,
49 durable: true,
50 auto_delete: false,
51 }
52 }
53
54 /// A direct exchange: routes on an exact routing-key match.
55 #[must_use]
56 pub fn direct(name: impl Into<String>) -> Self {
57 Self::new(name, ExchangeKind::Direct)
58 }
59
60 /// A topic exchange: routes on dot-separated routing-key patterns (`order.*`).
61 #[must_use]
62 pub fn topic(name: impl Into<String>) -> Self {
63 Self::new(name, ExchangeKind::Topic)
64 }
65
66 /// A fanout exchange: routes every message to every bound queue.
67 #[must_use]
68 pub fn fanout(name: impl Into<String>) -> Self {
69 Self::new(name, ExchangeKind::Fanout)
70 }
71
72 /// A headers exchange: routes on the binding's arguments instead of the routing key.
73 ///
74 /// A queue binds to one with [`bind_with`](crate::RabbitQueue::bind_with), whose argument
75 /// table carries `x-match` (`all` or `any`) and the headers a message has to match. The
76 /// plain [`bind`](crate::RabbitQueue::bind) sends no arguments, and an empty table here
77 /// matches every message the exchange gets rather than none.
78 ///
79 /// # Examples
80 ///
81 /// ```
82 /// use ruststream_lapin::prelude::*;
83 /// use serde::Deserialize;
84 ///
85 /// #[derive(Deserialize)]
86 /// struct Order {
87 /// id: u64,
88 /// }
89 ///
90 /// // The binding arguments: how to match, then the headers matched on.
91 /// fn gold_in_eu(x_match: &str) -> FieldTable {
92 /// let mut arguments = FieldTable::default();
93 /// arguments.insert("x-match".into(), AMQPValue::LongString(x_match.into()));
94 /// arguments.insert("region".into(), AMQPValue::LongString("eu".into()));
95 /// arguments.insert("tier".into(), AMQPValue::LongString("gold".into()));
96 /// arguments
97 /// }
98 ///
99 /// #[subscriber(RabbitQueue::new("orders.eu.gold")
100 /// .bind_with(RabbitExchange::headers("orders"), "", gold_in_eu("all")))]
101 /// async fn on_gold_in_eu(order: &Order) -> HandlerOutcome {
102 /// tracing::info!(order.id, "gold order in the EU");
103 /// HandlerOutcome::ack()
104 /// }
105 ///
106 /// // One header is enough here, where the binding above needs both.
107 /// #[subscriber(RabbitQueue::new("orders.eu.or.gold")
108 /// .bind_with(RabbitExchange::headers("orders"), "", gold_in_eu("any")))]
109 /// async fn on_gold_or_eu(order: &Order) -> HandlerOutcome {
110 /// tracing::info!(order.id, "gold order or an order in the EU");
111 /// HandlerOutcome::ack()
112 /// }
113 ///
114 /// fn app() -> RustStream {
115 /// let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
116 /// RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
117 /// b.include(on_gold_in_eu);
118 /// b.include(on_gold_or_eu);
119 /// })
120 /// }
121 /// ```
122 #[must_use]
123 pub fn headers(name: impl Into<String>) -> Self {
124 Self::new(name, ExchangeKind::Headers)
125 }
126
127 /// An exchange of a plugin-provided type, for example `"x-delayed-message"`.
128 ///
129 /// A plugin that routes by binding arguments takes them through
130 /// [`bind_with`](crate::RabbitQueue::bind_with).
131 #[must_use]
132 pub fn custom(name: impl Into<String>, kind: impl Into<String>) -> Self {
133 Self::new(name, ExchangeKind::Custom(kind.into()))
134 }
135
136 /// A consistent-hash exchange (the `x-consistent-hash` type from the
137 /// [`rabbitmq_consistent_hash_exchange`](https://github.com/rabbitmq/rabbitmq-server/tree/main/deps/rabbitmq_consistent_hash_exchange)
138 /// plugin): distributes messages across the queues bound to it by hashing the routing key,
139 /// so partition-like fan-out is a server-side concern rather than a client one.
140 ///
141 /// Each queue binds with a routing key that is its integer weight (`"1"`, `"2"`, ...); the
142 /// broker splits the hash space proportionally. Requires the plugin to be enabled on the
143 /// broker, so it lives behind the `plugin-consistent-hash` feature.
144 ///
145 /// # Examples
146 ///
147 /// ```
148 /// use ruststream_lapin::prelude::*;
149 /// use serde::Deserialize;
150 ///
151 /// #[derive(Deserialize)]
152 /// struct Order {
153 /// id: u64,
154 /// }
155 ///
156 /// // This queue binds with weight 1, so it takes its share of the hash space.
157 /// #[subscriber(RabbitQueue::new("orders.shard-a")
158 /// .bind(RabbitExchange::consistent_hash("orders-by-key"), "1"))]
159 /// async fn on_order(order: &Order) -> HandlerOutcome {
160 /// tracing::info!(order.id, "shard a took the order");
161 /// HandlerOutcome::ack()
162 /// }
163 ///
164 /// fn app() -> RustStream {
165 /// let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
166 /// RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
167 /// b.include(on_order);
168 /// })
169 /// }
170 /// ```
171 #[cfg(feature = "plugin-consistent-hash")]
172 #[must_use]
173 pub fn consistent_hash(name: impl Into<String>) -> Self {
174 Self::new(name, ExchangeKind::Custom("x-consistent-hash".to_owned()))
175 }
176
177 /// Whether the exchange survives a broker restart. Defaults to `true`.
178 #[must_use]
179 pub fn durable(mut self, durable: bool) -> Self {
180 self.durable = durable;
181 self
182 }
183
184 /// Whether the exchange is deleted when its last binding is removed. Defaults to `false`.
185 #[must_use]
186 pub fn auto_delete(mut self, auto_delete: bool) -> Self {
187 self.auto_delete = auto_delete;
188 self
189 }
190
191 /// The exchange name.
192 #[must_use]
193 pub fn name(&self) -> &str {
194 &self.name
195 }
196
197 pub(crate) fn kind(&self) -> &ExchangeKind {
198 &self.kind
199 }
200
201 pub(crate) fn is_durable(&self) -> bool {
202 self.durable
203 }
204
205 pub(crate) fn is_auto_delete(&self) -> bool {
206 self.auto_delete
207 }
208}