ruststream_fred/
context.rs

1//! Optional typed per-delivery context exposing native Redis metadata, one struct per transport.
2//!
3//! A handler can read native Redis metadata for the message it is processing by compile-time
4//! [`Field`] key, with no hashing, boxing, or downcasting. The runtime builds the context value once
5//! per delivery (via [`BuildContext`]) from the concrete broker message, then the handler reads a
6//! field with `ctx.context(key)`.
7//!
8//! This is purely additive. A handler that declares the default `()` context (the vast majority) is
9//! unaffected: the blanket `impl BuildContext<M> for ()` still applies, so opting in costs nothing
10//! to those who do not.
11//!
12//! # What is exposed
13//!
14//! Only genuinely-native metadata that is not already reachable off the payload or
15//! [`Headers`](ruststream::Headers) is surfaced here:
16//!
17//! * [`StreamContext`] (Redis Streams) - the stream entry id and the consumer group. The native
18//!   reclaim delivery-count and idle time stay header-surfaced
19//!   ([`DELIVERY_COUNT_HEADER`](crate::DELIVERY_COUNT_HEADER) /
20//!   [`IDLE_MS_HEADER`](crate::IDLE_MS_HEADER)) and are deliberately not duplicated.
21//! * [`PubSubContext`] (Redis Pub/Sub) - the concrete channel the message arrived on and whether it
22//!   matched through a `PSUBSCRIBE` pattern (for a pattern subscription the channel differs from the
23//!   registered glob).
24//!
25//! # Examples
26//!
27//! ```
28//! use ruststream::runtime::{Context, HandlerResult};
29//! use ruststream_fred::context::{StreamContext, keys};
30//!
31//! // A handler over the Streams transport reading the native entry id and consumer group.
32//! async fn handle(order: &Vec<u8>, ctx: &mut Context<'_, StreamContext>) -> HandlerResult {
33//!     if let Some(id) = ctx.context(keys::EntryId) {
34//!         let _ = id; // e.g. log the stream entry id `1700000000000-0`
35//!     }
36//!     let _group = ctx.context(keys::ConsumerGroup);
37//!     HandlerResult::Ack
38//! }
39//! # let _ = handle;
40//! ```
41
42use ruststream::{BuildContext, Field};
43
44use crate::message::RedisMessage;
45use crate::pubsub::RedisPubSubMessage;
46
47/// Per-delivery context for a Redis Streams delivery ([`RedisMessage`]).
48///
49/// Built once per delivery from the message. Read its fields by [`keys`] key off a
50/// [`Context`](ruststream::runtime::Context).
51#[derive(Debug, Clone, Default, PartialEq, Eq)]
52pub struct StreamContext {
53    entry_id: Option<String>,
54    consumer_group: Option<String>,
55}
56
57impl StreamContext {
58    /// Constructs a context directly from its native fields (mainly for tests).
59    #[must_use]
60    pub fn new(entry_id: Option<String>, consumer_group: Option<String>) -> Self {
61        Self {
62            entry_id,
63            consumer_group,
64        }
65    }
66
67    /// The stream entry id (for example `1700000000000-0`) this delivery was read at.
68    #[must_use]
69    pub fn entry_id(&self) -> Option<&str> {
70        self.entry_id.as_deref()
71    }
72
73    /// The consumer group this delivery was read through.
74    #[must_use]
75    pub fn consumer_group(&self) -> Option<&str> {
76        self.consumer_group.as_deref()
77    }
78}
79
80impl BuildContext<RedisMessage> for StreamContext {
81    fn build(msg: &RedisMessage) -> Self {
82        Self {
83            entry_id: msg.id().map(str::to_owned),
84            consumer_group: msg.group().map(str::to_owned),
85        }
86    }
87}
88
89/// Per-delivery context for a Redis Pub/Sub delivery ([`RedisPubSubMessage`]).
90#[derive(Debug, Clone, Default, PartialEq, Eq)]
91pub struct PubSubContext {
92    channel: String,
93    from_pattern: bool,
94}
95
96impl PubSubContext {
97    /// Constructs a context directly from its native fields (mainly for tests).
98    #[must_use]
99    pub fn new(channel: impl Into<String>, from_pattern: bool) -> Self {
100        Self {
101            channel: channel.into(),
102            from_pattern,
103        }
104    }
105
106    /// The concrete channel this message arrived on (the matched channel, not the subscription
107    /// glob, for a pattern subscription).
108    #[must_use]
109    pub fn channel(&self) -> &str {
110        &self.channel
111    }
112
113    /// Whether the delivery matched through a `PSUBSCRIBE` pattern rather than an exact subscribe.
114    #[must_use]
115    pub fn from_pattern(&self) -> bool {
116        self.from_pattern
117    }
118}
119
120impl BuildContext<RedisPubSubMessage> for PubSubContext {
121    fn build(msg: &RedisPubSubMessage) -> Self {
122        Self {
123            channel: msg.channel().to_owned(),
124            from_pattern: msg.from_pattern(),
125        }
126    }
127}
128
129/// Compile-time [`Field`] keys, one per native field, read with `ctx.context(key)`.
130///
131/// Each key is a zero-sized selector implementing [`Field`] only for the context type that carries
132/// its field, so applying a key to the wrong transport's context is a compile error.
133pub mod keys {
134    use ruststream::ContextField;
135
136    use super::{Field, PubSubContext, StreamContext};
137
138    /// Reads the stream entry id off a [`StreamContext`].
139    #[derive(Debug, Clone, Copy, Default)]
140    pub struct EntryId;
141
142    impl Field<StreamContext> for EntryId {
143        type Value<'a> = Option<&'a str>;
144        fn get(self, src: &StreamContext) -> Option<&str> {
145            src.entry_id()
146        }
147    }
148
149    impl ContextField for EntryId {
150        type Context = StreamContext;
151        type Value = Option<String>;
152        fn read(self, src: &StreamContext) -> Option<String> {
153            src.entry_id().map(str::to_owned)
154        }
155    }
156
157    /// Reads the consumer group off a [`StreamContext`].
158    #[derive(Debug, Clone, Copy, Default)]
159    pub struct ConsumerGroup;
160
161    impl Field<StreamContext> for ConsumerGroup {
162        type Value<'a> = Option<&'a str>;
163        fn get(self, src: &StreamContext) -> Option<&str> {
164            src.consumer_group()
165        }
166    }
167
168    impl ContextField for ConsumerGroup {
169        type Context = StreamContext;
170        type Value = Option<String>;
171        fn read(self, src: &StreamContext) -> Option<String> {
172            src.consumer_group().map(str::to_owned)
173        }
174    }
175
176    /// Reads the concrete channel off a [`PubSubContext`].
177    #[derive(Debug, Clone, Copy, Default)]
178    pub struct Channel;
179
180    impl Field<PubSubContext> for Channel {
181        type Value<'a> = &'a str;
182        fn get(self, src: &PubSubContext) -> &str {
183            src.channel()
184        }
185    }
186
187    impl ContextField for Channel {
188        type Context = PubSubContext;
189        type Value = String;
190        fn read(self, src: &PubSubContext) -> String {
191            src.channel().to_owned()
192        }
193    }
194
195    /// Reads whether a [`PubSubContext`] delivery matched through a `PSUBSCRIBE` pattern.
196    #[derive(Debug, Clone, Copy, Default)]
197    pub struct FromPattern;
198
199    impl Field<PubSubContext> for FromPattern {
200        type Value<'a> = bool;
201        fn get(self, src: &PubSubContext) -> bool {
202            src.from_pattern()
203        }
204    }
205
206    impl ContextField for FromPattern {
207        type Context = PubSubContext;
208        type Value = bool;
209        fn read(self, src: &PubSubContext) -> bool {
210            src.from_pattern()
211        }
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::keys::{Channel, ConsumerGroup, EntryId, FromPattern};
218    use super::{PubSubContext, StreamContext};
219    use ruststream::{ContextField, Field};
220
221    #[test]
222    fn stream_keys_read_native_fields() {
223        let cx = StreamContext::new(
224            Some("1700000000000-0".to_owned()),
225            Some("workers".to_owned()),
226        );
227        assert_eq!(EntryId.get(&cx), Some("1700000000000-0"));
228        assert_eq!(ConsumerGroup.get(&cx), Some("workers"));
229    }
230
231    #[test]
232    fn stream_keys_absent_when_settled() {
233        let cx = StreamContext::new(None, None);
234        assert_eq!(EntryId.get(&cx), None);
235        assert_eq!(ConsumerGroup.get(&cx), None);
236    }
237
238    #[test]
239    fn pubsub_keys_read_channel_and_pattern_flag() {
240        let exact = PubSubContext::new("events", false);
241        assert_eq!(Channel.get(&exact), "events");
242        assert!(!FromPattern.get(&exact));
243
244        let matched = PubSubContext::new("events.user", true);
245        assert_eq!(Channel.get(&matched), "events.user");
246        assert!(FromPattern.get(&matched));
247    }
248
249    #[test]
250    fn context_field_keys_yield_owned_values() {
251        let stream = StreamContext::new(
252            Some("1700000000000-0".to_owned()),
253            Some("workers".to_owned()),
254        );
255        assert_eq!(
256            <EntryId as ContextField>::read(EntryId, &stream),
257            Some("1700000000000-0".to_owned())
258        );
259        assert_eq!(
260            <ConsumerGroup as ContextField>::read(ConsumerGroup, &stream),
261            Some("workers".to_owned())
262        );
263
264        let pubsub = PubSubContext::new("orders.eu", true);
265        assert_eq!(
266            <Channel as ContextField>::read(Channel, &pubsub),
267            "orders.eu".to_owned()
268        );
269        assert!(<FromPattern as ContextField>::read(FromPattern, &pubsub));
270    }
271}
ruststream_fred/context.rs

ruststream_fred/
context.rs
