use std::convert::TryInto;
use std::marker::PhantomData;
use std::pin::Pin;
use std::ptr::NonNull;
use std::time::{Duration, Instant};
use foundationdb_sys as fdb_sys;
use crate::options;
use crate::transaction::*;
use crate::{error, FdbError, FdbResult};
use futures::prelude::*;
pub struct Database {
pub(crate) inner: NonNull<fdb_sys::FDBDatabase>,
}
unsafe impl Send for Database {}
unsafe impl Sync for Database {}
impl Drop for Database {
fn drop(&mut self) {
unsafe {
fdb_sys::fdb_database_destroy(self.inner.as_ptr());
}
}
}
#[cfg(not(any(feature = "fdb-5_1", feature = "fdb-5_2", feature = "fdb-6_0")))]
impl Database {
pub fn new(path: Option<&str>) -> FdbResult<Database> {
let path_str =
path.map(|path| std::ffi::CString::new(path).expect("path to be convertible to CStr"));
let path_ptr = path_str
.as_ref()
.map(|path| path.as_ptr())
.unwrap_or(std::ptr::null());
let mut v: *mut fdb_sys::FDBDatabase = std::ptr::null_mut();
let err = unsafe { fdb_sys::fdb_create_database(path_ptr, &mut v) };
drop(path_str);
error::eval(err)?;
Ok(Database {
inner: NonNull::new(v)
.expect("fdb_create_database to not return null if there is no error"),
})
}
pub fn from_path(path: &str) -> FdbResult<Database> {
Self::new(Some(path))
}
pub fn default() -> FdbResult<Database> {
Self::new(None)
}
}
impl Database {
pub async fn new_compat(path: Option<&str>) -> FdbResult<Database> {
#[cfg(any(feature = "fdb-5_1", feature = "fdb-5_2", feature = "fdb-6_0"))]
{
let cluster = crate::cluster::Cluster::new(path).await?;
let database = cluster.create_database().await?;
Ok(database)
}
#[cfg(not(any(feature = "fdb-5_1", feature = "fdb-5_2", feature = "fdb-6_0")))]
{
Database::new(path)
}
}
pub fn set_option(&self, opt: options::DatabaseOption) -> FdbResult<()> {
unsafe { opt.apply(self.inner.as_ptr()) }
}
pub fn create_trx(&self) -> FdbResult<Transaction> {
let mut trx: *mut fdb_sys::FDBTransaction = std::ptr::null_mut();
let err =
unsafe { fdb_sys::fdb_database_create_transaction(self.inner.as_ptr(), &mut trx) };
error::eval(err)?;
Ok(Transaction::new(NonNull::new(trx).expect(
"fdb_database_create_transaction to not return null if there is no error",
)))
}
pub async fn transact<F>(&self, mut f: F, options: TransactOption) -> Result<F::Item, F::Error>
where
F: DatabaseTransact,
{
let is_idempotent = options.is_idempotent;
let time_out = options.time_out.map(|d| Instant::now() + d);
let retry_limit = options.retry_limit;
let mut tries: u32 = 0;
let mut trx = self.create_trx()?;
let mut can_retry = move || {
tries += 1;
retry_limit.map(|limit| tries < limit).unwrap_or(true)
&& time_out.map(|t| Instant::now() < t).unwrap_or(true)
};
loop {
let r = f.transact(trx).await;
f = r.0;
trx = r.1;
trx = match r.2 {
Ok(item) => match trx.commit().await {
Ok(_) => break Ok(item),
Err(e) => {
if (is_idempotent || !e.is_maybe_committed()) && can_retry() {
e.on_error().await?
} else {
break Err(F::Error::from(e.into()));
}
}
},
Err(user_err) => match user_err.try_into_fdb_error() {
Ok(e) => {
if (is_idempotent || !e.is_maybe_committed()) && can_retry() {
trx.on_error(e).await?
} else {
break Err(F::Error::from(e));
}
}
Err(user_err) => break Err(user_err),
},
};
}
}
pub fn transact_boxed<'trx, F, D, T, E>(
&'trx self,
data: D,
f: F,
options: TransactOption,
) -> impl Future<Output = Result<T, E>> + Send + 'trx
where
for<'a> F: FnMut(
&'a Transaction,
&'a mut D,
) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>,
E: TransactError,
F: Send + 'trx,
T: Send + 'trx,
E: Send + 'trx,
D: Send + 'trx,
{
self.transact(
boxed::FnMutBoxed {
f,
d: data,
m: PhantomData,
},
options,
)
}
pub fn transact_boxed_local<'trx, F, D, T, E>(
&'trx self,
data: D,
f: F,
options: TransactOption,
) -> impl Future<Output = Result<T, E>> + 'trx
where
for<'a> F:
FnMut(&'a Transaction, &'a mut D) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>,
E: TransactError,
F: 'trx,
T: 'trx,
E: 'trx,
D: 'trx,
{
self.transact(
boxed_local::FnMutBoxedLocal {
f,
d: data,
m: PhantomData,
},
options,
)
}
}
pub trait DatabaseTransact: Sized {
type Item;
type Error: TransactError;
type Future: Future<Output = (Self, Transaction, Result<Self::Item, Self::Error>)>;
fn transact(self, trx: Transaction) -> Self::Future;
}
#[allow(clippy::needless_lifetimes)]
#[allow(clippy::type_complexity)]
mod boxed {
use super::*;
async fn boxed_data_fut<'t, F, T, E, D>(
mut f: FnMutBoxed<'t, F, D>,
trx: Transaction,
) -> (FnMutBoxed<'t, F, D>, Transaction, Result<T, E>)
where
F: for<'a> FnMut(
&'a Transaction,
&'a mut D,
) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>,
E: TransactError,
{
let r = (&mut f.f)(&trx, &mut f.d).await;
(f, trx, r)
}
pub struct FnMutBoxed<'t, F, D> {
pub f: F,
pub d: D,
pub m: PhantomData<&'t ()>,
}
impl<'t, F, T, E, D> DatabaseTransact for FnMutBoxed<'t, F, D>
where
F: for<'a> FnMut(
&'a Transaction,
&'a mut D,
) -> Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>,
F: 't + Send,
T: 't,
E: 't,
D: 't + Send,
E: TransactError,
{
type Item = T;
type Error = E;
type Future = Pin<
Box<
dyn Future<Output = (Self, Transaction, Result<Self::Item, Self::Error>)>
+ Send
+ 't,
>,
>;
fn transact(self, trx: Transaction) -> Self::Future {
boxed_data_fut(self, trx).boxed()
}
}
}
#[allow(clippy::needless_lifetimes)]
#[allow(clippy::type_complexity)]
mod boxed_local {
use super::*;
async fn boxed_local_data_fut<'t, F, T, E, D>(
mut f: FnMutBoxedLocal<'t, F, D>,
trx: Transaction,
) -> (FnMutBoxedLocal<'t, F, D>, Transaction, Result<T, E>)
where
F: for<'a> FnMut(
&'a Transaction,
&'a mut D,
) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>,
E: TransactError,
{
let r = (&mut f.f)(&trx, &mut f.d).await;
(f, trx, r)
}
pub struct FnMutBoxedLocal<'t, F, D> {
pub f: F,
pub d: D,
pub m: PhantomData<&'t ()>,
}
impl<'t, F, T, E, D> DatabaseTransact for FnMutBoxedLocal<'t, F, D>
where
F: for<'a> FnMut(
&'a Transaction,
&'a mut D,
) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>,
F: 't,
T: 't,
E: 't,
D: 't,
E: TransactError,
{
type Item = T;
type Error = E;
type Future = Pin<
Box<dyn Future<Output = (Self, Transaction, Result<Self::Item, Self::Error>)> + 't>,
>;
fn transact(self, trx: Transaction) -> Self::Future {
boxed_local_data_fut(self, trx).boxed_local()
}
}
}
pub trait TransactError: From<FdbError> {
fn try_into_fdb_error(self) -> Result<FdbError, Self>;
}
impl<T> TransactError for T
where
T: From<FdbError> + TryInto<FdbError, Error = T>,
{
fn try_into_fdb_error(self) -> Result<FdbError, Self> {
self.try_into()
}
}
impl TransactError for FdbError {
fn try_into_fdb_error(self) -> Result<FdbError, Self> {
Ok(self)
}
}
#[derive(Default, Clone)]
pub struct TransactOption {
pub retry_limit: Option<u32>,
pub time_out: Option<Duration>,
pub is_idempotent: bool,
}
impl TransactOption {
pub fn idempotent() -> Self {
Self {
is_idempotent: true,
..TransactOption::default()
}
}
}