link-common 0.5.2-rc.2

Shared Rust implementation for KalamDB link crates
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

use serde::{Deserialize, Serialize};

use super::{PayloadMode, TopicOp};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConsumerRecord {
    pub topic_id: String,
    pub topic_name: String,
    pub partition_id: u32,
    pub offset: u64,
    pub message_id: Option<String>,
    pub source_table: String,
    pub op: TopicOp,
    pub timestamp_ms: u64,
    pub payload_mode: PayloadMode,
    pub payload: Vec<u8>,
}

#[cfg(feature = "tokio-runtime")]
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct ConsumerRecordWire {
    pub topic_id: String,
    pub partition_id: u32,
    pub offset: u64,
    #[serde(default, alias = "key")]
    pub message_id: Option<String>,
    #[serde(default)]
    pub source_table: Option<String>,
    #[serde(default)]
    pub op: Option<TopicOp>,
    #[serde(default, rename = "timestamp_ms", alias = "ts")]
    pub timestamp_ms: u64,
    #[serde(default)]
    pub payload_mode: Option<PayloadMode>,
    #[serde(default, with = "base64_bytes")]
    pub payload: Vec<u8>,
}

#[cfg(feature = "tokio-runtime")]
impl ConsumerRecordWire {
    pub fn into_record(self, topic_name: String) -> ConsumerRecord {
        ConsumerRecord {
            topic_id: self.topic_id,
            topic_name,
            partition_id: self.partition_id,
            offset: self.offset,
            message_id: self.message_id,
            source_table: self.source_table.unwrap_or_default(),
            op: self.op.unwrap_or(TopicOp::Insert),
            timestamp_ms: self.timestamp_ms,
            payload_mode: self.payload_mode.unwrap_or(PayloadMode::Full),
            payload: self.payload,
        }
    }
}

#[cfg(feature = "tokio-runtime")]
mod base64_bytes {
    use base64::{engine::general_purpose::STANDARD, Engine};
    use serde::{de::Error, Deserialize, Deserializer, Serializer};

    #[allow(dead_code)]
    pub fn serialize<S>(bytes: &Vec<u8>, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let encoded = STANDARD.encode(bytes);
        serializer.serialize_str(&encoded)
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
    where
        D: Deserializer<'de>,
    {
        let encoded = String::deserialize(deserializer)?;
        STANDARD.decode(encoded.as_bytes()).map_err(D::Error::custom)
    }
}