ruststream_rdkafka/
context.rs

1//! Per-delivery context fields exposed to handlers.
2//!
3//! [`KafkaContext`] carries the Kafka delivery metadata that is not part of the payload or the
4//! headers. Request it in a handler by typing the context parameter as
5//! `Context<'_, KafkaContext>` and read individual fields with the zero-sized keys in [`keys`].
6
7use bytes::Bytes;
8use ruststream::{BuildContext, Field};
9
10use crate::message::KafkaMessage;
11
12/// Native Kafka delivery metadata, built once per delivery.
13#[derive(Debug, Clone, Default, PartialEq, Eq)]
14pub struct KafkaContext {
15    topic: String,
16    partition: i32,
17    offset: i64,
18    timestamp_millis: Option<i64>,
19    key: Option<Bytes>,
20}
21
22impl KafkaContext {
23    /// The topic the record was consumed from.
24    #[must_use]
25    pub fn topic(&self) -> &str {
26        &self.topic
27    }
28
29    /// The partition the record was consumed from.
30    #[must_use]
31    pub fn partition(&self) -> i32 {
32        self.partition
33    }
34
35    /// The record's offset within its partition.
36    #[must_use]
37    pub fn offset(&self) -> i64 {
38        self.offset
39    }
40
41    /// The record's timestamp in milliseconds since the epoch, when the broker provided one.
42    #[must_use]
43    pub fn timestamp_millis(&self) -> Option<i64> {
44        self.timestamp_millis
45    }
46
47    /// The record key, or `None` for keyless records.
48    #[must_use]
49    pub fn key(&self) -> Option<&[u8]> {
50        self.key.as_deref()
51    }
52}
53
54impl BuildContext<KafkaMessage> for KafkaContext {
55    fn build(msg: &KafkaMessage) -> Self {
56        Self {
57            topic: msg.topic().to_owned(),
58            partition: msg.partition(),
59            offset: msg.offset(),
60            timestamp_millis: msg.timestamp_millis(),
61            key: msg.key().map(Bytes::copy_from_slice),
62        }
63    }
64}
65
66/// Zero-sized [`Field`] keys reading one [`KafkaContext`] field each.
67pub mod keys {
68    use super::{Field, KafkaContext};
69
70    /// Reads the source topic name.
71    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
72    pub struct Topic;
73
74    impl Field<KafkaContext> for Topic {
75        type Value<'a> = &'a str;
76
77        fn get(self, src: &KafkaContext) -> &str {
78            src.topic()
79        }
80    }
81
82    /// Reads the source partition.
83    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
84    pub struct Partition;
85
86    impl Field<KafkaContext> for Partition {
87        type Value<'a> = i32;
88
89        fn get(self, src: &KafkaContext) -> i32 {
90            src.partition()
91        }
92    }
93
94    /// Reads the record offset.
95    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
96    pub struct Offset;
97
98    impl Field<KafkaContext> for Offset {
99        type Value<'a> = i64;
100
101        fn get(self, src: &KafkaContext) -> i64 {
102            src.offset()
103        }
104    }
105
106    /// Reads the record timestamp in milliseconds since the epoch, when present.
107    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
108    pub struct TimestampMillis;
109
110    impl Field<KafkaContext> for TimestampMillis {
111        type Value<'a> = Option<i64>;
112
113        fn get(self, src: &KafkaContext) -> Option<i64> {
114            src.timestamp_millis()
115        }
116    }
117
118    /// Reads the record key, when present.
119    #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
120    pub struct Key;
121
122    impl Field<KafkaContext> for Key {
123        type Value<'a> = Option<&'a [u8]>;
124
125        fn get(self, src: &KafkaContext) -> Option<&[u8]> {
126            src.key()
127        }
128    }
129}
ruststream_rdkafka/context.rs

ruststream_rdkafka/
context.rs
