ruststream_lapin/
publish_policy.rs

1//! The declaration half of publishing: the policies and what they pair into.
2//!
3//! A policy holds nothing but publish options, so it is constructible anywhere - in a router
4//! definition, in configuration, before anything connects. Pairing it with a
5//! [`ConnectedLapinBroker`] produces the live publisher (see [`crate::publisher`]), which is the
6//! only value with a publish surface. The publishing mode is a policy transition:
7//! [`LapinPublish::confirms`] and [`LapinPublish::server_tx`] move to the transactional policies,
8//! keeping the options.
9
10use ruststream::{PairError, PublishPolicy};
11
12use crate::broker::ConnectedLapinBroker;
13use crate::publisher::{ConfirmsPublisher, LapinPublisher, ServerTxPublisher};
14
15use self::sealed::Sealed;
16
17mod sealed {
18    /// Seals [`LapinPublishPolicy`](super::LapinPublishPolicy): pairing an AMQP publisher opens
19    /// no channel of its own, and the synchronous
20    /// [`publisher`](crate::ConnectedLapinBroker::publisher) accessor depends on that.
21    pub trait Sealed {}
22
23    impl Sealed for super::LapinPublish {}
24    impl Sealed for super::ConfirmsPublish {}
25    impl Sealed for super::ServerTxPublish {}
26    impl Sealed for crate::requester::LapinRequest {}
27}
28
29/// The options every `RabbitMQ` publish policy carries: where to publish and how durably.
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub(crate) struct PublishOptions {
32    pub(crate) exchange: String,
33    pub(crate) persistent: bool,
34}
35
36impl Default for PublishOptions {
37    fn default() -> Self {
38        Self {
39            exchange: String::new(),
40            persistent: true,
41        }
42    }
43}
44
45/// A publish policy that pairs with a connected `RabbitMQ` broker without opening a channel.
46///
47/// All of this crate's policies hold nothing but publish options, so bringing one alive is a
48/// constructor call rather than broker work. That is what lets
49/// [`ConnectedLapinBroker::publisher`] be synchronous; [`PublishPolicy::pair`], the
50/// framework-side entry point, delegates here.
51pub trait LapinPublishPolicy: PublishPolicy<ConnectedLapinBroker> + Sealed {
52    /// Pairs the policy with the connected broker, producing the live publisher.
53    #[must_use]
54    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live;
55}
56
57/// The fire-and-forget publish policy: pure declaration, constructible anywhere.
58///
59/// [`OutgoingMessage::name`](ruststream::OutgoingMessage::name) is the routing key; the target
60/// exchange is a property of the policy (the default exchange unless
61/// [`exchange`](Self::exchange) says otherwise). On the default exchange the routing key
62/// addresses the queue with that name. Messages are published persistent (delivery mode 2)
63/// unless [`persistent(false)`](Self::persistent) opts out.
64///
65/// It pairs into [`LapinPublisher`], and it is the broker's
66/// [`DefaultPublish`](ruststream::DefaultPublish) policy, so a `publish("dest")` handler mounted
67/// without an explicit publisher replies through it. [`confirms`](Self::confirms) and
68/// [`server_tx`](Self::server_tx) move to the transactional policies, keeping the options.
69///
70/// # Examples
71///
72/// ```
73/// use ruststream_lapin::LapinPublish;
74///
75/// let events = LapinPublish::default().exchange("events");
76/// let shipments = LapinPublish::default().confirms();
77/// # let _ = (events, shipments);
78/// ```
79#[derive(Debug, Clone, Default, PartialEq, Eq)]
80#[must_use]
81pub struct LapinPublish(PublishOptions);
82
83impl LapinPublish {
84    /// Publishes to `exchange` instead of the default exchange.
85    pub fn exchange(mut self, exchange: impl Into<String>) -> Self {
86        self.0.exchange = exchange.into();
87        self
88    }
89
90    /// Whether messages are marked persistent (delivery mode 2). Defaults to `true`.
91    pub fn persistent(mut self, persistent: bool) -> Self {
92        self.0.persistent = persistent;
93        self
94    }
95
96    /// Moves to the policy that awaits broker confirms, with buffering transactions.
97    ///
98    /// The recommended transactional publisher: durable and much faster than AMQP server
99    /// transactions.
100    pub fn confirms(self) -> ConfirmsPublish {
101        ConfirmsPublish(self.0)
102    }
103
104    /// Moves to the policy backed by AMQP server transactions (`tx.select`).
105    ///
106    /// Server-side atomicity, at the cost of a synchronous commit round trip that is
107    /// significantly slower than [`confirms`](Self::confirms).
108    pub fn server_tx(self) -> ServerTxPublish {
109        ServerTxPublish(self.0)
110    }
111}
112
113impl PublishPolicy<ConnectedLapinBroker> for LapinPublish {
114    type Live = LapinPublisher;
115
116    async fn pair(self, connected: &ConnectedLapinBroker) -> Result<Self::Live, PairError> {
117        Ok(self.bind(connected))
118    }
119}
120
121impl LapinPublishPolicy for LapinPublish {
122    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live {
123        LapinPublisher::new(connected, self.0)
124    }
125}
126
127/// The confirm-transactional publish policy: same options as [`LapinPublish`], pairing into
128/// [`ConfirmsPublisher`].
129///
130/// Reached with [`LapinPublish::confirms`].
131///
132/// # Examples
133///
134/// ```
135/// use ruststream_lapin::LapinPublish;
136///
137/// let shipments = LapinPublish::default().exchange("shipments").confirms();
138/// # let _ = shipments;
139/// ```
140#[derive(Debug, Clone, Default, PartialEq, Eq)]
141#[must_use]
142pub struct ConfirmsPublish(PublishOptions);
143
144impl ConfirmsPublish {
145    /// Publishes to `exchange` instead of the default exchange.
146    pub fn exchange(mut self, exchange: impl Into<String>) -> Self {
147        self.0.exchange = exchange.into();
148        self
149    }
150
151    /// Whether messages are marked persistent (delivery mode 2). Defaults to `true`.
152    pub fn persistent(mut self, persistent: bool) -> Self {
153        self.0.persistent = persistent;
154        self
155    }
156}
157
158impl PublishPolicy<ConnectedLapinBroker> for ConfirmsPublish {
159    type Live = ConfirmsPublisher;
160
161    async fn pair(self, connected: &ConnectedLapinBroker) -> Result<Self::Live, PairError> {
162        Ok(self.bind(connected))
163    }
164}
165
166impl LapinPublishPolicy for ConfirmsPublish {
167    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live {
168        ConfirmsPublisher::new(connected, self.0)
169    }
170}
171
172/// The server-transactional publish policy: same options as [`LapinPublish`], pairing into
173/// [`ServerTxPublisher`].
174///
175/// Reached with [`LapinPublish::server_tx`].
176///
177/// # Examples
178///
179/// ```
180/// use ruststream_lapin::LapinPublish;
181///
182/// let ledger = LapinPublish::default().exchange("ledger").server_tx();
183/// # let _ = ledger;
184/// ```
185#[derive(Debug, Clone, Default, PartialEq, Eq)]
186#[must_use]
187pub struct ServerTxPublish(PublishOptions);
188
189impl ServerTxPublish {
190    /// Publishes to `exchange` instead of the default exchange.
191    pub fn exchange(mut self, exchange: impl Into<String>) -> Self {
192        self.0.exchange = exchange.into();
193        self
194    }
195
196    /// Whether messages are marked persistent (delivery mode 2). Defaults to `true`.
197    pub fn persistent(mut self, persistent: bool) -> Self {
198        self.0.persistent = persistent;
199        self
200    }
201}
202
203impl PublishPolicy<ConnectedLapinBroker> for ServerTxPublish {
204    type Live = ServerTxPublisher;
205
206    async fn pair(self, connected: &ConnectedLapinBroker) -> Result<Self::Live, PairError> {
207        Ok(self.bind(connected))
208    }
209}
210
211impl LapinPublishPolicy for ServerTxPublish {
212    fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live {
213        ServerTxPublisher::new(connected, self.0)
214    }
215}
ruststream_lapin/publish_policy.rs

ruststream_lapin/
publish_policy.rs
