ruststream_fred/pipeline/
message.rs

1//! The delivery of a pipelined subscription: the form's own delivery, with a slot in the window.
2
3use std::fmt::{Debug, Formatter};
4use std::sync::Arc;
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::time::Duration;
7
8use ruststream::{AckError, HeaderMap, IncomingMessage, Partitioned};
9
10use super::window::{Form, Round, Window};
11
12/// A delivery of a `.pipeline()` subscription.
13///
14/// It reads as the form's own delivery does, and settles into the window instead of on the
15/// connection: `ack` commits what the handler queued together with the delivery's settle, and
16/// every other outcome drops it. A delivery dropped unsettled drops its segment and stays with
17/// the broker.
18///
19/// # Examples
20///
21/// ```
22/// # mod demo {
23/// use ruststream_fred::stream::prelude::*;
24/// # #[derive(serde::Deserialize)]
25/// # struct Order { id: u64 }
26///
27/// #[subscriber(RedisStream::new("orders").group("workers"))]
28/// async fn work(order: &Order) -> HandlerOutcome {
29///     tracing::info!(order.id, "order");
30///     HandlerOutcome::ack()
31/// }
32///
33/// fn app() -> RustStream {
34///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
35///         RedisBroker::standalone("redis://localhost:6379"),
36///         |b| {
37///             // The subscription yields this type; the handler reads it as any delivery.
38///             b.include(work.pipeline());
39///         },
40///     )
41/// }
42/// # }
43/// ```
44pub struct RoundMessage<F: Form> {
45    /// `None` once settled.
46    inner: Option<F::Message>,
47    window: Arc<Window<F>>,
48    round: Round,
49    /// Set once the round is recorded against the task handling the delivery.
50    entered: AtomicBool,
51}
52
53impl<F: Form> Debug for RoundMessage<F>
54where
55    F::Message: Debug,
56{
57    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
58        f.debug_struct("RoundMessage")
59            .field("inner", &self.inner)
60            .finish_non_exhaustive()
61    }
62}
63
64impl<F: Form> RoundMessage<F> {
65    pub(crate) const fn new(inner: F::Message, window: Arc<Window<F>>, round: Round) -> Self {
66        Self {
67            inner: Some(inner),
68            window,
69            round,
70            entered: AtomicBool::new(false),
71        }
72    }
73
74    /// The form's own delivery.
75    ///
76    /// # Panics
77    ///
78    /// Panics once the delivery has settled, which consumes it; nothing reads it after that.
79    pub(crate) fn inner(&self) -> &F::Message {
80        self.inner
81            .as_ref()
82            .expect("a pipelined delivery is read only before it settles")
83    }
84
85    pub(crate) fn round(&self) -> &Round {
86        &self.round
87    }
88
89    async fn settle(
90        mut self,
91        commit: bool,
92        settle: impl FnOnce(&F, F::Message, &mut Vec<F::Op>) -> Result<(), AckError>,
93    ) -> Result<(), AckError> {
94        #[cfg_attr(not(feature = "testing"), allow(unused_mut))]
95        let mut inner = self
96            .inner
97            .take()
98            .expect("a pipelined delivery settles once");
99        let window = Arc::clone(&self.window);
100        #[cfg(feature = "testing")]
101        window.hold(F::take_flight(&mut inner));
102        window
103            .settle(&self.round, commit, |form, ops| settle(form, inner, ops))
104            .await
105    }
106}
107
108impl<F: Form> Drop for RoundMessage<F> {
109    fn drop(&mut self) {
110        if self.inner.is_some() {
111            self.window.abandon(&self.round);
112        }
113    }
114}
115
116impl<F: Form> Partitioned for RoundMessage<F>
117where
118    F::Message: Partitioned,
119{
120    fn partition_key(&self) -> Option<&[u8]> {
121        Partitioned::partition_key(self.inner())
122    }
123}
124
125impl<F: Form> RoundMessage<F> {
126    /// Records the delivery's round against the task reading it, once: the runtime decodes a
127    /// delivery in the task that handles it, before the handler runs, and a reply marked to join
128    /// the round finds it there.
129    fn enter(&self) {
130        // The load keeps every read after the first to a plain one.
131        if !self.entered.load(Ordering::Relaxed) && !self.entered.swap(true, Ordering::Relaxed) {
132            self.round.enter();
133        }
134    }
135}
136
137impl<F: Form> IncomingMessage for RoundMessage<F> {
138    /// Also enters the round: the payload is what the runtime reads to decode the input, and it
139    /// reads the headers only when something asks for them.
140    fn payload(&self) -> &[u8] {
141        self.enter();
142        self.inner().payload()
143    }
144
145    /// Also enters the round, for a handler that reads the headers alone.
146    fn headers(&self) -> &HeaderMap {
147        self.enter();
148        self.inner().headers()
149    }
150
151    fn partition_key(&self) -> Option<&[u8]> {
152        self.inner().partition_key()
153    }
154
155    fn redelivery_count(&self) -> Option<u64> {
156        self.inner().redelivery_count()
157    }
158
159    fn supports_nack_after(&self) -> bool {
160        self.inner().supports_nack_after()
161    }
162
163    async fn ack(self) -> Result<(), AckError> {
164        self.settle(true, F::ack).await
165    }
166
167    async fn nack(self, requeue: bool) -> Result<(), AckError> {
168        self.settle(false, move |form, msg, ops| form.nack(msg, requeue, ops))
169            .await
170    }
171
172    async fn nack_after(self, delay: Duration) -> Result<(), AckError> {
173        self.settle(false, move |form, msg, ops| {
174            form.nack_after(msg, delay, ops)
175        })
176        .await
177    }
178}
ruststream_fred/pipeline/message.rs

ruststream_fred/pipeline/
message.rs
