use super::{Dialect, ExecError};
use crate::sql::Pool;
pub enum PoolTx<'a> {
#[cfg(feature = "postgres")]
Postgres(sqlx::Transaction<'a, sqlx::Postgres>),
#[cfg(feature = "mysql")]
Mysql(sqlx::Transaction<'a, sqlx::MySql>),
#[cfg(feature = "sqlite")]
Sqlite(sqlx::Transaction<'a, sqlx::Sqlite>),
}
impl<'a> PoolTx<'a> {
pub async fn commit(self) -> Result<(), sqlx::Error> {
match self {
#[cfg(feature = "postgres")]
PoolTx::Postgres(tx) => tx.commit().await,
#[cfg(feature = "mysql")]
PoolTx::Mysql(tx) => tx.commit().await,
#[cfg(feature = "sqlite")]
PoolTx::Sqlite(tx) => tx.commit().await,
}
}
pub async fn rollback(self) -> Result<(), sqlx::Error> {
match self {
#[cfg(feature = "postgres")]
PoolTx::Postgres(tx) => tx.rollback().await,
#[cfg(feature = "mysql")]
PoolTx::Mysql(tx) => tx.rollback().await,
#[cfg(feature = "sqlite")]
PoolTx::Sqlite(tx) => tx.rollback().await,
}
}
pub(crate) async fn execute_unprepared(&mut self, sql: &str) -> Result<(), sqlx::Error> {
use sqlx::Executor as _;
match self {
#[cfg(feature = "postgres")]
PoolTx::Postgres(tx) => (&mut **tx).execute(sqlx::raw_sql(sql)).await.map(drop),
#[cfg(feature = "mysql")]
PoolTx::Mysql(tx) => (&mut **tx).execute(sqlx::raw_sql(sql)).await.map(drop),
#[cfg(feature = "sqlite")]
PoolTx::Sqlite(tx) => (&mut **tx).execute(sqlx::raw_sql(sql)).await.map(drop),
}
}
pub(crate) async fn still_open(&mut self) -> bool {
match self {
#[cfg(feature = "postgres")]
PoolTx::Postgres(_) => self.execute_unprepared("SELECT 1").await.is_ok(),
#[cfg(feature = "mysql")]
PoolTx::Mysql(tx) => {
use sqlx::Connection as _;
match (**tx).begin().await {
Ok(probe) => probe.commit().await.is_ok(),
Err(_) => false,
}
}
#[cfg(feature = "sqlite")]
PoolTx::Sqlite(_) => true,
}
}
pub(crate) async fn on_sqlite_rollback(
&mut self,
f: impl FnMut() + Send + 'static,
) -> Result<(), sqlx::Error> {
match self {
#[cfg(feature = "sqlite")]
PoolTx::Sqlite(tx) => {
(**tx).lock_handle().await?.set_rollback_hook(f);
Ok(())
}
#[allow(unreachable_patterns)]
_ => {
drop(f);
Ok(())
}
}
}
pub(crate) async fn clear_sqlite_rollback(&mut self) {
#[cfg(feature = "sqlite")]
#[allow(irrefutable_let_patterns)]
if let PoolTx::Sqlite(tx) = self {
if let Ok(mut h) = (**tx).lock_handle().await {
h.remove_rollback_hook();
}
}
}
#[must_use]
pub fn dialect(&self) -> &'static dyn Dialect {
match self {
#[cfg(feature = "postgres")]
PoolTx::Postgres(_) => crate::sql::postgres::DIALECT,
#[cfg(feature = "mysql")]
PoolTx::Mysql(_) => crate::sql::mysql::DIALECT,
#[cfg(feature = "sqlite")]
PoolTx::Sqlite(_) => crate::sql::sqlite::DIALECT,
}
}
}
pub async fn transaction_pool(pool: &Pool) -> Result<PoolTx<'static>, ExecError> {
match pool {
#[cfg(feature = "postgres")]
Pool::Postgres(pg) => Ok(PoolTx::Postgres(pg.begin().await?)),
#[cfg(feature = "mysql")]
Pool::Mysql(my) => Ok(PoolTx::Mysql(my.begin().await?)),
#[cfg(feature = "sqlite")]
Pool::Sqlite(sq) => Ok(PoolTx::Sqlite(sq.begin().await?)),
}
}
pub(crate) async fn write_transaction_pool(pool: &Pool) -> Result<PoolTx<'static>, ExecError> {
match pool {
#[cfg(feature = "sqlite")]
Pool::Sqlite(sq) => Ok(PoolTx::Sqlite(sq.begin_with("BEGIN IMMEDIATE").await?)),
#[allow(unreachable_patterns)]
_ => transaction_pool(pool).await,
}
}