welds-connections 0.4.11

An async ORM for (postgres, mssql, mysql, sqlite)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

use super::super::MssqlParam;
use super::ConnectionStatus;
use super::TiberiusConn;
use super::{Client, Param};
use crate::errors::Result;
use crate::row::Row;
use crate::ExecuteResult;
use async_mutex::Mutex as AsyncMutex;
use async_trait::async_trait;
use std::sync::mpsc::Sender;
use tiberius::ToSql;

pub struct PooledConnection {
    pub(crate) status: ConnectionStatus,
    // NOTE: this is an option so it can be taken when dropped
    pub(super) tiberius_conn: AsyncMutex<Option<TiberiusConn>>,
    pub(super) conn_return: Sender<(TiberiusConn, ConnectionStatus)>,
}

impl PooledConnection {
    pub(crate) async fn simple_query(&self, query: &str) -> Result<()> {
        let mut guard = self.tiberius_conn.lock().await;
        let conn: &mut TiberiusConn = guard.as_mut().unwrap();
        conn.simple_query(query).await?;
        Ok(())
    }

    /// returns the number of open transactions for this connection
    pub(crate) async fn transaction_count(&self) -> Result<i32> {
        let mut guard = self.tiberius_conn.lock().await;
        let conn: &mut TiberiusConn = guard.as_mut().unwrap();
        let sql = "SELECT @@TRANCOUNT AS OpenTransactions";
        let set = conn.simple_query(sql).await?;
        let row = set.into_row().await?.unwrap();
        Ok(row.get("OpenTransactions").unwrap())
    }
}

impl Drop for PooledConnection {
    fn drop(&mut self) {
        // take the conn from ourself
        let guard = self.tiberius_conn.get_mut();
        let mut conn: Option<TiberiusConn> = None;
        std::mem::swap(&mut conn, guard);
        let conn = conn.unwrap();
        self.conn_return
            .send((conn, self.status.clone()))
            .expect("Unable to return connection to pool");
    }
}

#[async_trait]
impl Client for PooledConnection {
    async fn execute(&self, sql: &str, params: &[&(dyn Param + Sync)]) -> Result<ExecuteResult> {
        let mut guard = self.tiberius_conn.lock().await;
        let conn: &mut TiberiusConn = guard.as_mut().unwrap();

        let mut args: Vec<&dyn ToSql> = Vec::new();
        for &p in params {
            args = MssqlParam::add_param(p, args);
        }
        log::debug!("MSSQL_EXEC: {}", sql);
        let r = conn.execute(sql, &args).await?;
        Ok(ExecuteResult {
            rows_affected: r.rows_affected().iter().sum(),
        })
    }

    async fn fetch_rows(&self, sql: &str, params: &[&(dyn Param + Sync)]) -> Result<Vec<Row>> {
        let mut guard = self.tiberius_conn.lock().await;
        let conn: &mut TiberiusConn = guard.as_mut().unwrap();

        let mut args: Vec<&dyn ToSql> = Vec::new();
        for &p in params {
            args = MssqlParam::add_param(p, args);
        }
        log::debug!("MSSQL_QUERY: {}", sql);
        let stream = conn.query(sql, &args).await?;

        let mssql_rows = stream.into_results().await?;
        let mut all = Vec::default();
        for batch in mssql_rows {
            for r in batch {
                all.push(Row::from(r))
            }
        }
        Ok(all)
    }

    async fn fetch_many<'s, 'args, 't>(
        &self,
        args: &[crate::Fetch<'s, 'args, 't>],
    ) -> Result<Vec<Vec<Row>>> {
        let mut resultset = Vec::default();
        let mut guard = self.tiberius_conn.lock().await;
        let conn: &mut TiberiusConn = guard.as_mut().unwrap();
        for fetch in args {
            let sql = fetch.sql;
            let params = fetch.params;
            let mut args: Vec<&dyn ToSql> = Vec::new();
            for &p in params {
                args = MssqlParam::add_param(p, args);
            }
            log::debug!("MSSQL_QUERY: {}", sql);
            let stream = conn.query(sql, &args).await?;
            let mssql_rows = stream.into_results().await?;
            let mut all = Vec::default();
            for batch in mssql_rows {
                for r in batch {
                    all.push(Row::from(r))
                }
            }
            resultset.push(all)
        }
        Ok(resultset)
    }

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