cufflink_types/manifest/
stream.rs

1use serde::{Deserialize, Serialize};
2
3/// A JetStream stream the service needs. The platform creates it, or
4/// extends the one that exists; a deploy never deletes a stream or removes
5/// a subject from one.
6#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
7pub struct StreamDefinition {
8    pub name: String,
9    pub subjects: Vec<String>,
10    pub retention: StreamRetention,
11    #[serde(default, skip_serializing_if = "Option::is_none")]
12    pub max_age_secs: Option<u64>,
13    #[serde(default, skip_serializing_if = "Option::is_none")]
14    pub max_msgs: Option<i64>,
15    #[serde(default, skip_serializing_if = "Option::is_none")]
16    pub max_bytes: Option<i64>,
17    #[serde(default, skip_serializing_if = "StreamStorage::is_file")]
18    pub storage: StreamStorage,
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    pub duplicate_window_secs: Option<u64>,
21    #[serde(default, skip_serializing_if = "DiscardSpec::is_old")]
22    pub discard: DiscardSpec,
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub description: Option<String>,
25}
26
27#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "snake_case")]
29pub enum StreamRetention {
30    #[default]
31    Limits,
32    WorkQueue,
33    Interest,
34}
35
36impl StreamRetention {
37    pub fn as_str(self) -> &'static str {
38        match self {
39            StreamRetention::Limits => "limits",
40            StreamRetention::WorkQueue => "work_queue",
41            StreamRetention::Interest => "interest",
42        }
43    }
44}
45
46#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
47#[serde(rename_all = "snake_case")]
48pub enum StreamStorage {
49    #[default]
50    File,
51    Memory,
52}
53
54impl StreamStorage {
55    pub fn is_file(&self) -> bool {
56        *self == StreamStorage::File
57    }
58}
59
60/// What a full stream does with a new message.
61#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(rename_all = "snake_case")]
63pub enum DiscardSpec {
64    /// Drops its oldest message.
65    #[default]
66    Old,
67    /// Refuses the new one.
68    New,
69}
70
71impl DiscardSpec {
72    pub fn is_old(&self) -> bool {
73        *self == DiscardSpec::Old
74    }
75}
76
77#[cfg(test)]
78mod tests {
79    use super::*;
80    use serde_json::json;
81
82    #[test]
83    fn a_minimal_stream_carries_only_what_it_declares() {
84        let stream = StreamDefinition {
85            name: "NOTES".into(),
86            subjects: vec!["notes.created".into()],
87            retention: StreamRetention::Limits,
88            max_age_secs: None,
89            max_msgs: None,
90            max_bytes: None,
91            storage: StreamStorage::File,
92            duplicate_window_secs: None,
93            discard: DiscardSpec::Old,
94            description: None,
95        };
96        let wire = json!({"name": "NOTES", "subjects": ["notes.created"], "retention": "limits"});
97        assert_eq!(serde_json::to_value(&stream).unwrap(), wire);
98        assert_eq!(
99            serde_json::from_value::<StreamDefinition>(wire).unwrap(),
100            stream
101        );
102    }
103
104    #[test]
105    fn every_field_round_trips() {
106        let stream = StreamDefinition {
107            name: "QUOTES".into(),
108            subjects: vec!["quotes.>".into()],
109            retention: StreamRetention::WorkQueue,
110            max_age_secs: Some(2_592_000),
111            max_msgs: Some(100_000),
112            max_bytes: Some(268_435_456),
113            storage: StreamStorage::Memory,
114            duplicate_window_secs: Some(120),
115            discard: DiscardSpec::New,
116            description: Some("quote jobs".into()),
117        };
118        let wire = json!({
119            "name": "QUOTES",
120            "subjects": ["quotes.>"],
121            "retention": "work_queue",
122            "max_age_secs": 2592000,
123            "max_msgs": 100000,
124            "max_bytes": 268435456,
125            "storage": "memory",
126            "duplicate_window_secs": 120,
127            "discard": "new",
128            "description": "quote jobs"
129        });
130        assert_eq!(serde_json::to_value(&stream).unwrap(), wire);
131        assert_eq!(
132            serde_json::from_value::<StreamDefinition>(wire).unwrap(),
133            stream
134        );
135        assert_eq!(
136            serde_json::to_value(StreamRetention::Interest).unwrap(),
137            json!(StreamRetention::Interest.as_str())
138        );
139    }
140}
cufflink_types/manifest/stream.rs

cufflink_types/manifest/
stream.rs
