ruststream_nats/
context.rs

1//! Optional typed per-delivery context exposing native `JetStream` metadata.
2//!
3//! A handler reads native `JetStream` delivery metadata - the stream and consumer names, the stream
4//! and consumer sequence numbers, the server-side redelivery count, and the pending count - by
5//! declaring [`JetStreamContext`] as its per-delivery context and reading fields with the
6//! compile-time [`keys`]. The runtime builds the context once per delivery via
7//! [`BuildContext`](ruststream::BuildContext); resolving a key is a direct field read, with no
8//! hashing, boxing, or downcasting.
9//!
10//! This is purely additive: the default context is `()` (no fields), so existing handlers are
11//! unaffected. Opting in means changing the handler's context type to [`JetStreamContext`].
12//!
13//! These fields are genuinely native: they come from the `JetStream` `$JS.ACK` reply subject, not
14//! from the payload or the message [`Headers`](ruststream::Headers), so they are not reachable any
15//! other way. Core (non-JetStream) NATS deliveries carry no such metadata - their only native datum
16//! is the reply inbox, already surfaced as the `reply-to` header - so Core handlers should keep the
17//! default `()` context. A handler bound to [`JetStreamContext`] still works on a Core subscription;
18//! every key just reads `None` there.
19//!
20//! # Examples
21//!
22//! ```
23//! use ruststream::IncomingMessage;
24//! use ruststream::runtime::{Context, HandlerResult};
25//! use ruststream_nats::context::{JetStreamContext, keys};
26//!
27//! async fn handle<M: IncomingMessage>(
28//!     _msg: &M,
29//!     ctx: &mut Context<'_, JetStreamContext>,
30//! ) -> HandlerResult {
31//!     // `None` on a core delivery; the stream sequence on a JetStream one.
32//!     if let Some(seq) = ctx.context(keys::STREAM_SEQUENCE) {
33//!         println!("stream sequence {seq}");
34//!     }
35//!     // The server-side delivery count distinguishes a first delivery from a redelivery.
36//!     if ctx.context(keys::DELIVERED).is_some_and(|n| n > 1) {
37//!         println!("redelivery");
38//!     }
39//!     HandlerResult::Ack
40//! }
41//! ```
42
43use ruststream::BuildContext;
44
45use crate::message::NatsMessage;
46
47/// Native `JetStream` per-delivery metadata, built once per delivery from the broker message.
48///
49/// Read its fields by the compile-time [`keys`] (for example
50/// [`keys::STREAM_SEQUENCE`]). On a core (non-JetStream) delivery there is no such metadata, so the
51/// context is empty and every key reads `None`; the same handler therefore works on both kinds of
52/// subscription.
53///
54/// The context owns its fields (it copies them out of the delivery), so it does not borrow the
55/// message: numbers are a stack copy and the stream/consumer names are cloned once per delivery.
56///
57/// # Examples
58///
59/// ```
60/// use ruststream::runtime::Context;
61/// use ruststream_nats::context::{JetStreamContext, keys};
62///
63/// // The context a handler reads through; the runtime supplies it per delivery.
64/// fn read(ctx: &Context<'_, JetStreamContext>) -> Option<u64> {
65///     ctx.context(keys::STREAM_SEQUENCE)
66/// }
67/// ```
68#[derive(Debug, Clone, Default, PartialEq, Eq)]
69pub struct JetStreamContext {
70    /// `None` for a core delivery, or when the `JetStream` reply subject could not be parsed.
71    info: Option<JetStreamInfo>,
72}
73
74/// The owned snapshot of one `JetStream` delivery's native metadata.
75#[derive(Debug, Clone, PartialEq, Eq)]
76struct JetStreamInfo {
77    stream: String,
78    consumer: String,
79    stream_sequence: u64,
80    consumer_sequence: u64,
81    delivered: i64,
82    pending: u64,
83}
84
85impl BuildContext<NatsMessage> for JetStreamContext {
86    fn build(msg: &NatsMessage) -> Self {
87        let info = match msg {
88            NatsMessage::JetStream(m) => m.info().map(|i| JetStreamInfo {
89                stream: i.stream.to_owned(),
90                consumer: i.consumer.to_owned(),
91                stream_sequence: i.stream_sequence,
92                consumer_sequence: i.consumer_sequence,
93                delivered: i.delivered,
94                pending: i.pending,
95            }),
96            NatsMessage::Core(_) => None,
97        };
98        Self { info }
99    }
100}
101
102// Compile-time guarantee that the context is buildable from the subscriber's actual delivery type,
103// so a handler declaring `Context<'_, JetStreamContext>` satisfies the runtime's
104// `Cx: BuildContext<S::Message>` mount bound. The JetStream `build` arm itself is only exercised
105// against a real server (`async_nats::jetstream::Message` has no in-process constructor), so this
106// type-level check is what validates the mount path here.
107const _: fn() = || {
108    fn assert_build_context<C: BuildContext<M>, M>() {}
109    assert_build_context::<
110        JetStreamContext,
111        <crate::NatsSubscriber as ruststream::Subscriber>::Message,
112    >();
113};
114
115/// Compile-time [`Field`](ruststream::Field) keys, one per native `JetStream` metadatum.
116///
117/// Each key is a zero-sized selector exported both as a type (for naming in bounds) and as a
118/// `const` value (for use at the call site, `ctx.context(keys::STREAM_SEQUENCE)`). Every key reads
119/// `None` on a core delivery or when the `JetStream` reply subject could not be parsed.
120pub mod keys {
121    use ruststream::{ContextField, Field};
122
123    use super::JetStreamContext;
124
125    /// Selector for the stream sequence number; see [`STREAM_SEQUENCE`].
126    #[derive(Debug, Clone, Copy, Default)]
127    pub struct StreamSequence;
128
129    /// The monotonically increasing sequence the message holds within its stream.
130    pub const STREAM_SEQUENCE: StreamSequence = StreamSequence;
131
132    impl Field<JetStreamContext> for StreamSequence {
133        type Value<'a> = Option<u64>;
134        fn get(self, cx: &JetStreamContext) -> Option<u64> {
135            cx.info.as_ref().map(|i| i.stream_sequence)
136        }
137    }
138
139    impl ContextField for StreamSequence {
140        type Context = JetStreamContext;
141        type Value = Option<u64>;
142        fn read(self, cx: &JetStreamContext) -> Option<u64> {
143            cx.info.as_ref().map(|i| i.stream_sequence)
144        }
145    }
146
147    /// Selector for the consumer sequence number; see [`CONSUMER_SEQUENCE`].
148    #[derive(Debug, Clone, Copy, Default)]
149    pub struct ConsumerSequence;
150
151    /// The sequence the message holds within this consumer's delivery stream.
152    pub const CONSUMER_SEQUENCE: ConsumerSequence = ConsumerSequence;
153
154    impl Field<JetStreamContext> for ConsumerSequence {
155        type Value<'a> = Option<u64>;
156        fn get(self, cx: &JetStreamContext) -> Option<u64> {
157            cx.info.as_ref().map(|i| i.consumer_sequence)
158        }
159    }
160
161    impl ContextField for ConsumerSequence {
162        type Context = JetStreamContext;
163        type Value = Option<u64>;
164        fn read(self, cx: &JetStreamContext) -> Option<u64> {
165            cx.info.as_ref().map(|i| i.consumer_sequence)
166        }
167    }
168
169    /// Selector for the server-side delivery count; see [`DELIVERED`].
170    #[derive(Debug, Clone, Copy, Default)]
171    pub struct Delivered;
172
173    /// The number of times the server has delivered this message.
174    ///
175    /// `1` on first delivery, higher on a redelivery (after a `nack` or an `ack_wait` timeout).
176    /// This is the native redelivery count, not reconstructable from the payload or headers.
177    pub const DELIVERED: Delivered = Delivered;
178
179    impl Field<JetStreamContext> for Delivered {
180        type Value<'a> = Option<i64>;
181        fn get(self, cx: &JetStreamContext) -> Option<i64> {
182            cx.info.as_ref().map(|i| i.delivered)
183        }
184    }
185
186    impl ContextField for Delivered {
187        type Context = JetStreamContext;
188        type Value = Option<i64>;
189        fn read(self, cx: &JetStreamContext) -> Option<i64> {
190            cx.info.as_ref().map(|i| i.delivered)
191        }
192    }
193
194    /// Selector for the pending count; see [`PENDING`].
195    #[derive(Debug, Clone, Copy, Default)]
196    pub struct Pending;
197
198    /// The number of messages the server still has pending for this consumer behind this one.
199    pub const PENDING: Pending = Pending;
200
201    impl Field<JetStreamContext> for Pending {
202        type Value<'a> = Option<u64>;
203        fn get(self, cx: &JetStreamContext) -> Option<u64> {
204            cx.info.as_ref().map(|i| i.pending)
205        }
206    }
207
208    impl ContextField for Pending {
209        type Context = JetStreamContext;
210        type Value = Option<u64>;
211        fn read(self, cx: &JetStreamContext) -> Option<u64> {
212            cx.info.as_ref().map(|i| i.pending)
213        }
214    }
215
216    /// Selector for the stream name; see [`STREAM`].
217    #[derive(Debug, Clone, Copy, Default)]
218    pub struct Stream;
219
220    /// The name of the stream this message was delivered from.
221    pub const STREAM: Stream = Stream;
222
223    impl Field<JetStreamContext> for Stream {
224        type Value<'a> = Option<&'a str>;
225        fn get(self, cx: &JetStreamContext) -> Option<&str> {
226            cx.info.as_ref().map(|i| i.stream.as_str())
227        }
228    }
229
230    impl ContextField for Stream {
231        type Context = JetStreamContext;
232        type Value = Option<String>;
233        fn read(self, cx: &JetStreamContext) -> Option<String> {
234            cx.info.as_ref().map(|i| i.stream.clone())
235        }
236    }
237
238    /// Selector for the consumer name; see [`CONSUMER`].
239    #[derive(Debug, Clone, Copy, Default)]
240    pub struct Consumer;
241
242    /// The name of the durable (or ephemeral) consumer this message was delivered through.
243    pub const CONSUMER: Consumer = Consumer;
244
245    impl Field<JetStreamContext> for Consumer {
246        type Value<'a> = Option<&'a str>;
247        fn get(self, cx: &JetStreamContext) -> Option<&str> {
248            cx.info.as_ref().map(|i| i.consumer.as_str())
249        }
250    }
251
252    impl ContextField for Consumer {
253        type Context = JetStreamContext;
254        type Value = Option<String>;
255        fn read(self, cx: &JetStreamContext) -> Option<String> {
256            cx.info.as_ref().map(|i| i.consumer.clone())
257        }
258    }
259}
260
261#[cfg(test)]
262mod tests {
263    use ruststream::{BuildContext, Field};
264
265    use super::keys::{CONSUMER, CONSUMER_SEQUENCE, DELIVERED, PENDING, STREAM, STREAM_SEQUENCE};
266    use super::{JetStreamContext, JetStreamInfo};
267    use crate::message::{CoreMessage, NatsMessage};
268
269    fn populated() -> JetStreamContext {
270        JetStreamContext {
271            info: Some(JetStreamInfo {
272                stream: "ORDERS".to_owned(),
273                consumer: "orders-worker".to_owned(),
274                stream_sequence: 42,
275                consumer_sequence: 7,
276                delivered: 3,
277                pending: 5,
278            }),
279        }
280    }
281
282    #[test]
283    fn keys_read_populated_jetstream_fields() {
284        let cx = populated();
285        assert_eq!(STREAM_SEQUENCE.get(&cx), Some(42));
286        assert_eq!(CONSUMER_SEQUENCE.get(&cx), Some(7));
287        assert_eq!(DELIVERED.get(&cx), Some(3));
288        assert_eq!(PENDING.get(&cx), Some(5));
289        assert_eq!(STREAM.get(&cx), Some("ORDERS"));
290        assert_eq!(CONSUMER.get(&cx), Some("orders-worker"));
291    }
292
293    #[test]
294    fn context_field_keys_yield_owned_values() {
295        use ruststream::ContextField;
296
297        use super::keys::{Consumer, Delivered, Stream, StreamSequence};
298
299        let cx = populated();
300        assert_eq!(StreamSequence.read(&cx), Some(42));
301        assert_eq!(Delivered.read(&cx), Some(3));
302        assert_eq!(Stream.read(&cx), Some("ORDERS".to_owned()));
303        assert_eq!(Consumer.read(&cx), Some("orders-worker".to_owned()));
304    }
305
306    #[test]
307    fn keys_read_none_without_jetstream_info() {
308        let cx = JetStreamContext::default();
309        assert_eq!(STREAM_SEQUENCE.get(&cx), None);
310        assert_eq!(CONSUMER_SEQUENCE.get(&cx), None);
311        assert_eq!(DELIVERED.get(&cx), None);
312        assert_eq!(PENDING.get(&cx), None);
313        assert_eq!(STREAM.get(&cx), None);
314        assert_eq!(CONSUMER.get(&cx), None);
315    }
316
317    #[test]
318    fn build_over_core_message_has_no_info() {
319        // A core delivery is constructible in-process; a JetStream `async_nats::jetstream::Message`
320        // is not, so the JetStream build path is compile-validated and exercised against a real
321        // server only. Core must yield an empty context (no native JetStream metadata).
322        let msg = NatsMessage::Core(Box::new(CoreMessage::new(async_nats::Message {
323            subject: "orders.created".into(),
324            reply: None,
325            payload: bytes::Bytes::from_static(b"{}"),
326            headers: None,
327            status: None,
328            description: None,
329            length: 2,
330        })));
331        let cx = JetStreamContext::build(&msg);
332        assert_eq!(cx, JetStreamContext::default());
333        assert_eq!(STREAM_SEQUENCE.get(&cx), None);
334    }
335}
ruststream_nats/context.rs

ruststream_nats/
context.rs
