ruststream_lapin/
publish_step.rs

1//! Per-message AMQP basic properties: the policy's defaults, and the steps that adjust one
2//! message.
3//!
4//! [`LapinPublishOptions`] is what one publish may differ from the next in - the `priority`, the
5//! per-message `expiration` (TTL), and the delivery mode. Every field is optional: what a call
6//! leaves alone keeps what the publish policy fixed at the mount site.
7//!
8//! ```text
9//! b.include(expedite).out(Alerts, Publish::default().priority(3));  // the default
10//! alerts.message(&alert).to("alerts").priority(9).publish().await?; // this one message
11//! ```
12//!
13//! [`LapinPublishSteps`] puts those steps on the publish builder, bounded on the options type, so
14//! they appear over this crate's publishers and over no others. Nothing wraps the publisher: the
15//! publish still leaves through the mount site's own entry, with the codec and the transforms that
16//! entry named, and the test harness still attributes it to its [`Out`](ruststream::runtime::Out)
17//! slot.
18//!
19//! The values reach the frame as the protocol fields they are, never as headers. Written into the
20//! header table under the protocol's own names (`priority`, `expiration`) they would travel as
21//! table entries `RabbitMQ` reads for neither purpose; that quiet failure is what the steps exist
22//! to prevent.
23
24use std::time::Duration;
25
26use ruststream::runtime::{PublishBuilder, PublishSink};
27
28/// Header carrying the AMQP `priority` property of a DELIVERY.
29///
30/// A delivery reports the property under this name, so a handler reads it from the message
31/// headers. Publishing under it sets no property: the priority of an outgoing message is
32/// [`LapinPublishSteps::priority`] or the policy's own
33/// [`LapinPublish::priority`](crate::LapinPublish::priority).
34///
35/// The property only orders deliveries on a queue declared with `x-max-priority`; elsewhere the
36/// broker carries it to the consumer and nothing more.
37pub const PRIORITY_HEADER: &str = "amqp-priority";
38
39/// Header carrying the AMQP per-message `expiration` property of a DELIVERY: whole milliseconds,
40/// as decimal digits, which is how AMQP spells a TTL.
41///
42/// A delivery reports the property under this name. Publishing under it sets no property: the TTL
43/// of an outgoing message is [`LapinPublishSteps::expiration`] or the policy's own
44/// [`LapinPublish::expiration`](crate::LapinPublish::expiration).
45///
46/// The broker drops the message once the TTL has passed without it being consumed, dead-lettering
47/// it when the queue says so.
48pub const EXPIRATION_HEADER: &str = "amqp-expiration";
49
50/// The AMQP basic properties one publish may differ from the next in.
51///
52/// Every field is optional because a publish carries only what its call site adjusted; the rest is
53/// what the publish policy fixed when it paired the publisher. The policy carries the defaults
54/// ([`LapinPublish::priority`](crate::LapinPublish::priority),
55/// [`expiration`](crate::LapinPublish::expiration),
56/// [`persistent`](crate::LapinPublish::persistent)), the steps of [`LapinPublishSteps`] adjust one
57/// message, and the publisher resolves the two before it builds the frame.
58///
59/// A test reads the value back from the slot view: `tb.out::<Alerts>().with_options(..)` takes
60/// this type, which is why it derives `Debug` and `PartialEq`.
61///
62/// # Examples
63///
64/// ```
65/// # #[cfg(feature = "testing")]
66/// # async fn demo() -> Result<(), Box<dyn std::error::Error>> {
67/// use ruststream::OutSlot;
68/// use ruststream::testing::TestApp;
69/// use ruststream_lapin::prelude::*;
70/// use serde::{Deserialize, Serialize};
71///
72/// #[derive(Debug, Serialize, Deserialize, PartialEq, Outgoing)]
73/// struct Alert {
74///     level: u8,
75/// }
76///
77/// #[derive(OutSlot)]
78/// #[publishes(Alert)]
79/// struct Alerts;
80///
81/// #[subscriber("alerts.raw")]
82/// async fn escalate(
83///     alert: &Alert,
84///     Out(alerts): Out<impl Publisher<Options = LapinPublishOptions>, Alerts>,
85/// ) -> HandlerOutcome {
86///     let sent = alerts.message(alert).to("alerts").priority(alert.level).publish().await;
87///     if sent.is_err() {
88///         return HandlerOutcome::retry();
89///     }
90///     HandlerOutcome::ack()
91/// }
92///
93/// let app = RustStream::new(AppInfo::new("alerts", "0.1.0")).with_broker(
94///     LapinBroker::new("amqp://localhost:5672"),
95///     |b| {
96///         b.include(escalate).out(Alerts, Publish::default()).build();
97///     },
98/// );
99/// let tb = TestApp::start(app).await?;
100///
101/// tb.broker::<LapinBroker>()
102///     .message(&Alert { level: 9 })
103///     .to("alerts.raw")
104///     .publish()
105///     .await?;
106///
107/// tb.out::<Alerts>()
108///     .assert_called_once()
109///     .with_options(&LapinPublishOptions {
110///         priority: Some(9),
111///         ..LapinPublishOptions::default()
112///     });
113/// # Ok(())
114/// # }
115/// ```
116#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
117pub struct LapinPublishOptions {
118    /// The AMQP `priority` property (0..=255), or `None` to keep the policy's.
119    pub priority: Option<u8>,
120    /// The AMQP per-message `expiration` (TTL), or `None` to keep the policy's. AMQP counts it in
121    /// whole milliseconds; a shorter non-zero value becomes one millisecond.
122    pub expiration: Option<Duration>,
123    /// Whether the message is marked persistent (delivery mode 2), or `None` to keep the policy's.
124    pub persistent: Option<bool>,
125}
126
127impl LapinPublishOptions {
128    /// What a publish with neither a call site nor a policy carries: persistent, nothing else.
129    ///
130    /// The native delayed redelivery publishes a copy on the delivery's own channel, outside any
131    /// policy, and a copy that outlives a broker restart is what a delay is for.
132    pub(crate) const PERSISTENT: Self = Self {
133        priority: None,
134        expiration: None,
135        persistent: Some(true),
136    };
137
138    /// This value's fields where it has them, `defaults` where it does not.
139    ///
140    /// What a publisher does first: the call site's options over the policy's.
141    #[must_use]
142    pub(crate) fn over(&self, defaults: &Self) -> Self {
143        Self {
144            priority: self.priority.or(defaults.priority),
145            expiration: self.expiration.or(defaults.expiration),
146            persistent: self.persistent.or(defaults.persistent),
147        }
148    }
149}
150
151/// The AMQP basic properties, as steps on the publish builder.
152///
153/// A step adjusts one field for one message and hands the builder back, so a publish reads as one
154/// chain. It reaches every publish surface this crate ships - a publisher, an
155/// [`Out`](ruststream::runtime::Out) slot inside a handler, a transaction scope - because it is
156/// bounded on the options type rather than on a publisher type.
157///
158/// A handler body that takes a step is the one place a body names this crate: bound the slot
159/// `Out<impl Publisher<Options = LapinPublishOptions>, Marker>` and import this crate's prelude.
160///
161/// # Examples
162///
163/// ```
164/// use std::time::Duration;
165///
166/// use ruststream_lapin::prelude::*;
167/// use serde::{Deserialize, Serialize};
168///
169/// #[derive(Deserialize)]
170/// struct Order {
171///     id: u64,
172///     expedited: bool,
173/// }
174///
175/// #[derive(Serialize, Outgoing)]
176/// #[outgoing(name = "shipments")]
177/// struct Shipment {
178///     order_id: u64,
179/// }
180///
181/// #[subscriber("orders")]
182/// async fn ship(
183///     order: &Order,
184///     Out(shipments): Out<impl Publisher<Options = LapinPublishOptions>>,
185/// ) -> HandlerOutcome {
186///     let shipment = Shipment { order_id: order.id };
187///     let sent = if order.expedited {
188///         shipments
189///             .message(&shipment)
190///             .priority(9)
191///             .expiration(Duration::from_secs(30))
192///             .publish()
193///             .await
194///     } else {
195///         shipments.message(&shipment).publish().await
196///     };
197///     if sent.is_err() {
198///         return HandlerOutcome::retry();
199///     }
200///     HandlerOutcome::ack()
201/// }
202///
203/// // Every shipment takes priority 3 from the policy; an expedited one overrides it.
204/// fn app() -> RustStream {
205///     let broker = LapinBroker::new("amqp://localhost:5672");
206///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
207///         b.include(ship)
208///             .out(DefaultSlot, Publish::default().priority(3))
209///             .build();
210///     })
211/// }
212/// ```
213pub trait LapinPublishSteps {
214    /// Sends this one message with the AMQP `priority` property set, whatever the policy's
215    /// default is.
216    ///
217    /// The property only orders deliveries on a queue declared with `x-max-priority`; elsewhere
218    /// the broker carries it to the consumer and nothing more.
219    ///
220    /// # Examples
221    ///
222    /// ```
223    /// use ruststream_lapin::prelude::*;
224    /// use serde::{Deserialize, Serialize};
225    ///
226    /// #[derive(Deserialize)]
227    /// struct Ticket {
228    ///     id: u64,
229    ///     vip: bool,
230    /// }
231    ///
232    /// #[derive(Serialize, Outgoing)]
233    /// #[outgoing(name = "support.queue")]
234    /// struct Escalation {
235    ///     ticket: u64,
236    /// }
237    ///
238    /// #[subscriber("tickets")]
239    /// async fn escalate(
240    ///     ticket: &Ticket,
241    ///     Out(support): Out<impl Publisher<Options = LapinPublishOptions>>,
242    /// ) -> HandlerOutcome {
243    ///     let priority = if ticket.vip { 9 } else { 1 };
244    ///     let escalation = Escalation { ticket: ticket.id };
245    ///     let sent = support.message(&escalation).priority(priority).publish().await;
246    ///     if sent.is_err() {
247    ///         return HandlerOutcome::retry();
248    ///     }
249    ///     HandlerOutcome::ack()
250    /// }
251    /// ```
252    #[must_use]
253    fn priority(self, priority: u8) -> Self;
254
255    /// Sends this one message with the AMQP per-message `expiration` (TTL) set: the broker drops
256    /// it once `ttl` has passed without it being consumed, dead-lettering it when the queue says
257    /// so.
258    ///
259    /// AMQP counts the TTL in whole milliseconds; a shorter non-zero `ttl` becomes one
260    /// millisecond.
261    ///
262    /// # Examples
263    ///
264    /// ```
265    /// use std::time::Duration;
266    ///
267    /// use ruststream_lapin::prelude::*;
268    /// use serde::{Deserialize, Serialize};
269    ///
270    /// #[derive(Deserialize)]
271    /// struct QuoteRequest {
272    ///     sku: String,
273    /// }
274    ///
275    /// #[derive(Serialize, Outgoing)]
276    /// #[outgoing(name = "quotes")]
277    /// struct Quote {
278    ///     sku: String,
279    ///     cents: u64,
280    /// }
281    ///
282    /// // A price is only good for half a minute; nobody should act on a stale one.
283    /// #[subscriber("quote.requests")]
284    /// async fn price(
285    ///     request: &QuoteRequest,
286    ///     Out(quotes): Out<impl Publisher<Options = LapinPublishOptions>>,
287    /// ) -> HandlerOutcome {
288    ///     let quote = Quote {
289    ///         sku: request.sku.clone(),
290    ///         cents: 1_999,
291    ///     };
292    ///     let sent = quotes
293    ///         .message(&quote)
294    ///         .expiration(Duration::from_secs(30))
295    ///         .publish()
296    ///         .await;
297    ///     if sent.is_err() {
298    ///         return HandlerOutcome::retry();
299    ///     }
300    ///     HandlerOutcome::ack()
301    /// }
302    /// ```
303    #[must_use]
304    fn expiration(self, ttl: Duration) -> Self;
305
306    /// Sends this one message persistent (delivery mode 2) or transient (1), whatever the
307    /// policy's default is.
308    ///
309    /// A persistent message on a durable queue survives a broker restart. The publish policies
310    /// default to persistent; [`LapinRequest`](crate::LapinRequest) defaults to transient,
311    /// because a request nobody is waiting for after the timeout gains nothing from surviving a
312    /// restart.
313    ///
314    /// # Examples
315    ///
316    /// ```
317    /// use ruststream_lapin::prelude::*;
318    /// use serde::{Deserialize, Serialize};
319    ///
320    /// #[derive(Deserialize)]
321    /// struct Order {
322    ///     id: u64,
323    /// }
324    ///
325    /// #[derive(Serialize, Outgoing)]
326    /// #[outgoing(name = "metrics")]
327    /// struct Sample {
328    ///     order_id: u64,
329    /// }
330    ///
331    /// // The order is the record; a lost sample after a broker restart costs nothing.
332    /// #[subscriber("orders")]
333    /// async fn measure(
334    ///     order: &Order,
335    ///     Out(metrics): Out<impl Publisher<Options = LapinPublishOptions>>,
336    /// ) -> HandlerOutcome {
337    ///     let sample = Sample { order_id: order.id };
338    ///     let sent = metrics.message(&sample).persistent(false).publish().await;
339    ///     if sent.is_err() {
340    ///         return HandlerOutcome::retry();
341    ///     }
342    ///     HandlerOutcome::ack()
343    /// }
344    /// ```
345    #[must_use]
346    fn persistent(self, persistent: bool) -> Self;
347}
348
349impl<Sink, Body, Enc, Hdrs, Dest> LapinPublishSteps for PublishBuilder<Sink, Body, Enc, Hdrs, Dest>
350where
351    Sink: PublishSink<Options = LapinPublishOptions>,
352{
353    fn priority(mut self, priority: u8) -> Self {
354        self.options_mut()
355            .get_or_insert_with(LapinPublishOptions::default)
356            .priority = Some(priority);
357        self
358    }
359
360    fn expiration(mut self, ttl: Duration) -> Self {
361        self.options_mut()
362            .get_or_insert_with(LapinPublishOptions::default)
363            .expiration = Some(ttl);
364        self
365    }
366
367    fn persistent(mut self, persistent: bool) -> Self {
368        self.options_mut()
369            .get_or_insert_with(LapinPublishOptions::default)
370            .persistent = Some(persistent);
371        self
372    }
373}
374
375#[cfg(test)]
376mod tests {
377    use super::{Duration, LapinPublishOptions};
378
379    #[test]
380    fn a_call_site_field_wins_over_the_policy_and_the_rest_is_kept() {
381        let defaults = LapinPublishOptions {
382            priority: Some(3),
383            expiration: Some(Duration::from_mins(1)),
384            persistent: Some(true),
385        };
386        let call = LapinPublishOptions {
387            priority: Some(9),
388            ..LapinPublishOptions::default()
389        };
390
391        let resolved = call.over(&defaults);
392        assert_eq!(resolved.priority, Some(9));
393        assert_eq!(resolved.expiration, Some(Duration::from_mins(1)));
394        assert_eq!(resolved.persistent, Some(true));
395    }
396
397    #[test]
398    fn a_publish_that_adjusted_nothing_is_the_policy_alone() {
399        let defaults = LapinPublishOptions {
400            priority: Some(3),
401            ..LapinPublishOptions::default()
402        };
403
404        assert_eq!(LapinPublishOptions::default().over(&defaults), defaults);
405    }
406}
ruststream_lapin/publish_step.rs

ruststream_lapin/
publish_step.rs
