#[macro_use] extern crate tracing; #[cfg(feature = "lmdb")] pub mod lmdb_adapter; #[cfg(feature = "lmdb")] pub mod metric_proxy; #[cfg(feature = "sqlite")] pub mod sqlite_adapter; pub mod open; #[cfg(test)] pub mod test; use core::ops::{Bound, RangeBounds}; use std::borrow::Cow; use std::cell::Cell; use std::path::PathBuf; use std::sync::Arc; use err_derive::Error; pub use open::*; pub(crate) type OnCommit = Vec>; #[derive(Clone)] pub struct Db(pub(crate) Arc); pub struct Transaction<'a> { tx: &'a mut dyn ITx, on_commit: OnCommit, } #[derive(Clone)] pub struct Tree(Arc, usize); pub type Value = Vec; pub type ValueIter<'a> = Box> + 'a>; pub type TxValueIter<'a> = Box> + 'a>; // ---- #[derive(Debug, Error)] #[error(display = "{}", _0)] pub struct Error(pub Cow<'static, str>); impl From for Error { fn from(e: std::io::Error) -> Error { Error(format!("IO: {}", e).into()) } } pub type Result = std::result::Result; #[derive(Debug, Error)] #[error(display = "{}", _0)] pub struct TxOpError(pub(crate) Error); pub type TxOpResult = std::result::Result; #[derive(Debug)] pub enum TxError { Abort(E), Db(Error), } pub type TxResult = std::result::Result>; impl From for TxError { fn from(e: TxOpError) -> TxError { TxError::Db(e.0) } } pub fn unabort(res: TxResult) -> TxOpResult> { match res { Ok(v) => Ok(Ok(v)), Err(TxError::Abort(e)) => Ok(Err(e)), Err(TxError::Db(e)) => Err(TxOpError(e)), } } // ---- impl Db { pub fn engine(&self) -> String { self.0.engine() } pub fn open_tree>(&self, name: S) -> Result { let tree_id = self.0.open_tree(name.as_ref())?; Ok(Tree(self.0.clone(), tree_id)) } pub fn list_trees(&self) -> Result> { self.0.list_trees() } pub fn transaction(&self, fun: F) -> TxResult where F: Fn(&mut Transaction<'_>) -> TxResult, { let f = TxFn { function: fun, result: Cell::new(None), }; let tx_res = self.0.transaction(&f); let ret = f .result .into_inner() .expect("Transaction did not store result"); match tx_res { Ok(on_commit) => match ret { Ok(value) => { on_commit.into_iter().for_each(|f| f()); Ok(value) } _ => unreachable!(), }, Err(TxError::Abort(())) => match ret { Err(TxError::Abort(e)) => Err(TxError::Abort(e)), _ => unreachable!(), }, Err(TxError::Db(e2)) => match ret { // Ok was stored -> the error occured when finalizing // transaction Ok(_) => Err(TxError::Db(e2)), // An error was already stored: that's the one we want to // return Err(TxError::Db(e)) => Err(TxError::Db(e)), _ => unreachable!(), }, } } pub fn snapshot(&self, path: &PathBuf) -> Result<()> { self.0.snapshot(path) } pub fn import(&self, other: &Db) -> Result<()> { let existing_trees = self.list_trees()?; if !existing_trees.is_empty() { return Err(Error( format!( "destination database already contains data: {:?}", existing_trees ) .into(), )); } let tree_names = other.list_trees()?; for name in tree_names { let tree = self.open_tree(&name)?; if tree.len()? > 0 { return Err(Error(format!("tree {} already contains data", name).into())); } let ex_tree = other.open_tree(&name)?; let tx_res = self.transaction(|tx| { let mut i = 0; for item in ex_tree.iter().map_err(TxError::Abort)? { let (k, v) = item.map_err(TxError::Abort)?; tx.insert(&tree, k, v)?; i += 1; if i % 1000 == 0 { println!("{}: imported {}", name, i); } } Ok(i) }); let total = match tx_res { Err(TxError::Db(e)) => return Err(e), Err(TxError::Abort(e)) => return Err(e), Ok(x) => x, }; println!("{}: finished importing, {} items", name, total); } Ok(()) } } #[allow(clippy::len_without_is_empty)] impl Tree { #[inline] pub fn db(&self) -> Db { Db(self.0.clone()) } #[inline] pub fn get>(&self, key: T) -> Result> { self.0.get(self.1, key.as_ref()) } #[inline] pub fn len(&self) -> Result { self.0.len(self.1) } #[inline] pub fn first(&self) -> Result> { self.iter()?.next().transpose() } #[inline] pub fn get_gt>(&self, from: T) -> Result> { self.range((Bound::Excluded(from), Bound::Unbounded))? .next() .transpose() } /// Returns the old value if there was one #[inline] pub fn insert, U: AsRef<[u8]>>( &self, key: T, value: U, ) -> Result> { self.0.insert(self.1, key.as_ref(), value.as_ref()) } /// Returns the old value if there was one #[inline] pub fn remove>(&self, key: T) -> Result> { self.0.remove(self.1, key.as_ref()) } /// Clears all values from the tree #[inline] pub fn clear(&self) -> Result<()> { self.0.clear(self.1) } #[inline] pub fn iter(&self) -> Result> { self.0.iter(self.1) } #[inline] pub fn iter_rev(&self) -> Result> { self.0.iter_rev(self.1) } #[inline] pub fn range(&self, range: R) -> Result> where K: AsRef<[u8]>, R: RangeBounds, { let sb = range.start_bound(); let eb = range.end_bound(); self.0.range(self.1, get_bound(sb), get_bound(eb)) } #[inline] pub fn range_rev(&self, range: R) -> Result> where K: AsRef<[u8]>, R: RangeBounds, { let sb = range.start_bound(); let eb = range.end_bound(); self.0.range_rev(self.1, get_bound(sb), get_bound(eb)) } } #[allow(clippy::len_without_is_empty)] impl<'a> Transaction<'a> { #[inline] pub fn get>(&self, tree: &Tree, key: T) -> TxOpResult> { self.tx.get(tree.1, key.as_ref()) } #[inline] pub fn len(&self, tree: &Tree) -> TxOpResult { self.tx.len(tree.1) } /// Returns the old value if there was one #[inline] pub fn insert, U: AsRef<[u8]>>( &mut self, tree: &Tree, key: T, value: U, ) -> TxOpResult> { self.tx.insert(tree.1, key.as_ref(), value.as_ref()) } /// Returns the old value if there was one #[inline] pub fn remove>(&mut self, tree: &Tree, key: T) -> TxOpResult> { self.tx.remove(tree.1, key.as_ref()) } /// Clears all values in a tree #[inline] pub fn clear(&mut self, tree: &Tree) -> TxOpResult<()> { self.tx.clear(tree.1) } #[inline] pub fn iter(&self, tree: &Tree) -> TxOpResult> { self.tx.iter(tree.1) } #[inline] pub fn iter_rev(&self, tree: &Tree) -> TxOpResult> { self.tx.iter_rev(tree.1) } #[inline] pub fn range(&self, tree: &Tree, range: R) -> TxOpResult> where K: AsRef<[u8]>, R: RangeBounds, { let sb = range.start_bound(); let eb = range.end_bound(); self.tx.range(tree.1, get_bound(sb), get_bound(eb)) } #[inline] pub fn range_rev(&self, tree: &Tree, range: R) -> TxOpResult> where K: AsRef<[u8]>, R: RangeBounds, { let sb = range.start_bound(); let eb = range.end_bound(); self.tx.range_rev(tree.1, get_bound(sb), get_bound(eb)) } #[inline] pub fn on_commit(&mut self, f: F) { self.on_commit.push(Box::new(f)); } } // ---- Internal interfaces pub(crate) trait IDb: Send + Sync { fn engine(&self) -> String; fn open_tree(&self, name: &str) -> Result; fn list_trees(&self) -> Result>; fn snapshot(&self, path: &PathBuf) -> Result<()>; fn get(&self, tree: usize, key: &[u8]) -> Result>; fn len(&self, tree: usize) -> Result; fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> Result>; fn remove(&self, tree: usize, key: &[u8]) -> Result>; fn clear(&self, tree: usize) -> Result<()>; fn iter(&self, tree: usize) -> Result>; fn iter_rev(&self, tree: usize) -> Result>; fn range<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, ) -> Result>; fn range_rev<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, ) -> Result>; fn transaction(&self, f: &dyn ITxFn) -> TxResult; } pub(crate) trait ITx { fn get(&self, tree: usize, key: &[u8]) -> TxOpResult>; fn len(&self, tree: usize) -> TxOpResult; fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> TxOpResult>; fn remove(&mut self, tree: usize, key: &[u8]) -> TxOpResult>; fn clear(&mut self, tree: usize) -> TxOpResult<()>; fn iter(&self, tree: usize) -> TxOpResult>; fn iter_rev(&self, tree: usize) -> TxOpResult>; fn range<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, ) -> TxOpResult>; fn range_rev<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, ) -> TxOpResult>; } pub(crate) trait ITxFn { fn try_on(&self, tx: &mut dyn ITx) -> TxFnResult; } pub(crate) enum TxFnResult { Ok(OnCommit), Abort, DbErr, } struct TxFn where F: Fn(&mut Transaction<'_>) -> TxResult, { function: F, result: Cell>>, } impl ITxFn for TxFn where F: Fn(&mut Transaction<'_>) -> TxResult, { fn try_on(&self, tx: &mut dyn ITx) -> TxFnResult { let mut tx = Transaction { tx, on_commit: vec![], }; let res = (self.function)(&mut tx); let res2 = match &res { Ok(_) => TxFnResult::Ok(tx.on_commit), Err(TxError::Abort(_)) => TxFnResult::Abort, Err(TxError::Db(_)) => TxFnResult::DbErr, }; self.result.set(Some(res)); res2 } } // ---- fn get_bound>(b: Bound<&K>) -> Bound<&[u8]> { match b { Bound::Included(v) => Bound::Included(v.as_ref()), Bound::Excluded(v) => Bound::Excluded(v.as_ref()), Bound::Unbounded => Bound::Unbounded, } }