camel_component_sql/
lib.rs

1pub mod bundle;
2pub mod config;
3pub mod consumer;
4pub mod endpoint;
5pub mod headers;
6pub mod health;
7pub mod pool_factory;
8pub mod producer;
9pub mod query;
10pub(crate) mod utils;
11
12use std::sync::Arc;
13
14use camel_api::datasource::DatasourceCatalog;
15use camel_component_api::CamelError;
16use camel_component_api::UriConfig;
17use camel_component_api::{Component, Endpoint};
18use sqlx::AnyPool;
19use tokio::sync::OnceCell;
20
21pub use bundle::SqlBundle;
22pub use config::{
23    PollStrategy, ProcessingStrategy, SqlEndpointConfig, SqlGlobalConfig, SqlOutputType,
24    TransactionMode,
25};
26pub use health::SqlHealthCheck;
27
28type SharedPool = Arc<OnceCell<Arc<AnyPool>>>;
29
30pub struct SqlComponent {
31    config: Option<SqlGlobalConfig>,
32    catalog: Option<Arc<dyn DatasourceCatalog>>,
33}
34
35impl SqlComponent {
36    pub fn new() -> Self {
37        Self {
38            config: None,
39            catalog: None,
40        }
41    }
42
43    pub fn with_config(config: SqlGlobalConfig) -> Self {
44        Self {
45            config: Some(config),
46            catalog: None,
47        }
48    }
49
50    pub fn with_optional_config(config: Option<SqlGlobalConfig>) -> Self {
51        Self {
52            config,
53            catalog: None,
54        }
55    }
56
57    pub fn with_config_and_catalog(
58        config: SqlGlobalConfig,
59        catalog: Arc<dyn DatasourceCatalog>,
60    ) -> Self {
61        Self {
62            config: Some(config),
63            catalog: Some(catalog),
64        }
65    }
66
67    pub fn catalog(&self) -> Option<&Arc<dyn DatasourceCatalog>> {
68        self.catalog.as_ref()
69    }
70}
71
72impl Default for SqlComponent {
73    fn default() -> Self {
74        Self::new()
75    }
76}
77
78impl Component for SqlComponent {
79    fn scheme(&self) -> &str {
80        "sql"
81    }
82
83    fn create_endpoint(
84        &self,
85        uri: &str,
86        ctx: &dyn camel_component_api::ComponentContext,
87    ) -> Result<Box<dyn Endpoint>, CamelError> {
88        let mut config = SqlEndpointConfig::from_uri(uri)?;
89
90        if config.datasource_name.is_some() && self.catalog.is_none() {
91            return Err(CamelError::Config(
92                "datasource parameter requires catalog — no datasource catalog configured".into(),
93            ));
94        }
95
96        if let Some(ref ds_name) = config.datasource_name
97            && let Some(ref catalog) = self.catalog
98        {
99            let ds_config = catalog.get_config(ds_name).ok_or_else(|| {
100                CamelError::Config(format!("datasource '{}' not found in catalog", ds_name))
101            })?;
102            config.db_url = ds_config.db_url;
103        }
104
105        if let Some(ref global_config) = self.config {
106            config.apply_defaults(global_config);
107        }
108        config.resolve_defaults();
109        let pool: SharedPool = Arc::new(OnceCell::new());
110        let health_check = SqlHealthCheck::new(
111            Arc::clone(&pool),
112            self.catalog.clone(),
113            config.datasource_name.clone(),
114        );
115        ctx.register_current_route_health_check(Arc::new(health_check));
116
117        if config.datasource_name.is_some()
118            && let Some(ref catalog) = self.catalog
119        {
120            Ok(Box::new(endpoint::SqlEndpoint::new_with_pool_and_catalog(
121                uri.to_string(),
122                config,
123                pool,
124                Arc::clone(catalog),
125            )))
126        } else {
127            Ok(Box::new(endpoint::SqlEndpoint::new_with_pool(
128                uri.to_string(),
129                config,
130                pool,
131            )))
132        }
133    }
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139    use camel_component_api::Component;
140    use camel_component_api::NoOpComponentContext;
141
142    #[test]
143    fn test_component_scheme() {
144        let c = SqlComponent::new();
145        assert_eq!(c.scheme(), "sql");
146    }
147
148    #[test]
149    fn test_component_creates_endpoint() {
150        let c = SqlComponent::new();
151        let ctx = NoOpComponentContext;
152        let ep = c.create_endpoint("sql:select 1?db_url=postgres://localhost/test", &ctx);
153        assert!(ep.is_ok());
154    }
155
156    #[test]
157    fn test_component_rejects_wrong_scheme() {
158        let c = SqlComponent::new();
159        let ctx = NoOpComponentContext;
160        let ep = c.create_endpoint("redis://localhost", &ctx);
161        assert!(ep.is_err());
162    }
163
164    #[test]
165    fn test_endpoint_uri() {
166        let c = SqlComponent::new();
167        let ctx = NoOpComponentContext;
168        let ep = c
169            .create_endpoint("sql:select 1?db_url=postgres://localhost/test", &ctx)
170            .unwrap();
171        assert_eq!(ep.uri(), "sql:select 1?db_url=postgres://localhost/test");
172    }
173
174    #[test]
175    fn test_component_with_global_config() {
176        let global = SqlGlobalConfig::default().with_max_connections(20);
177        let c = SqlComponent::with_config(global);
178        let ctx = NoOpComponentContext;
179        // Verify the component can create endpoints with global config applied
180        assert_eq!(c.scheme(), "sql");
181        let ep = c.create_endpoint("sql:select 1?db_url=postgres://localhost/test", &ctx);
182        assert!(ep.is_ok());
183    }
184
185    #[test]
186    fn test_global_config_applied_to_endpoint() {
187        // Verify that when URI does NOT set pool params, global config fills them in.
188        // Tests the same logic as create_endpoint: from_uri + apply_defaults + resolve_defaults.
189        let global = SqlGlobalConfig::default()
190            .with_max_connections(20)
191            .with_min_connections(3)
192            .with_idle_timeout_secs(600)
193            .with_max_lifetime_secs(3600);
194        let mut cfg =
195            config::SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test")
196                .unwrap();
197        cfg.apply_defaults(&global);
198        cfg.resolve_defaults();
199        assert_eq!(cfg.max_connections, Some(20));
200        assert_eq!(cfg.min_connections, Some(3));
201        assert_eq!(cfg.idle_timeout_secs, Some(600));
202        assert_eq!(cfg.max_lifetime_secs, Some(3600));
203    }
204
205    #[test]
206    fn test_uri_param_wins_over_global_config() {
207        // Verify that URI-set pool params are NOT overridden by global config.
208        let global = SqlGlobalConfig::default()
209            .with_max_connections(20)
210            .with_min_connections(3);
211        let mut cfg = config::SqlEndpointConfig::from_uri(
212            "sql:select 1?db_url=postgres://localhost/test&maxConnections=99&minConnections=7",
213        )
214        .unwrap();
215        cfg.apply_defaults(&global);
216        cfg.resolve_defaults();
217        assert_eq!(cfg.max_connections, Some(99)); // URI wins
218        assert_eq!(cfg.min_connections, Some(7)); // URI wins
219    }
220}
camel_component_sql/lib.rs

camel_component_sql/
lib.rs
