datum-mq 0.11.2

Native Kafka sources and sinks for Datum streams
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

use bytes::Bytes;

/// Kafka record header.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KafkaHeader {
    pub key: String,
    pub value: Option<Bytes>,
}

impl KafkaHeader {
    #[must_use]
    pub fn new(key: impl Into<String>, value: impl Into<Option<Bytes>>) -> Self {
        Self {
            key: key.into(),
            value: value.into(),
        }
    }
}

/// Kafka record timestamp.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum KafkaTimestamp {
    NotAvailable,
    CreateTime(i64),
    LogAppendTime(i64),
}

/// Owned Kafka consumer record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumerRecord {
    pub topic: String,
    pub partition: i32,
    pub offset: i64,
    pub timestamp: KafkaTimestamp,
    pub key: Option<Bytes>,
    pub payload: Option<Bytes>,
    pub headers: Vec<KafkaHeader>,
}

impl ConsumerRecord {
    #[must_use]
    pub fn topic_partition(&self) -> crate::TopicPartition {
        crate::TopicPartition::new(self.topic.clone(), self.partition)
    }
}

/// Owned Kafka producer record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProducerRecord {
    pub topic: String,
    pub key: Option<Bytes>,
    pub payload: Option<Bytes>,
    pub partition: Option<i32>,
    pub timestamp: Option<i64>,
    pub headers: Vec<KafkaHeader>,
}

impl ProducerRecord {
    #[must_use]
    pub fn new(topic: impl Into<String>, payload: impl Into<Option<Bytes>>) -> Self {
        Self {
            topic: topic.into(),
            key: None,
            payload: payload.into(),
            partition: None,
            timestamp: None,
            headers: Vec::new(),
        }
    }

    #[must_use]
    pub fn with_key(mut self, key: impl Into<Option<Bytes>>) -> Self {
        self.key = key.into();
        self
    }

    #[must_use]
    pub fn with_partition(mut self, partition: i32) -> Self {
        self.partition = Some(partition);
        self
    }

    #[must_use]
    pub fn with_timestamp(mut self, timestamp: i64) -> Self {
        self.timestamp = Some(timestamp);
        self
    }

    #[must_use]
    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<Option<Bytes>>) -> Self {
        self.headers.push(KafkaHeader::new(key, value));
        self
    }
}