Merge branch 'refactor/GH-65-general-refactor-sync-mutex' into refactor/GH-65-general-refactor

Author: TheRustifyer

.DS_Store

@@ -19,7 +19,7 @@ chrono = { workspace = true }
 async-std = { workspace = true, optional = true }
 regex = { workspace = true }
 
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["sync"] }
 tokio-util = { workspace = true }
 
 futures = { workspace = true }

canyon_core/Cargo.toml

@@ -1,7 +1,7 @@
 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};
+use crate::connection::{CANYON_INSTANCE, db_connector, get_canyon_tokio_runtime, pool::get_pool_manager};
 use db_connector::DatabaseConnection;
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -53,8 +53,8 @@ pub type SharedConnection = Arc<Mutex<DatabaseConnection>>;
 /// - `get_mut_connection`: Retrieves a mutable connection from the cache.
 pub struct Canyon {
     config: Datasources,
-    connections: HashMap<&'static str, SharedConnection>,
-    default_connection: Option<SharedConnection>,
+    connections: HashMap<&'static str, DatabaseConnection>,
+    default_connection: Option<DatabaseConnection>,
     default_db_type: Option<DatabaseType>,
 }
 
@@ -113,9 +113,9 @@ 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::new();
-        let mut default_connection = None;
-        let mut default_db_type = None;
+        let mut connections: HashMap<&str, DatabaseConnection> = HashMap::new();
+        let mut default_connection: Option<DatabaseConnection> = None;
+        let mut default_db_type: Option<DatabaseType> = None;
 
         for ds in config.datasources.iter() {
             __impl::process_new_conn_by_datasource(
@@ -179,9 +179,9 @@ impl Canyon {
     }
 
     // Retrieve a read-only connection from the cache
-    pub fn get_default_connection(&self) -> Result<SharedConnection, DatasourceNotFound> {
+    pub fn get_default_connection(&self) -> Result<&DatabaseConnection, DatasourceNotFound> {
         self.default_connection
-            .clone()
+            .as_ref()
             .ok_or_else(|| DatasourceNotFound::from(None))
     }
 
@@ -190,7 +190,7 @@ impl Canyon {
     /// This is a fast and efficient operation: cloning the [`SharedConnection`]
     /// simply increases the reference count [`Arc`] without duplicating the underlying
     /// [`DatabaseConnection`]. Returns an error if no default connection is configured.
-    pub fn get_connection(&self, name: &str) -> Result<SharedConnection, DatasourceNotFound> {
+    pub fn get_connection(&self, name: &str) -> Result<&DatabaseConnection, DatasourceNotFound> {
         if name.is_empty() {
             return self.get_default_connection();
         }
@@ -200,20 +200,47 @@ impl Canyon {
             .get(name)
             .ok_or_else(|| DatasourceNotFound::from(Some(name)))?;
 
-        Ok(conn.clone())
+        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(&self, name: &str) -> Result<&DatabaseConnection, DatasourceNotFound> {
+        // For now, fall back to the regular connection
+        // In the future, this could automatically use the pool
+        self.get_connection(name)
+    }
+
+
 }
 
 mod __impl {
-    use crate::canyon::SharedConnection;
     use crate::connection::database_type::DatabaseType;
     use crate::connection::datasources::DatasourceConfig;
     use crate::connection::db_connector::DatabaseConnection;
     use std::collections::HashMap;
     use std::error::Error;
     use std::path::PathBuf;
-    use std::sync::Arc;
-    use tokio::sync::Mutex;
     use walkdir::WalkDir;
 
     // Internal helper to locate the config file
@@ -240,23 +267,22 @@ mod __impl {
 
     pub(crate) async fn process_new_conn_by_datasource(
         ds: &DatasourceConfig,
-        connections: &mut HashMap<&str, SharedConnection>,
-        default: &mut Option<SharedConnection>,
+        connections: &mut HashMap<&str, DatabaseConnection>,
+        default: &mut Option<DatabaseConnection>,
         default_db_type: &mut Option<DatabaseType>,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
+        if default.is_none() {
+            let cloned_ds_for_default = ds.clone();
+            *default = Some(DatabaseConnection::new(&cloned_ds_for_default).await?); // Only cloning the smart pointer
+        }
         let conn = DatabaseConnection::new(ds).await?;
         let name: &'static str = Box::leak(ds.name.clone().into_boxed_str());
 
         if default_db_type.is_none() {
             *default_db_type = Some(conn.get_db_type());
         }
 
-        let connection_sp = Arc::new(Mutex::new(conn));
-
-        if default.is_none() {
-            *default = Some(connection_sp.clone()); // Only cloning the smart pointer
-        }
-
+        let connection_sp = conn;
         connections.insert(name, connection_sp);
 
         Ok(())

canyon_core/src/canyon.rs

@@ -62,6 +62,60 @@ impl DbConnection for DatabaseConnection {
     }
 }
 
+impl DbConnection for &DatabaseConnection {
+    async fn query_rows<'a>(
+        &self,
+        stmt: &str,
+        params: &[&'a dyn QueryParameter],
+    ) -> Result<CanyonRows, Box<(dyn Error + Send + Sync)>> {
+        db_conn_query_rows_impl(self, stmt, params).await
+    }
+
+    async fn query<'a, S, R>(
+        &self,
+        stmt: S,
+        params: &[&'a (dyn QueryParameter)],
+    ) -> Result<Vec<R>, Box<(dyn Error + Send + Sync)>>
+    where
+        S: AsRef<str> + Send,
+        R: RowMapper,
+        Vec<R>: FromIterator<<R as RowMapper>::Output>,
+    {
+        db_conn_query_impl(self, stmt, params).await
+    }
+
+    async fn query_one<'a, R>(
+        &self,
+        stmt: &str,
+        params: &[&'a dyn QueryParameter],
+    ) -> Result<Option<R::Output>, Box<(dyn Error + Send + Sync)>>
+    where
+        R: RowMapper,
+    {
+        db_conn_query_one_impl::<R>(self, stmt, params).await
+    }
+
+    async fn query_one_for<'a, T: FromSqlOwnedValue<T>>(
+        &self,
+        stmt: &str,
+        params: &[&'a dyn QueryParameter],
+    ) -> Result<T, Box<(dyn Error + Send + Sync)>> {
+        db_conn_query_one_for_impl::<T>(self, stmt, params).await
+    }
+
+    async fn execute<'a>(
+        &self,
+        stmt: &str,
+        params: &[&'a dyn QueryParameter],
+    ) -> Result<u64, Box<(dyn Error + Send + Sync)>> {
+        db_conn_execute_impl(self, stmt, params).await
+    }
+
+    fn get_database_type(&self) -> Result<DatabaseType, Box<(dyn Error + Send + Sync)>> {
+        Ok(self.get_db_type())
+    }
+}
+
 impl DbConnection for &mut DatabaseConnection {
     async fn query_rows<'a>(
         &self,

canyon_core/src/connection/contracts/impl/database_connection.rs

@@ -11,18 +11,18 @@ use crate::{
 use std::{error::Error, future::Future};
 
 impl DbConnection for SqlServerConnection {
-    fn query_rows<'a>(
+    fn query_rows(
         &self,
         stmt: &str,
-        params: &[&'a dyn QueryParameter],
+        params: &[&dyn QueryParameter],
     ) -> impl Future<Output = Result<CanyonRows, Box<(dyn Error + Send + Sync)>>> + Send {
         sqlserver_query_launcher::query_rows(stmt, params, self)
     }
 
-    fn query<'a, S, R>(
+    fn query<S, R>(
         &self,
         stmt: S,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<Vec<R>, Box<(dyn Error + Send + Sync)>>> + Send
     where
         S: AsRef<str> + Send,
@@ -32,29 +32,29 @@ impl DbConnection for SqlServerConnection {
         sqlserver_query_launcher::query(stmt, params, self)
     }
 
-    fn query_one<'a, R>(
+    fn query_one<R>(
         &self,
         stmt: &str,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<Option<R::Output>, Box<(dyn Error + Send + Sync)>>> + Send
     where
         R: RowMapper,
     {
         sqlserver_query_launcher::query_one::<R>(stmt, params, self)
     }
 
-    fn query_one_for<'a, T: FromSqlOwnedValue<T>>(
+    fn query_one_for<T: FromSqlOwnedValue<T>>(
         &self,
         stmt: &str,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<T, Box<(dyn Error + Send + Sync)>>> + Send {
         sqlserver_query_launcher::query_one_for(stmt, params, self)
     }
 
-    fn execute<'a>(
+    fn execute(
         &self,
         stmt: &str,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<u64, Box<(dyn Error + Send + Sync)>>> + Send {
         sqlserver_query_launcher::execute(stmt, params, self)
     }

canyon_core/src/connection/contracts/impl/mssql.rs

@@ -10,18 +10,18 @@ use crate::{
 use std::{error::Error, future::Future};
 
 impl DbConnection for MysqlConnection {
-    fn query_rows<'a>(
+    fn query_rows(
         &self,
         stmt: &str,
-        params: &[&'a dyn QueryParameter],
+        params: &[&dyn QueryParameter],
     ) -> impl Future<Output = Result<CanyonRows, Box<(dyn Error + Send + Sync)>>> + Send {
         mysql_query_launcher::query_rows(stmt, params, self)
     }
 
-    fn query<'a, S, R>(
+    fn query<S, R>(
         &self,
         stmt: S,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<Vec<R>, Box<(dyn Error + Send + Sync)>>> + Send
     where
         S: AsRef<str> + Send,
@@ -31,29 +31,29 @@ impl DbConnection for MysqlConnection {
         mysql_query_launcher::query(stmt, params, self)
     }
 
-    fn query_one<'a, R>(
+    fn query_one<R>(
         &self,
         stmt: &str,
-        params: &[&'a dyn QueryParameter],
+        params: &[&dyn QueryParameter],
     ) -> impl Future<Output = Result<Option<R::Output>, Box<(dyn Error + Send + Sync)>>> + Send
     where
         R: RowMapper,
     {
         mysql_query_launcher::query_one::<R>(stmt, params, self)
     }
 
-    fn query_one_for<'a, T: FromSqlOwnedValue<T>>(
+    fn query_one_for<T: FromSqlOwnedValue<T>>(
         &self,
         stmt: &str,
-        params: &[&'a dyn QueryParameter],
+        params: &[&dyn QueryParameter],
     ) -> impl Future<Output = Result<T, Box<(dyn Error + Send + Sync)>>> + Send {
         mysql_query_launcher::query_one_for(stmt, params, self)
     }
 
-    fn execute<'a>(
+    fn execute(
         &self,
         stmt: &str,
-        params: &[&'a dyn QueryParameter],
+        params: &[&dyn QueryParameter],
     ) -> impl Future<Output = Result<u64, Box<(dyn Error + Send + Sync)>>> + Send {
         mysql_query_launcher::execute(stmt, params, self)
     }

canyon_core/src/connection/contracts/impl/mysql.rs

@@ -11,18 +11,18 @@ use crate::{
 use std::{error::Error, future::Future};
 
 impl DbConnection for PostgreSqlConnection {
-    fn query_rows<'a>(
+    fn query_rows(
         &self,
         stmt: &str,
-        params: &[&'a dyn QueryParameter],
+        params: &[&dyn QueryParameter],
     ) -> impl Future<Output = Result<CanyonRows, Box<(dyn Error + Send + Sync)>>> + Send {
         postgres_query_launcher::query_rows(stmt, params, self)
     }
 
-    fn query<'a, S, R>(
+    fn query<S, R>(
         &self,
         stmt: S,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<Vec<R>, Box<(dyn Error + Send + Sync)>>> + Send
     where
         S: AsRef<str> + Send,
@@ -32,29 +32,29 @@ impl DbConnection for PostgreSqlConnection {
         postgres_query_launcher::query(stmt, params, self)
     }
 
-    fn query_one<'a, R>(
+    fn query_one<R>(
         &self,
         stmt: &str,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<Option<R::Output>, Box<(dyn Error + Send + Sync)>>> + Send
     where
         R: RowMapper,
     {
         postgres_query_launcher::query_one::<R>(stmt, params, self)
     }
 
-    fn query_one_for<'a, T: FromSqlOwnedValue<T>>(
+    fn query_one_for<T: FromSqlOwnedValue<T>>(
         &self,
         stmt: &str,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<T, Box<(dyn Error + Send + Sync)>>> + Send {
         postgres_query_launcher::query_one_for(stmt, params, self)
     }
 
-    fn execute<'a>(
+    fn execute(
         &self,
         stmt: &str,
-        params: &[&'a (dyn QueryParameter)],
+        params: &[&(dyn QueryParameter)],
     ) -> impl Future<Output = Result<u64, Box<(dyn Error + Send + Sync)>>> + Send {
         postgres_query_launcher::execute(stmt, params, self)
     }