welds-connections 0.4.22

use super::Row;
use super::TransactStart;
use super::transaction::{TransT, Transaction};
use super::{Client, Param};
use crate::ExecuteResult;
use crate::errors::Result;
use async_trait::async_trait;
use std::sync::Arc;

#[cfg(feature = "unstable-api")]
use futures_core::stream::BoxStream;

#[cfg(feature = "unstable-api")]
use crate::StreamClient;

#[cfg(feature = "unstable-api")]
mod row_stream;

use tiberius::ToSql;
mod pool;
use pool::Pool;

pub(crate) mod transaction;

#[derive(Clone)]
pub struct MssqlClient {
    pool: Arc<Pool>,
}

#[async_trait]
impl TransactStart for MssqlClient {
    async fn begin<'t>(&'t self) -> Result<Transaction<'t>> {
        let conn = self.pool.get().await?;
        log::debug!("TransactStart: building transaction");
        let trans = transaction::MssqlTransaction::new(conn).await?;
        let inner = TransT::Mssql(trans);
        Ok(Transaction::new(inner))
    }
}

pub async fn connect(cs: &str) -> Result<MssqlClient> {
    let pool = Pool::new(cs);
    Ok(MssqlClient { pool })
}

#[async_trait]
impl Client for MssqlClient {
    async fn execute(&self, sql: &str, params: &[&(dyn Param + Sync)]) -> Result<ExecuteResult> {
        let conn = self.pool.get().await?;
        conn.execute(sql, params).await
    }

    async fn fetch_rows(&self, sql: &str, params: &[&(dyn Param + Sync)]) -> Result<Vec<Row>> {
        let conn = self.pool.get().await?;
        conn.fetch_rows(sql, params).await
    }

    async fn fetch_many<'s, 'args, 't>(
        &self,
        args: &[crate::Fetch<'s, 'args, 't>],
    ) -> Result<Vec<Vec<Row>>> {
        let conn = self.pool.get().await?;
        conn.fetch_many(args).await
    }

    fn syntax(&self) -> crate::Syntax {
        crate::Syntax::Mssql
    }
}

#[cfg(feature = "unstable-api")]
#[async_trait]
impl StreamClient for MssqlClient {
    /// Run the SQL streaming the results back in a future::stream
    async fn stream<'client, 'e, 'params>(
        &'client self,
        sql: &str,
        params: &[&'params (dyn Param + Sync)],
    ) -> BoxStream<'e, Result<Row>>
    where
        'client: 'e,
        'params: 'e,
    {
        use futures::StreamExt;
        let conn = self.pool.get().await;
        let conn = match conn {
            Ok(c) => c,
            Err(err) => return futures::stream::iter([Err(err)]).boxed(),
        };
        row_stream::MssqlClientStream::new(conn, sql, params)
            .await
            .boxed()
    }
}

pub trait MssqlParam {
    fn add_param<'a>(&'a self, args: Vec<&'a dyn ToSql>) -> Vec<&'a dyn ToSql>;
}

impl<T> MssqlParam for T
where
    T: 'static + ToSql,
{
    fn add_param<'a>(&'a self, mut args: Vec<&'a dyn ToSql>) -> Vec<&'a dyn ToSql> {
        args.push(self);
        args
    }
}