krabka_connect_postgres/
config.rs

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

krabka_connect_postgres/
config.rs
