feat: connection pooling for all the supported databases

Author: TheRustifyer

canyon_core/src/canyon.rs

@@ -1,15 +1,12 @@
 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};
 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>>;
 
@@ -54,14 +51,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<'a> {
+pub struct Canyon {
     config: Datasources,
-    connections: HashMap<&'static str, CanyonConnection<'a>>,
+    connections: HashMap<&'static str, DatabaseConnection>,
     default_connection: Option<DatabaseConnection>,
     default_db_type: Option<DatabaseType>,
 }
 
-impl<'a> Canyon<'a> {
+impl Canyon {
     /// Returns the global singleton instance of `Canyon`.
     ///
     /// This function allows access to the singleton instance of the Canyon engine
@@ -115,8 +112,8 @@ impl<'a> Canyon<'a> {
         let config_content = fs::read_to_string(&path)?;
         let config: Datasources = toml::from_str::<CanyonSqlConfig>(&config_content)?.canyon_sql;
 
-        let mut connections: HashMap<&str, CanyonConnection<'a>> = HashMap::new();
-        let mut default_connection: Option<CanyonConnection<'a>> = 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() {
@@ -248,7 +245,7 @@ mod __impl {
             })
     }
 
-    pub(crate) async fn process_new_conn_by_datasource<'a>(
+    pub(crate) async fn process_new_conn_by_datasource(
         ds: &DatasourceConfig,
         connections: &mut HashMap<&str, DatabaseConnection>,
         default: &mut Option<DatabaseConnection>,

canyon_core/src/connection/clients/mssql.rs

@@ -1,130 +1,46 @@
-use crate::{query::parameters::QueryParameter, rows::CanyonRows};
-#[cfg(feature = "mssql")]
-use async_std::net::TcpStream;
+use crate::connection::{MsManager, SqlServerConnectionPool};
+use crate::query::parameters::QueryParameter;
+use bb8::PooledConnection;
 use std::error::Error;
+#[cfg(feature = "mssql")]
 use tiberius::Query;
 
 /// A connection with a `SqlServer` database
-#[cfg(feature = "mssql")]
-pub struct SqlServerConnection {
-    pub client: &'static mut tiberius::Client<TcpStream>,
+#[cfg(feature = "mssql")] // TODO: remove the local cfg and put them at module level
+pub struct SqlServerConnection(SqlServerConnectionPool);
+
+impl SqlServerConnection {
+    pub fn new(pool: SqlServerConnectionPool) -> Result<Self, Box<dyn Error + Send + Sync>> {
+        Ok(Self(pool))
+    }
+    pub async fn get_pooled(
+        &self,
+    ) -> Result<PooledConnection<'_, MsManager>, Box<dyn Error + Send + Sync>> {
+        Ok(self.0.get().await?)
+    }
 }
 
 #[cfg(feature = "mssql")]
 pub(crate) mod sqlserver_query_launcher {
     use super::*;
-    use crate::mapper::RowMapper;
-    use crate::rows::FromSqlOwnedValue;
     use tiberius::QueryStream;
 
-    #[inline(always)]
-    pub(crate) async fn query<S, R>(
-        stmt: S,
-        params: &[&'_ dyn QueryParameter],
-        conn: &SqlServerConnection,
-    ) -> Result<Vec<R>, Box<dyn Error + Send + Sync>>
-    where
-        S: AsRef<str> + Send,
-        R: RowMapper,
-        Vec<R>: FromIterator<<R as RowMapper>::Output>,
-    {
-        Ok(execute_query(stmt.as_ref(), params, conn)
-            .await?
-            .into_results()
-            .await?
-            .into_iter()
-            .flatten()
-            .flat_map(|row| R::deserialize_sqlserver(&row))
-            .collect::<Vec<R>>())
-    }
-
-    #[inline(always)]
-    pub(crate) async fn query_rows(
-        stmt: &str,
-        params: &[&'_ dyn QueryParameter],
-        conn: &SqlServerConnection,
-    ) -> Result<CanyonRows, Box<dyn Error + Send + Sync>> {
-        let result = execute_query(stmt, params, conn)
-            .await?
-            .into_results()
-            .await?
-            .into_iter()
-            .flatten()
-            .collect();
-
-        Ok(CanyonRows::Tiberius(result))
-    }
-
-    pub(crate) async fn query_one<R>(
-        stmt: &str,
-        params: &[&'_ dyn QueryParameter],
-        conn: &SqlServerConnection,
-    ) -> Result<Option<R::Output>, Box<dyn Error + Send + Sync>>
-    where
-        R: RowMapper,
-    {
-        let result = execute_query(stmt, params, conn).await?.into_row().await?;
-
-        match result {
-            Some(r) => Ok(Some(R::deserialize_sqlserver(&r)?)),
-            None => Ok(None),
-        }
-    }
-
-    pub(crate) async fn query_one_for<T: FromSqlOwnedValue<T>>(
-        stmt: &str,
-        params: &[&'_ dyn QueryParameter],
-        conn: &SqlServerConnection,
-    ) -> Result<T, Box<dyn Error + Send + Sync>> {
-        let row = execute_query(stmt, params, conn)
-            .await?
-            .into_row()
-            .await?
-            .ok_or_else(|| format!("Failure executing 'query_one_for' while retrieving the first row with stmt: {:?}", stmt))?;
-
-        Ok(row
-            .into_iter()
-            .map(T::from_sql_owned)
-            .collect::<Vec<_>>()
-            .remove(0)?
-            .ok_or_else(|| format!("Failure executing 'query_one_for' while retrieving the first column value on the first row with stmt: {:?}", stmt))?
-        )
-    }
-
-    pub(crate) async fn execute(
+    pub(crate) async fn execute_query<'a>(
         stmt: &str,
-        params: &[&'_ dyn QueryParameter],
-        conn: &SqlServerConnection,
-    ) -> Result<u64, Box<dyn Error + Send + Sync>> {
-        let mssql_query = generate_mssql_stmt(stmt, params).await;
-
-        #[allow(mutable_transmutes)] // TODO: pls solve this elegantly someday :(
-        let sqlservconn =
-            unsafe { std::mem::transmute::<&SqlServerConnection, &mut SqlServerConnection>(conn) };
-
-        mssql_query
-            .execute(sqlservconn.client)
-            .await
-            .map(|r| r.total())
-            .map_err(From::from)
+        params: &[&dyn QueryParameter],
+        conn: &'a mut bb8::PooledConnection<'_, bb8_tiberius::ConnectionManager>,
+    ) -> Result<QueryStream<'a>, Box<dyn Error + Send + Sync>> {
+        let mssql_query = generate_mssql_query_client(stmt, params).await;
+        mssql_query.query(conn).await.map_err(From::from)
     }
 
-    async fn execute_query<'a>(
+    pub(crate) async fn generate_mssql_query_client<'a>(
         stmt: &str,
         params: &[&'a dyn QueryParameter],
-        conn: &SqlServerConnection,
-    ) -> Result<QueryStream<'a>, Box<dyn Error + Send + Sync>> {
-        let mssql_query = generate_mssql_stmt(stmt, params).await;
-
-        #[allow(mutable_transmutes)] // TODO: pls solve this elegantly someday :(
-        let sqlservconn =
-            unsafe { std::mem::transmute::<&SqlServerConnection, &mut SqlServerConnection>(conn) };
-        Ok(mssql_query.query(sqlservconn.client).await?)
-    }
-
-    async fn generate_mssql_stmt<'a>(stmt: &str, params: &[&'a dyn QueryParameter]) -> Query<'a> {
+    ) -> Query<'a> {
         let mut stmt = String::from(stmt);
         if stmt.contains("RETURNING") {
+            // TODO: when the InsertQuerybuilder with a api on the builder for the returning clause
             let c = stmt.clone();
             let temp = c.split_once("RETURNING").unwrap();
             let temp2 = temp.0.split_once("VALUES").unwrap();
@@ -137,13 +53,19 @@ pub(crate) mod sqlserver_query_launcher {
             );
         }
 
-        // TODO: We must address the query generation
-        // NOTE: ready to apply the change now that the querybuilder knows what's the underlying db type
-        let mut mssql_query = Query::new(stmt.to_owned().replace('$', "@P"));
+        let stmt = stmt.replace('$', "@P"); // TODO: this should be solved by the querybuilder
+        generate_query_and_bind_params(stmt, params)
+    }
+
+    // Query and parameters are generated in this procedure together to avoid lifetime errors
+    fn generate_query_and_bind_params<'a>(
+        stmt: String,
+        params: &[&'a (dyn QueryParameter + 'a)],
+    ) -> Query<'a> {
+        let mut mssql_query = Query::new(stmt);
         params.iter().for_each(|param| {
             mssql_query.bind(*param);
         });
-
         mssql_query
     }
 }

canyon_core/src/connection/clients/mysql.rs

@@ -10,8 +10,12 @@ use std::error::Error;
 
 /// A connection with a `Mysql` database
 #[cfg(feature = "mysql")]
-pub struct MysqlConnection {
-    pub client: Pool,
+pub struct MySQLConnector(mysql_async::Pool);
+
+impl MySQLConnector {
+    pub fn new(pool: Pool) -> Self {
+        Self(pool)
+    }
 }
 
 #[cfg(feature = "mysql")]
@@ -33,7 +37,7 @@ pub(crate) mod mysql_query_launcher {
     pub async fn query<S, R>(
         stmt: S,
         params: &[&'_ dyn QueryParameter],
-        conn: &MysqlConnection,
+        conn: &MySQLConnector,
     ) -> Result<Vec<R>, Box<dyn Error + Send + Sync>>
     where
         S: AsRef<str> + Send,
@@ -51,7 +55,7 @@ pub(crate) mod mysql_query_launcher {
     pub(crate) async fn query_rows(
         stmt: &str,
         params: &[&'_ dyn QueryParameter],
-        conn: &MysqlConnection,
+        conn: &MySQLConnector,
     ) -> Result<CanyonRows, Box<dyn Error + Send + Sync>> {
         Ok(CanyonRows::MySQL(execute_query(stmt, params, conn).await?))
     }
@@ -60,7 +64,7 @@ pub(crate) mod mysql_query_launcher {
     pub(crate) async fn query_one<R>(
         stmt: &str,
         params: &[&'_ dyn QueryParameter],
-        conn: &MysqlConnection,
+        conn: &MySQLConnector,
     ) -> Result<Option<R::Output>, Box<dyn Error + Send + Sync>>
     where
         R: RowMapper,
@@ -77,7 +81,7 @@ pub(crate) mod mysql_query_launcher {
     pub(crate) async fn query_one_for<T: FromSqlOwnedValue<T>>(
         stmt: &str,
         params: &[&'_ dyn QueryParameter],
-        conn: &MysqlConnection,
+        conn: &MySQLConnector,
     ) -> Result<T, Box<dyn Error + Send + Sync>> {
         Ok(execute_query(stmt, params, conn)
             .await?
@@ -92,12 +96,12 @@ pub(crate) mod mysql_query_launcher {
     async fn execute_query<S>(
         stmt: S,
         params: &[&'_ dyn QueryParameter],
-        conn: &MysqlConnection,
+        conn: &MySQLConnector,
     ) -> Result<Vec<Row>, Box<dyn Error + Send + Sync>>
     where
         S: AsRef<str> + Send,
     {
-        let mysql_connection = conn.client.get_conn().await?;
+        let mysql_connection = conn.0.get_conn().await?;
         let is_insert = stmt.as_ref().find(" RETURNING");
         let mysql_stmt = generate_mysql_stmt(stmt.as_ref(), params)?;
 
@@ -122,12 +126,12 @@ pub(crate) mod mysql_query_launcher {
     pub(crate) async fn execute<S>(
         stmt: S,
         params: &[&'_ dyn QueryParameter],
-        conn: &MysqlConnection,
+        conn: &MySQLConnector,
     ) -> Result<u64, Box<dyn Error + Send + Sync>>
     where
         S: AsRef<str> + Send,
     {
-        let mysql_connection = conn.client.get_conn().await?;
+        let mysql_connection = conn.0.get_conn().await?;
         let mysql_stmt = generate_mysql_stmt(stmt.as_ref(), params)?;
 
         Ok(mysql_stmt.run(mysql_connection).await?.affected_rows())