ruststream_pulsar/
message.rs

1//! [`PulsarMessage`] and the mapping between `RustStream` headers and message properties.
2//!
3//! Message properties carry headers directly - no envelope format is invented - and the
4//! partition key rides the message's own `partition_key` in both directions.
5
6use bytes::Bytes;
7use pulsar::proto::MessageIdData;
8use ruststream::{AckError, Headers, IncomingMessage, OutgoingMessage, Partitioned, Positioned};
9use tokio::sync::{mpsc, oneshot};
10
11use crate::error::PulsarError;
12
13/// Header carrying the partition key, mapped onto the message's `partition_key` (which
14/// `KeyShared` subscriptions order by).
15///
16/// Mirrors the in-memory broker's convention, so services can switch brokers without changing
17/// their headers.
18pub const PARTITION_KEY_HEADER: &str = "partition-key";
19
20/// How a delivered message asks its driver task to settle it.
21#[derive(Debug)]
22pub(crate) enum SettleKind {
23    /// Acknowledge the delivery.
24    Ack,
25    /// Ask the broker to redeliver it (negative acknowledgement).
26    Nack,
27}
28
29/// A position in a topic's retained log, accepted by
30/// [`Seeker::seek`](ruststream::Seeker::seek).
31///
32/// Captured positions ([`Positioned::position`]) carry the pinned semantics the framework
33/// defines: seeking to one redelivers exactly that message. The timestamp form keeps the
34/// broker's own publish-time semantics instead.
35///
36/// The constructors exist because the `start_at(..)` clause of `#[subscriber]` recovers the
37/// position type from the constructor path; a bare variant path does not name its type in the
38/// tokens the macro sees.
39///
40/// # Examples
41///
42/// ```
43/// use ruststream_pulsar::PulsarPosition;
44///
45/// let from_the_top = PulsarPosition::earliest();
46/// let from_now_on = PulsarPosition::latest();
47/// let from_a_point_in_time = PulsarPosition::timestamp(1_700_000_000_000);
48/// # let _ = (from_the_top, from_now_on, from_a_point_in_time);
49/// ```
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub enum PulsarPosition {
52    /// The beginning of the log: every message the topics still retain is redelivered.
53    ///
54    /// This is a seek, not the server-side initial position, which Pulsar applies only when a
55    /// subscription is first created. A subscription is durable broker-side state, so a
56    /// `start_at(PulsarPosition::earliest())` clause rewinds an existing subscription's cursor
57    /// on every startup, replaying the retained backlog each time the service starts.
58    Earliest,
59    /// The tip of the log: only messages published after the seek are delivered.
60    Latest,
61    /// The position of a delivered message.
62    MessageId(MessageIdData),
63    /// Publish time, in milliseconds since the Unix epoch.
64    Timestamp(u64),
65}
66
67impl PulsarPosition {
68    /// The beginning of the log; see [`PulsarPosition::Earliest`] for how it interacts with a
69    /// durable subscription's cursor.
70    #[must_use]
71    pub fn earliest() -> Self {
72        Self::Earliest
73    }
74
75    /// The tip of the log; see [`PulsarPosition::Latest`].
76    #[must_use]
77    pub fn latest() -> Self {
78        Self::Latest
79    }
80
81    /// A publish time, in milliseconds since the Unix epoch; see
82    /// [`PulsarPosition::Timestamp`].
83    #[must_use]
84    pub fn timestamp(millis: u64) -> Self {
85        Self::Timestamp(millis)
86    }
87}
88
89/// A repositioning request shipped from a seeker handle to the subscription's driver task.
90#[derive(Debug)]
91pub(crate) struct SeekCmd {
92    pub(crate) position: PulsarPosition,
93    pub(crate) done: oneshot::Sender<Result<(), PulsarError>>,
94}
95
96/// A settlement request shipped from a message handle to the subscription's driver task
97/// (the client's ack API needs `&mut Consumer`, which the driver owns).
98#[derive(Debug)]
99pub(crate) struct SettleCmd {
100    pub(crate) topic: String,
101    pub(crate) id: MessageIdData,
102    pub(crate) kind: SettleKind,
103    pub(crate) done: oneshot::Sender<Result<(), AckError>>,
104}
105
106/// Everything the driver task can be asked to do while its stream runs.
107#[derive(Debug)]
108pub(crate) enum DriverCmd {
109    Settle(SettleCmd),
110    Seek(SeekCmd),
111}
112
113pub(crate) type SettleSender = mpsc::UnboundedSender<DriverCmd>;
114
115/// A message delivered by a [`PulsarSubscriber`](crate::PulsarSubscriber).
116///
117/// `ack` acknowledges; `nack(requeue = true)` asks the broker to redeliver, which is what
118/// drives the delivery count towards the subscription's dead-letter policy.
119/// `nack(requeue = false)` acknowledges: Pulsar has no terminal reject verb - poison routing
120/// belongs to the dead-letter policy, reached by repeated redelivery. The client queues
121/// acknowledgements asynchronously, so `Ok` means "queued", not "broker confirmed".
122pub struct PulsarMessage {
123    payload: Bytes,
124    headers: Headers,
125    topic: String,
126    id: MessageIdData,
127    settle: SettleSender,
128}
129
130impl std::fmt::Debug for PulsarMessage {
131    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132        f.debug_struct("PulsarMessage")
133            .field("topic", &self.topic)
134            .field("payload_len", &self.payload.len())
135            .finish_non_exhaustive()
136    }
137}
138
139impl PulsarMessage {
140    pub(crate) fn new(message: &pulsar::consumer::Message<Vec<u8>>, settle: SettleSender) -> Self {
141        let metadata = message.metadata();
142        let mut headers = Headers::with_capacity(metadata.properties.len() + 1);
143        for kv in &metadata.properties {
144            headers.insert(kv.key.clone(), kv.value.clone());
145        }
146        if let Some(key) = &metadata.partition_key {
147            headers.insert(PARTITION_KEY_HEADER, key.clone());
148        }
149        Self {
150            payload: Bytes::copy_from_slice(&message.payload.data),
151            headers,
152            topic: message.topic.clone(),
153            id: message.message_id().clone(),
154            settle,
155        }
156    }
157
158    /// The fully resolved topic this message arrived on (with its partition suffix when the
159    /// topic is partitioned).
160    #[must_use]
161    pub fn topic(&self) -> &str {
162        &self.topic
163    }
164
165    async fn send_settle(self, kind: SettleKind) -> Result<(), AckError> {
166        let (done, wait) = oneshot::channel();
167        self.settle
168            .send(DriverCmd::Settle(SettleCmd {
169                topic: self.topic,
170                id: self.id,
171                kind,
172                done,
173            }))
174            .map_err(|_| {
175                AckError::Broker(Box::from("the subscription's driver task has shut down"))
176            })?;
177        wait.await.map_err(|_| {
178            AckError::Broker(Box::from("the subscription's driver task has shut down"))
179        })?
180    }
181}
182
183impl Positioned for PulsarMessage {
184    type Position = PulsarPosition;
185
186    fn position(&self) -> PulsarPosition {
187        PulsarPosition::MessageId(self.id.clone())
188    }
189}
190
191impl Partitioned for PulsarMessage {
192    fn partition_key(&self) -> Option<&[u8]> {
193        self.headers.get(PARTITION_KEY_HEADER)
194    }
195}
196
197impl IncomingMessage for PulsarMessage {
198    fn payload(&self) -> &[u8] {
199        &self.payload
200    }
201
202    fn headers(&self) -> &Headers {
203        &self.headers
204    }
205
206    async fn ack(self) -> Result<(), AckError> {
207        self.send_settle(SettleKind::Ack).await
208    }
209
210    async fn nack(self, requeue: bool) -> Result<(), AckError> {
211        if requeue {
212            self.send_settle(SettleKind::Nack).await
213        } else {
214            // Acknowledging IS the drop: Pulsar has no terminal reject, and the dead-letter
215            // policy owns poison-message routing via repeated redelivery.
216            self.send_settle(SettleKind::Ack).await
217        }
218    }
219
220    fn partition_key(&self) -> Option<&[u8]> {
221        Partitioned::partition_key(self)
222    }
223}
224
225/// Builds the client message for an outgoing publish.
226pub(crate) fn to_pulsar_message(msg: &OutgoingMessage<'_>) -> pulsar::producer::Message {
227    let headers = msg.headers();
228    let mut properties = std::collections::HashMap::with_capacity(headers.len());
229    let mut partition_key = None;
230    for (name, value) in headers.iter() {
231        let text = String::from_utf8_lossy(value).into_owned();
232        if name == PARTITION_KEY_HEADER {
233            partition_key = Some(text);
234        } else {
235            properties.insert(name.to_owned(), text);
236        }
237    }
238    pulsar::producer::Message {
239        payload: msg.payload().to_vec(),
240        properties,
241        partition_key,
242        ..Default::default()
243    }
244}
245
246#[cfg(test)]
247mod tests {
248    use super::*;
249
250    #[test]
251    fn partition_key_header_becomes_the_partition_key() {
252        let mut headers = Headers::new();
253        headers.insert(PARTITION_KEY_HEADER, "user-42");
254        headers.insert("x-tenant", "acme");
255        let outgoing = OutgoingMessage::new("orders", b"{}".as_slice()).with_headers(headers);
256
257        let message = to_pulsar_message(&outgoing);
258        assert_eq!(message.partition_key.as_deref(), Some("user-42"));
259        assert_eq!(
260            message.properties.get("x-tenant").map(String::as_str),
261            Some("acme")
262        );
263        assert!(!message.properties.contains_key(PARTITION_KEY_HEADER));
264    }
265}
ruststream_pulsar/message.rs

ruststream_pulsar/
message.rs
