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::component_metadata::ComponentMetadata;
15use camel_api::datasource::DatasourceCatalog;
16use camel_component_api::CamelError;
17use camel_component_api::UriConfig;
18use camel_component_api::{Component, Endpoint};
19use sqlx::AnyPool;
20use tokio::sync::OnceCell;
21
22pub use bundle::SqlBundle;
23pub use config::{
24    PollStrategy, ProcessingStrategy, SqlEndpointConfig, SqlGlobalConfig, SqlOutputType,
25    TransactionMode,
26};
27pub use health::SqlHealthCheck;
28
29type SharedPool = Arc<OnceCell<Arc<AnyPool>>>;
30
31pub struct SqlComponent {
32    config: Option<SqlGlobalConfig>,
33    catalog: Option<Arc<dyn DatasourceCatalog>>,
34}
35
36impl SqlComponent {
37    pub fn new() -> Self {
38        Self {
39            config: None,
40            catalog: None,
41        }
42    }
43
44    pub fn with_config(config: SqlGlobalConfig) -> Self {
45        Self {
46            config: Some(config),
47            catalog: None,
48        }
49    }
50
51    pub fn with_optional_config(config: Option<SqlGlobalConfig>) -> Self {
52        Self {
53            config,
54            catalog: None,
55        }
56    }
57
58    pub fn with_config_and_catalog(
59        config: SqlGlobalConfig,
60        catalog: Arc<dyn DatasourceCatalog>,
61    ) -> Self {
62        Self {
63            config: Some(config),
64            catalog: Some(catalog),
65        }
66    }
67
68    pub fn catalog(&self) -> Option<&Arc<dyn DatasourceCatalog>> {
69        self.catalog.as_ref()
70    }
71}
72
73impl Default for SqlComponent {
74    fn default() -> Self {
75        Self::new()
76    }
77}
78
79impl Component for SqlComponent {
80    fn scheme(&self) -> &str {
81        "sql"
82    }
83
84    fn metadata(&self) -> ComponentMetadata {
85        SqlEndpointConfig::metadata()
86    }
87
88    fn create_endpoint(
89        &self,
90        uri: &str,
91        ctx: &dyn camel_component_api::ComponentContext,
92    ) -> Result<Box<dyn Endpoint>, CamelError> {
93        let mut config = SqlEndpointConfig::from_uri(uri)?;
94
95        if config.datasource_name.is_some() && self.catalog.is_none() {
96            return Err(CamelError::Config(
97                "datasource parameter requires catalog — no datasource catalog configured".into(),
98            ));
99        }
100
101        if let Some(ref ds_name) = config.datasource_name
102            && let Some(ref catalog) = self.catalog
103        {
104            let ds_config = catalog.get_config(ds_name).ok_or_else(|| {
105                CamelError::Config(format!("datasource '{}' not found in catalog", ds_name))
106            })?;
107            config.db_url = ds_config.db_url;
108        }
109
110        if let Some(ref global_config) = self.config {
111            config.apply_defaults(global_config);
112        }
113        config.resolve_defaults();
114        let pool: SharedPool = Arc::new(OnceCell::new());
115        let health_check = SqlHealthCheck::new(
116            Arc::clone(&pool),
117            self.catalog.clone(),
118            config.datasource_name.clone(),
119        );
120        ctx.register_current_route_health_check(Arc::new(health_check));
121
122        if config.datasource_name.is_some()
123            && let Some(ref catalog) = self.catalog
124        {
125            Ok(Box::new(endpoint::SqlEndpoint::new_with_pool_and_catalog(
126                uri.to_string(),
127                config,
128                pool,
129                Arc::clone(catalog),
130            )))
131        } else {
132            Ok(Box::new(endpoint::SqlEndpoint::new_with_pool(
133                uri.to_string(),
134                config,
135                pool,
136            )))
137        }
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use camel_component_api::Component;
145    use camel_component_api::NoOpComponentContext;
146
147    #[test]
148    fn test_component_scheme() {
149        let c = SqlComponent::new();
150        assert_eq!(c.scheme(), "sql");
151    }
152
153    #[test]
154    fn test_component_metadata_nonempty() {
155        let c = SqlComponent::new();
156        let meta = c.metadata();
157        assert_eq!(meta.scheme, "sql");
158        assert!(!meta.description.is_empty());
159        assert!(
160            !meta.uri_options.is_empty(),
161            "uri_options must be non-empty"
162        );
163        let names: Vec<&str> = meta.uri_options.iter().map(|o| o.name.as_str()).collect();
164        assert!(
165            names.contains(&"db_url"),
166            "expected db_url in uri_options, got: {names:?}"
167        );
168        assert!(
169            names.contains(&"outputType"),
170            "expected outputType in uri_options, got: {names:?}"
171        );
172        // Verify secret field
173        let db_url_opt = meta
174            .uri_options
175            .iter()
176            .find(|o| o.name == "db_url")
177            .unwrap();
178        assert!(db_url_opt.secret, "db_url must be marked secret");
179    }
180
181    #[test]
182    fn test_config_uri_options_nonempty() {
183        let opts = SqlEndpointConfig::uri_options();
184        assert!(!opts.is_empty(), "uri_options() must be non-empty");
185        let names: Vec<&str> = opts.iter().map(|o| o.name.as_str()).collect();
186        assert!(names.contains(&"db_url"));
187        assert!(names.contains(&"delay"));
188        assert!(names.contains(&"batch"));
189    }
190
191    #[test]
192    fn test_config_metadata_capabilities() {
193        let meta = SqlEndpointConfig::metadata();
194        assert!(meta.capabilities.supports_producer);
195        assert!(meta.capabilities.supports_consumer);
196        assert!(!meta.capabilities.supports_polling_consumer);
197        assert!(!meta.capabilities.supports_streaming);
198    }
199
200    #[test]
201    fn test_component_creates_endpoint() {
202        let c = SqlComponent::new();
203        let ctx = NoOpComponentContext;
204        let ep = c.create_endpoint("sql:select 1?db_url=postgres://localhost/test", &ctx);
205        assert!(ep.is_ok());
206    }
207
208    #[test]
209    fn test_component_rejects_wrong_scheme() {
210        let c = SqlComponent::new();
211        let ctx = NoOpComponentContext;
212        let ep = c.create_endpoint("redis://localhost", &ctx);
213        assert!(ep.is_err());
214    }
215
216    #[test]
217    fn test_endpoint_uri() {
218        let c = SqlComponent::new();
219        let ctx = NoOpComponentContext;
220        let ep = c
221            .create_endpoint("sql:select 1?db_url=postgres://localhost/test", &ctx)
222            .unwrap();
223        assert_eq!(ep.uri(), "sql:select 1?db_url=postgres://localhost/test");
224    }
225
226    #[test]
227    fn test_component_with_global_config() {
228        let global = SqlGlobalConfig::default().with_max_connections(20);
229        let c = SqlComponent::with_config(global);
230        let ctx = NoOpComponentContext;
231        // Verify the component can create endpoints with global config applied
232        assert_eq!(c.scheme(), "sql");
233        let ep = c.create_endpoint("sql:select 1?db_url=postgres://localhost/test", &ctx);
234        assert!(ep.is_ok());
235    }
236
237    #[test]
238    fn test_global_config_applied_to_endpoint() {
239        // Verify that when URI does NOT set pool params, global config fills them in.
240        // Tests the same logic as create_endpoint: from_uri + apply_defaults + resolve_defaults.
241        let global = SqlGlobalConfig::default()
242            .with_max_connections(20)
243            .with_min_connections(3)
244            .with_idle_timeout_secs(600)
245            .with_max_lifetime_secs(3600);
246        let mut cfg =
247            config::SqlEndpointConfig::from_uri("sql:select 1?db_url=postgres://localhost/test")
248                .unwrap();
249        cfg.apply_defaults(&global);
250        cfg.resolve_defaults();
251        assert_eq!(cfg.max_connections, Some(20));
252        assert_eq!(cfg.min_connections, Some(3));
253        assert_eq!(cfg.idle_timeout_secs, Some(600));
254        assert_eq!(cfg.max_lifetime_secs, Some(3600));
255    }
256
257    #[test]
258    fn test_uri_param_wins_over_global_config() {
259        // Verify that URI-set pool params are NOT overridden by global config.
260        let global = SqlGlobalConfig::default()
261            .with_max_connections(20)
262            .with_min_connections(3);
263        let mut cfg = config::SqlEndpointConfig::from_uri(
264            "sql:select 1?db_url=postgres://localhost/test&maxConnections=99&minConnections=7",
265        )
266        .unwrap();
267        cfg.apply_defaults(&global);
268        cfg.resolve_defaults();
269        assert_eq!(cfg.max_connections, Some(99)); // URI wins
270        assert_eq!(cfg.min_connections, Some(7)); // URI wins
271    }
272}
camel_component_sql/lib.rs

camel_component_sql/
lib.rs
