drasi_source_sqlite/
config.rs1use anyhow::Result;
16use drasi_lib::bootstrap::BootstrapProvider;
17use drasi_lib::channels::DispatchMode;
18use drasi_lib::sources::base::SourceBaseParams;
19
20use crate::SqliteSource;
21
22#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct TableKeyConfig {
25 pub table: String,
27 pub key_columns: Vec<String>,
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct RestApiConfig {
34 pub host: String,
36 pub port: u16,
38}
39
40impl Default for RestApiConfig {
41 fn default() -> Self {
42 Self {
43 host: default_rest_host(),
44 port: default_rest_port(),
45 }
46 }
47}
48
49fn default_rest_host() -> String {
50 "127.0.0.1".to_string()
51}
52
53fn default_rest_port() -> u16 {
54 8080
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Default)]
60pub enum StartFrom {
61 #[default]
63 Beginning,
64 Now,
66 Timestamp(i64),
68}
69
70#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct SqliteSourceConfig {
73 pub path: Option<String>,
75 pub tables: Option<Vec<String>>,
77 pub table_keys: Vec<TableKeyConfig>,
79 pub rest_api: Option<RestApiConfig>,
81 pub start_from: StartFrom,
83}
84
85impl Default for SqliteSourceConfig {
86 fn default() -> Self {
87 Self {
88 path: None,
89 tables: None,
90 table_keys: Vec::new(),
91 rest_api: None,
92 start_from: StartFrom::Beginning,
93 }
94 }
95}
96
97pub struct SqliteSourceBuilder {
99 pub(crate) id: String,
100 path: Option<String>,
101 tables: Option<Vec<String>>,
102 table_keys: Vec<TableKeyConfig>,
103 rest_api: Option<RestApiConfig>,
104 start_from: StartFrom,
105 dispatch_mode: Option<DispatchMode>,
106 dispatch_buffer_capacity: Option<usize>,
107 bootstrap_provider: Option<Box<dyn BootstrapProvider + 'static>>,
108 auto_start: bool,
109}
110
111impl SqliteSourceBuilder {
112 pub fn new(id: impl Into<String>) -> Self {
114 Self {
115 id: id.into(),
116 path: None,
117 tables: None,
118 table_keys: Vec::new(),
119 rest_api: None,
120 start_from: StartFrom::Beginning,
121 dispatch_mode: None,
122 dispatch_buffer_capacity: None,
123 bootstrap_provider: None,
124 auto_start: true,
125 }
126 }
127
128 pub fn with_path(mut self, path: impl Into<String>) -> Self {
130 self.path = Some(path.into());
131 self
132 }
133
134 pub fn in_memory(mut self) -> Self {
136 self.path = None;
137 self
138 }
139
140 pub fn with_tables(mut self, tables: Vec<String>) -> Self {
142 self.tables = Some(tables);
143 self
144 }
145
146 pub fn with_table_keys(mut self, table_keys: Vec<TableKeyConfig>) -> Self {
148 self.table_keys = table_keys;
149 self
150 }
151
152 pub fn with_rest_api(mut self, config: RestApiConfig) -> Self {
154 self.rest_api = Some(config);
155 self
156 }
157
158 pub fn start_from(mut self, start_from: StartFrom) -> Self {
160 self.start_from = start_from;
161 self
162 }
163
164 pub fn with_dispatch_mode(mut self, mode: DispatchMode) -> Self {
166 self.dispatch_mode = Some(mode);
167 self
168 }
169
170 pub fn with_dispatch_buffer_capacity(mut self, capacity: usize) -> Self {
172 self.dispatch_buffer_capacity = Some(capacity);
173 self
174 }
175
176 pub fn with_bootstrap_provider(mut self, provider: impl BootstrapProvider + 'static) -> Self {
178 self.bootstrap_provider = Some(Box::new(provider));
179 self
180 }
181
182 pub fn auto_start(mut self, auto_start: bool) -> Self {
184 self.auto_start = auto_start;
185 self
186 }
187
188 pub fn build(self) -> Result<SqliteSource> {
190 let config = SqliteSourceConfig {
191 path: self.path,
192 tables: self.tables,
193 table_keys: self.table_keys,
194 rest_api: self.rest_api,
195 start_from: self.start_from,
196 };
197
198 let mut params = SourceBaseParams::new(self.id);
199 if let Some(mode) = self.dispatch_mode {
200 params = params.with_dispatch_mode(mode);
201 }
202 if let Some(capacity) = self.dispatch_buffer_capacity {
203 params = params.with_dispatch_buffer_capacity(capacity);
204 }
205 if let Some(provider) = self.bootstrap_provider {
206 params = params.with_bootstrap_provider(provider);
207 }
208 params = params.with_auto_start(self.auto_start);
209
210 let base = drasi_lib::sources::base::SourceBase::new(params)?;
211 SqliteSource::from_parts(base, config)
212 }
213}
214
215#[cfg(test)]
216mod tests {
217 use super::*;
218 use drasi_lib::Source;
219
220 #[test]
221 fn sqlite_source_config_defaults_are_stable() {
222 let config = SqliteSourceConfig::default();
223 assert!(config.path.is_none());
224 assert!(config.tables.is_none());
225 assert!(config.table_keys.is_empty());
226 assert!(config.rest_api.is_none());
227 assert_eq!(config.start_from, StartFrom::Beginning);
228 }
229
230 #[test]
231 fn builder_applies_rest_and_dispatch_configuration() {
232 let source = SqliteSource::builder("test-source")
233 .with_path("/tmp/test.db")
234 .with_tables(vec!["sensors".to_string()])
235 .with_table_keys(vec![TableKeyConfig {
236 table: "sensors".to_string(),
237 key_columns: vec!["id".to_string()],
238 }])
239 .with_rest_api(RestApiConfig {
240 host: "127.0.0.1".to_string(), port: 9001,
242 })
243 .with_dispatch_mode(DispatchMode::Channel)
244 .with_dispatch_buffer_capacity(42)
245 .auto_start(false)
246 .build()
247 .expect("failed to build sqlite source");
248
249 let properties = source.properties();
250 assert_eq!(
251 properties.get("path"),
252 Some(&serde_json::json!("/tmp/test.db"))
253 );
254 assert_eq!(
255 properties.get("tables"),
256 Some(&serde_json::json!(["sensors"]))
257 );
258 assert!(properties.contains_key("restApi"));
259 assert!(!source.auto_start());
260 }
261}