use std::{fmt, pin::Pin};
use tracing::debug;
use crate::{
Error, QueryOptions, TransactionOptions,
analyze::AnalyzedQuery,
answer::QueryAnswer,
common::{Promise, Result, TransactionType},
connection::TransactionStream,
given::GivenRows,
};
pub struct Transaction {
type_: TransactionType,
options: TransactionOptions,
transaction_stream: Pin<Box<TransactionStream>>,
}
impl Transaction {
pub(super) fn new(transaction_stream: TransactionStream) -> Self {
let transaction_stream = Box::pin(transaction_stream);
Transaction {
type_: transaction_stream.type_(),
options: transaction_stream.options().clone(),
transaction_stream,
}
}
pub fn is_open(&self) -> bool {
self.transaction_stream.is_open()
}
pub fn query(&self, query: impl AsRef<str>) -> impl Promise<'static, Result<QueryAnswer>> {
self.query_with_options_and_rows(query, QueryOptions::new(), None)
}
pub fn query_with_options(
&self,
query: impl AsRef<str>,
options: QueryOptions,
) -> impl Promise<'static, Result<QueryAnswer>> {
self.query_with_options_and_rows(query, options, None)
}
pub fn query_with_rows(
&self,
query: impl AsRef<str>,
rows: GivenRows,
) -> impl Promise<'static, Result<QueryAnswer>> {
self.query_with_options_and_rows(query, QueryOptions::new(), Some(rows))
}
pub fn query_with_options_and_rows(
&self,
query: impl AsRef<str>,
options: QueryOptions,
rows: Option<GivenRows>,
) -> impl Promise<'static, Result<QueryAnswer>> {
let query = query.as_ref();
debug!("Transaction submitting query: {}", query);
self.transaction_stream.query(query, options, rows)
}
pub fn analyze(&self, query: impl AsRef<str>) -> impl Promise<'static, Result<AnalyzedQuery>> {
self.transaction_stream.analyze(query.as_ref())
}
pub fn type_(&self) -> TransactionType {
self.type_
}
pub fn on_close(
&self,
callback: impl FnOnce(Option<Error>) + Send + Sync + 'static,
) -> impl Promise<'_, Result<()>> {
self.transaction_stream.on_close(callback)
}
#[cfg_attr(feature = "sync", doc = "transaction.close().resolve()")]
#[cfg_attr(not(feature = "sync"), doc = "transaction.close().await")]
pub fn close(&self) -> impl Promise<'_, Result<()>> {
self.transaction_stream.close()
}
#[cfg_attr(feature = "sync", doc = "transaction.commit()")]
#[cfg_attr(not(feature = "sync"), doc = "transaction.commit().await")]
pub fn commit(self) -> impl Promise<'static, Result> {
let stream = self.transaction_stream;
stream.commit()
}
#[cfg_attr(feature = "sync", doc = "transaction.rollback()")]
#[cfg_attr(not(feature = "sync"), doc = "transaction.rollback().await")]
pub fn rollback(&self) -> impl Promise<'_, Result> {
self.transaction_stream.rollback()
}
}
impl fmt::Debug for Transaction {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Transaction").field("type_", &self.type_).field("options", &self.options).finish()
}
}