camel_component_sql/
pool_factory.rs

1use std::any::Any;
2use std::sync::Arc;
3use std::time::Duration;
4
5use camel_api::datasource::{CheckFuture, CreatePoolFuture};
6use camel_api::datasource::{DatasourceConfig, DatasourceHandle, PoolFactory};
7use camel_api::error::CamelError;
8use camel_api::lifecycle::HealthStatus;
9use sqlx::AnyPool;
10use sqlx::any::AnyPoolOptions;
11
12use crate::config::{enrich_db_url_with_ssl_params, redact_db_url};
13
14pub struct SqlPoolFactory;
15
16impl PoolFactory for SqlPoolFactory {
17    fn create<'a>(&'a self, config: &'a DatasourceConfig) -> CreatePoolFuture<'a> {
18        Box::pin(async move {
19            // Install all compiled-in sqlx drivers so AnyPool can resolve them.
20            // This is idempotent; safe to call multiple times.
21            sqlx::any::install_default_drivers();
22
23            let max_conn = config.max_connections.unwrap_or(5);
24            let min_conn = config.min_connections.unwrap_or(1);
25            let idle_timeout = Duration::from_secs(config.idle_timeout_secs.unwrap_or(300));
26            let max_lifetime = Duration::from_secs(config.max_lifetime_secs.unwrap_or(1800));
27
28            let db_url = enrich_db_url_with_ssl_params(
29                &config.db_url,
30                config.ssl_mode.as_deref(),
31                config.ssl_root_cert.as_deref(),
32                config.ssl_cert.as_deref(),
33                config.ssl_key.as_deref(),
34            )?;
35
36            let pool = AnyPoolOptions::new()
37                .max_connections(max_conn)
38                .min_connections(min_conn)
39                .idle_timeout(idle_timeout)
40                .max_lifetime(max_lifetime)
41                .connect(&db_url)
42                .await
43                .map_err(|e| {
44                    CamelError::ProcessorError(format!(
45                        "failed to create datasource pool ({}): {}",
46                        redact_db_url(&config.db_url),
47                        e
48                    ))
49                })?;
50
51            tracing::info!("datasource pool created: max_connections={}", max_conn);
52            Ok(Arc::new(pool) as Arc<dyn Any + Send + Sync>)
53        })
54    }
55
56    fn check<'a>(&'a self, handle: &'a DatasourceHandle) -> CheckFuture<'a> {
57        Box::pin(async move {
58            match handle.downcast::<AnyPool>() {
59                Ok(pool) => match sqlx::query("SELECT 1").execute(&*pool).await {
60                    Ok(_) => HealthStatus::Healthy,
61                    Err(e) => {
62                        // log-policy: outside-contract
63                        tracing::warn!("datasource '{}' health check failed: {}", handle.name, e);
64                        HealthStatus::Unhealthy
65                    }
66                },
67                Err(e) => {
68                    // log-policy: outside-contract
69                    tracing::warn!(
70                        "datasource '{}' health check failed: pool downcast error: {}",
71                        handle.name,
72                        e
73                    );
74                    HealthStatus::Unhealthy
75                }
76            }
77        })
78    }
79
80    fn supported_schemes(&self) -> &[&str] {
81        &["postgres", "postgresql", "mysql", "sqlite"]
82    }
83
84    fn name(&self) -> &'static str {
85        "sqlx"
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92
93    #[test]
94    fn sql_pool_factory_name() {
95        let f = SqlPoolFactory;
96        assert_eq!(f.name(), "sqlx");
97    }
98
99    #[test]
100    fn sql_pool_factory_supported_schemes() {
101        let f = SqlPoolFactory;
102        assert!(f.supported_schemes().contains(&"postgres"));
103        assert!(f.supported_schemes().contains(&"mysql"));
104        assert!(f.supported_schemes().contains(&"sqlite"));
105    }
106
107    #[test]
108    fn sql_pool_factory_matches_postgres() {
109        let f = SqlPoolFactory;
110        let cfg = DatasourceConfig {
111            db_url: "postgres://localhost/test".into(),
112            provider: None,
113            max_connections: None,
114            min_connections: None,
115            idle_timeout_secs: None,
116            max_lifetime_secs: None,
117            ssl_mode: None,
118            ssl_root_cert: None,
119            ssl_cert: None,
120            ssl_key: None,
121            extra: std::collections::HashMap::new(),
122        };
123        assert!(f.matches(&cfg));
124    }
125}
camel_component_sql/pool_factory.rs

camel_component_sql/
pool_factory.rs
