singer 0.3.0

Rust Singer SDK
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
95
96
97
98
99
100
101
102
103
104
105
106
107

use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::schema::JsonSchema;

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "type", rename_all = "SCREAMING_SNAKE_CASE")]
pub enum Message {
    Schema(Schema),
    Record(Record),
    State(State),
    ActivateVersion(ActivateVersion),
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Record {
    pub stream: String,
    pub record: Value,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time_extracted: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub version: Option<i64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Schema {
    pub stream: String,
    pub schema: JsonSchema,
    pub key_properties: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub bookmark_properties: Option<Vec<String>>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct State {
    pub value: Value,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ActivateVersion {
    pub stream: String,
    pub version: i64,
}

#[cfg(test)]
pub mod tests {
    use serde_json::json;

    use crate::messages::{ActivateVersion, Message, Record, Schema, State};

    #[test]
    fn test_record() {
        let input =
            r#"{"type": "RECORD", "stream": "users", "record": {"id": 1, "name": "Chris"}}"#;

        let record: Message = serde_json::from_str(input).unwrap();

        let expected = Message::Record(Record {
            stream: "users".to_string(),
            record: json!({"id": 1, "name": "Chris"}),
            time_extracted: None,
        });
        assert_eq!(record, expected);
    }

    #[test]
    fn test_schema() {
        let input = r#"{"type": "SCHEMA", "stream": "users", "key_properties": ["id"], "schema": {"required": ["id"], "type": "object", "properties": {"id": {"type": "integer"}}}}"#;

        let schema: Message = serde_json::from_str(input).unwrap();

        let expected = Message::Schema(Schema {
            stream: "users".to_string(),
            key_properties: vec!["id".to_string()],
            schema: serde_json::from_value(
                json!({"required": ["id"], "type": "object", "properties": {"id": {"type": "integer"}}}),
            ).unwrap(),
            bookmark_properties: None,
        });
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_state() {
        let input = r#"{"type": "STATE", "value": {"users": 2, "locations": 1}}"#;

        let schema: Message = serde_json::from_str(input).unwrap();

        let expected = Message::State(State {
            value: json!({"users": 2, "locations": 1}),
        });
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_activate_version() {
        let input = r#"{"type": "ACTIVATE_VERSION", "stream": "users", "version": 1695106400957}"#;

        let schema: Message = serde_json::from_str(input).unwrap();

        let expected = Message::ActivateVersion(ActivateVersion {
            stream: "users".to_string(),
            version: 1695106400957,
        });
        assert_eq!(schema, expected);
    }
}