datum_mq/
record.rs

1use bytes::Bytes;
2
3/// Kafka record header.
4#[derive(Debug, Clone, PartialEq, Eq)]
5pub struct KafkaHeader {
6    pub key: String,
7    pub value: Option<Bytes>,
8}
9
10impl KafkaHeader {
11    #[must_use]
12    pub fn new(key: impl Into<String>, value: impl Into<Option<Bytes>>) -> Self {
13        Self {
14            key: key.into(),
15            value: value.into(),
16        }
17    }
18}
19
20/// Kafka record timestamp.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum KafkaTimestamp {
23    NotAvailable,
24    CreateTime(i64),
25    LogAppendTime(i64),
26}
27
28/// Owned Kafka consumer record.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct ConsumerRecord {
31    pub topic: String,
32    pub partition: i32,
33    pub offset: i64,
34    pub timestamp: KafkaTimestamp,
35    pub key: Option<Bytes>,
36    pub payload: Option<Bytes>,
37    pub headers: Vec<KafkaHeader>,
38}
39
40impl ConsumerRecord {
41    #[must_use]
42    pub fn topic_partition(&self) -> crate::TopicPartition {
43        crate::TopicPartition::new(self.topic.clone(), self.partition)
44    }
45}
46
47/// Owned Kafka producer record.
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct ProducerRecord {
50    pub topic: String,
51    pub key: Option<Bytes>,
52    pub payload: Option<Bytes>,
53    pub partition: Option<i32>,
54    pub timestamp: Option<i64>,
55    pub headers: Vec<KafkaHeader>,
56}
57
58impl ProducerRecord {
59    #[must_use]
60    pub fn new(topic: impl Into<String>, payload: impl Into<Option<Bytes>>) -> Self {
61        Self {
62            topic: topic.into(),
63            key: None,
64            payload: payload.into(),
65            partition: None,
66            timestamp: None,
67            headers: Vec::new(),
68        }
69    }
70
71    #[must_use]
72    pub fn with_key(mut self, key: impl Into<Option<Bytes>>) -> Self {
73        self.key = key.into();
74        self
75    }
76
77    #[must_use]
78    pub fn with_partition(mut self, partition: i32) -> Self {
79        self.partition = Some(partition);
80        self
81    }
82
83    #[must_use]
84    pub fn with_timestamp(mut self, timestamp: i64) -> Self {
85        self.timestamp = Some(timestamp);
86        self
87    }
88
89    #[must_use]
90    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<Option<Bytes>>) -> Self {
91        self.headers.push(KafkaHeader::new(key, value));
92        self
93    }
94}
datum_mq/record.rs

datum_mq/
record.rs
