crabka_connect_postgres/
config.rs

1use crabka_connect::{ConnectorConfig, SecretString};
2
3#[derive(Debug, Clone, ConnectorConfig)]
4pub struct PostgresSourceConfig {
5    #[config(required, secret)]
6    pub database_url: SecretString,
7    #[config(required)]
8    pub slot_name: String,
9    #[config(default = "crabka_connect")]
10    pub publication_name: String,
11    #[config(default = "public")]
12    pub schema: String,
13    #[config(name = "tables")]
14    pub table_names: Vec<String>,
15    #[config(default = 1000)]
16    pub max_messages_per_poll: u32,
17}
18
19#[cfg(test)]
20mod tests {
21    use crabka_connect::{ConfigKind, ConnectorConfig, EnvSecretResolver, ResolveOptions};
22    use serde_json::json;
23
24    use super::PostgresSourceConfig;
25
26    #[tokio::test]
27    async fn config_def_materializes_typed_postgres_config() {
28        let def = PostgresSourceConfig::config_def();
29        let keys = def
30            .keys()
31            .map(|key| (key.name.as_str(), key.kind, key.required))
32            .collect::<Vec<_>>();
33
34        assert!(keys.contains(&("database_url", ConfigKind::Secret, true)));
35        assert!(keys.contains(&("slot_name", ConfigKind::String, true)));
36        assert!(keys.contains(&("publication_name", ConfigKind::String, false)));
37        assert!(keys.contains(&("schema", ConfigKind::String, false)));
38        assert!(keys.contains(&("tables", ConfigKind::StringList, true)));
39        assert!(keys.contains(&("max_messages_per_poll", ConfigKind::UnsignedInteger, false,)));
40
41        let raw = serde_json::Map::from_iter([
42            (
43                "database_url".to_string(),
44                json!("postgres://localhost/app"),
45            ),
46            ("slot_name".to_string(), json!("crabka_slot")),
47            ("tables".to_string(), json!(["accounts", "transactions"])),
48        ]);
49
50        let resolved = def
51            .resolve_with_options(
52                raw,
53                &EnvSecretResolver,
54                ResolveOptions {
55                    allow_literal_secrets: true,
56                },
57            )
58            .await
59            .unwrap();
60        let config = PostgresSourceConfig::from_resolved(&resolved).unwrap();
61
62        assert_eq!(
63            config.database_url.expose_secret(),
64            "postgres://localhost/app"
65        );
66        assert_eq!(config.slot_name, "crabka_slot");
67        assert_eq!(config.publication_name, "crabka_connect");
68        assert_eq!(config.schema, "public");
69        assert_eq!(config.table_names, vec!["accounts", "transactions"]);
70        assert_eq!(config.max_messages_per_poll, 1000);
71    }
72}
crabka_connect_postgres/config.rs

crabka_connect_postgres/
config.rs
