feat(wip)!: Initial implementation of a real connection pooling

Author: TheRustifyer

canyon_core/Cargo.toml

@@ -27,6 +27,9 @@ toml = { workspace = true }
 serde = { workspace = true }
 walkdir = { workspace = true }
 cfg-if = "1.0.0"
+bb8-postgres = "0.9.0"
+bb8-tiberius = "0.16.0"
+bb8 = "0.9.0"
 
 [features]
 postgres = ["tokio-postgres"]

canyon_core/src/canyon.rs

@@ -2,13 +2,14 @@ use crate::connection::conn_errors::DatasourceNotFound;
 use crate::connection::database_type::DatabaseType;
 use crate::connection::datasources::{CanyonSqlConfig, DatasourceConfig, Datasources};
 use crate::connection::{
-    CANYON_INSTANCE, db_connector, get_canyon_tokio_runtime, pool::get_pool_manager,
+    CANYON_INSTANCE, db_connector, get_canyon_tokio_runtime,
 };
 use db_connector::DatabaseConnection;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::{error::Error, fs};
 use tokio::sync::Mutex;
+use crate::connection::pool::CanyonConnection;
 
 pub type SharedConnection = Arc<Mutex<DatabaseConnection>>;
 
@@ -53,14 +54,14 @@ pub type SharedConnection = Arc<Mutex<DatabaseConnection>>;
 /// - `find_datasource_by_name_or_default`: Finds a datasource by name or returns the default.
 /// - `get_connection`: Retrieves a read-only connection from the cache.
 /// - `get_mut_connection`: Retrieves a mutable connection from the cache.
-pub struct Canyon {
+pub struct Canyon<'a> {
     config: Datasources,
-    connections: HashMap<&'static str, DatabaseConnection>,
+    connections: HashMap<&'static str, CanyonConnection<'a>>,
     default_connection: Option<DatabaseConnection>,
     default_db_type: Option<DatabaseType>,
 }
 
-impl Canyon {
+impl<'a> Canyon<'a> {
     /// Returns the global singleton instance of `Canyon`.
     ///
     /// This function allows access to the singleton instance of the Canyon engine
@@ -114,8 +115,8 @@ impl Canyon {
         let config_content = fs::read_to_string(&path)?;
         let config: Datasources = toml::from_str::<CanyonSqlConfig>(&config_content)?.canyon_sql;
 
-        let mut connections: HashMap<&str, DatabaseConnection> = HashMap::new();
-        let mut default_connection: Option<DatabaseConnection> = None;
+        let mut connections: HashMap<&str, CanyonConnection<'a>> = HashMap::new();
+        let mut default_connection: Option<CanyonConnection<'a>> = None;
         let mut default_db_type: Option<DatabaseType> = None;
 
         for ds in config.datasources.iter() {
@@ -204,33 +205,6 @@ impl Canyon {
         Ok(conn)
     }
 
-    /// Gets a pooled connection for better performance
-    /// This is an internal method that uses the connection pool
-    pub async fn get_pooled_connection(
-        &self,
-        name: &str,
-    ) -> Result<crate::connection::pool::PooledConnection, DatasourceNotFound> {
-        let pool_manager = get_pool_manager();
-        let mut pool_manager_guard = pool_manager.lock().await;
-
-        // Find the datasource
-        let datasource = self.find_datasource_by_name_or_default(name)?;
-
-        // Create pool if it doesn't exist
-        if !pool_manager_guard.has_pool(name) {
-            pool_manager_guard
-                .create_pool(name, datasource)
-                .await
-                .map_err(|_| DatasourceNotFound::from(Some(name)))?;
-        }
-
-        // Get pooled connection
-        pool_manager_guard
-            .get_connection(name)
-            .await
-            .map_err(|_| DatasourceNotFound::from(Some(name)))
-    }
-
     /// Gets a fast connection that automatically uses pooling when available
     /// This method provides the best performance by using connection pooling
     pub async fn get_fast_connection(
@@ -251,6 +225,7 @@ mod __impl {
     use std::error::Error;
     use std::path::PathBuf;
     use walkdir::WalkDir;
+    use crate::connection::pool::CanyonConnection;
 
     // Internal helper to locate the config file
     pub(crate) fn find_config_path() -> Result<PathBuf, std::io::Error> {
@@ -274,10 +249,10 @@ mod __impl {
             })
     }
 
-    pub(crate) async fn process_new_conn_by_datasource(
+    pub(crate) async fn process_new_conn_by_datasource<'a>(
         ds: &DatasourceConfig,
-        connections: &mut HashMap<&str, DatabaseConnection>,
-        default: &mut Option<DatabaseConnection>,
+        connections: &mut HashMap<&str, CanyonConnection<'a>>,
+        default: &mut Option<CanyonConnection<'a>>,
         default_db_type: &mut Option<DatabaseType>,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
         if default.is_none() {

canyon_core/src/connection/db_connector.rs

@@ -46,29 +46,6 @@ impl DatabaseConnection {
         }
     }
 
-    /// Creates a connection with optimized settings for better performance
-    pub async fn new_optimized(
-        datasource: &DatasourceConfig,
-    ) -> Result<DatabaseConnection, Box<dyn Error + Send + Sync>> {
-        // Use optimized connection settings for better performance
-        match datasource.get_db_type() {
-            #[cfg(feature = "postgres")]
-            DatabaseType::PostgreSql => {
-                connection_helpers::create_postgres_connection_optimized(datasource).await
-            }
-
-            #[cfg(feature = "mssql")]
-            DatabaseType::SqlServer => {
-                connection_helpers::create_sqlserver_connection_optimized(datasource).await
-            }
-
-            #[cfg(feature = "mysql")]
-            DatabaseType::MySQL => {
-                connection_helpers::create_mysql_connection_optimized(datasource).await
-            }
-        }
-    }
-
     pub fn get_db_type(&self) -> DatabaseType {
         match self {
             #[cfg(feature = "postgres")]
@@ -116,28 +93,6 @@ mod connection_helpers {
         datasource: &DatasourceConfig,
     ) -> Result<DatabaseConnection, Box<dyn Error + Send + Sync>> {
         let (user, password) = auth::extract_postgres_auth(&datasource.auth)?;
-        let url = connection_string(user, password, datasource);
-
-        let (client, connection) = tokio_postgres::connect(&url, tokio_postgres::NoTls).await?;
-
-        tokio::spawn(async move {
-            if let Err(e) = connection.await {
-                eprintln!(
-                    "An error occurred while trying to connect to the PostgreSQL database: {e}"
-                );
-            }
-        });
-
-        Ok(DatabaseConnection::Postgres(PostgreSqlConnection {
-            client,
-        }))
-    }
-
-    #[cfg(feature = "postgres")]
-    pub async fn create_postgres_connection_optimized(
-        datasource: &DatasourceConfig,
-    ) -> Result<DatabaseConnection, Box<dyn Error + Send + Sync>> {
-        let (user, password) = auth::extract_postgres_auth(&datasource.auth)?;
 
         // Use optimized connection settings
         let mut config = tokio_postgres::Config::new();
@@ -196,62 +151,18 @@ mod connection_helpers {
         }))
     }
 
-    #[cfg(feature = "mssql")]
-    pub async fn create_sqlserver_connection_optimized(
-        datasource: &DatasourceConfig,
-    ) -> Result<DatabaseConnection, Box<dyn Error + Send + Sync>> {
-        use async_std::net::TcpStream;
-        let mut tiberius_config = tiberius::Config::new();
-
-        tiberius_config.host(&datasource.properties.host);
-        tiberius_config.port(datasource.properties.port.unwrap_or_default());
-        tiberius_config.database(&datasource.properties.db_name);
-
-        let auth_config = auth::extract_mssql_auth(&datasource.auth)?;
-        tiberius_config.authentication(auth_config);
-        tiberius_config.trust_cert(); // TODO: this should be specifically set via user input
-        tiberius_config.encryption(tiberius::EncryptionLevel::NotSupported); // TODO: user input
-
-        // Optimize connection settings for better performance
-        // Note: Tiberius doesn't expose these settings directly
-        // The optimization is handled at the TCP level
-
-        let tcp = TcpStream::connect(tiberius_config.get_addr()).await?;
-        tcp.set_nodelay(true)?;
-
-        let client = tiberius::Client::connect(tiberius_config, tcp).await?;
-
-        Ok(DatabaseConnection::SqlServer(SqlServerConnection {
-            client: Box::leak(Box::new(client)),
-        }))
-    }
-
     #[cfg(feature = "mysql")]
     pub async fn create_mysql_connection(
         datasource: &DatasourceConfig,
     ) -> Result<DatabaseConnection, Box<dyn Error + Send + Sync>> {
         use mysql_async::Pool;
 
-        let (user, password) = auth::extract_mysql_auth(&datasource.auth)?;
-        let url = connection_string(user, password, datasource);
-        let mysql_connection = Pool::from_url(url)?;
-
-        Ok(DatabaseConnection::MySQL(MysqlConnection {
-            client: mysql_connection,
-        }))
-    }
-
-    #[cfg(feature = "mysql")]
-    pub async fn create_mysql_connection_optimized(
-        datasource: &DatasourceConfig,
-    ) -> Result<DatabaseConnection, Box<dyn Error + Send + Sync>> {
-        use mysql_async::Pool;
-
         let (user, password) = auth::extract_mysql_auth(&datasource.auth)?;
         let url = connection_string(user, password, datasource);
 
         // Use optimized pool settings for better performance
-        let _pool_constraints = mysql_async::PoolConstraints::new(2, 10).unwrap();
+        let _pool_constraints = mysql_async::PoolConstraints::new(2, 10)
+            .ok_or_else(|| "Failure launching the MySQL pool")?;
 
         let mysql_connection = Pool::from_url(url)?;