use super::tx::Transaction;
use crate::ctx::Context;
use crate::dbs::Attach;
use crate::dbs::Executor;
use crate::dbs::Options;
use crate::dbs::Response;
use crate::dbs::Session;
use crate::dbs::Variables;
use crate::err::Error;
use crate::kvs::LOG;
use crate::sql;
use crate::sql::Query;
use crate::sql::Value;
use channel::Sender;
use futures::lock::Mutex;
use std::sync::Arc;
pub struct Datastore {
pub(super) inner: Inner,
}
#[allow(clippy::large_enum_variant)]
pub(super) enum Inner {
#[cfg(feature = "kv-mem")]
Mem(super::mem::Datastore),
#[cfg(feature = "kv-rocksdb")]
RocksDB(super::rocksdb::Datastore),
#[cfg(feature = "kv-indxdb")]
IndxDB(super::indxdb::Datastore),
#[cfg(feature = "kv-tikv")]
TiKV(super::tikv::Datastore),
#[cfg(feature = "kv-fdb")]
FDB(super::fdb::Datastore),
}
impl Datastore {
pub async fn new(path: &str) -> Result<Datastore, Error> {
match path {
#[cfg(feature = "kv-mem")]
"memory" => {
info!(target: LOG, "Starting kvs store in {}", path);
let v = super::mem::Datastore::new().await.map(|v| Datastore {
inner: Inner::Mem(v),
});
info!(target: LOG, "Started kvs store in {}", path);
v
}
#[cfg(feature = "kv-rocksdb")]
s if s.starts_with("file:") => {
info!(target: LOG, "Starting kvs store at {}", path);
let s = s.trim_start_matches("file://");
let s = s.trim_start_matches("file:");
let v = super::rocksdb::Datastore::new(s).await.map(|v| Datastore {
inner: Inner::RocksDB(v),
});
info!(target: LOG, "Started kvs store at {}", path);
v
}
#[cfg(feature = "kv-rocksdb")]
s if s.starts_with("rocksdb:") => {
info!(target: LOG, "Starting kvs store at {}", path);
let s = s.trim_start_matches("rocksdb://");
let s = s.trim_start_matches("rocksdb:");
let v = super::rocksdb::Datastore::new(s).await.map(|v| Datastore {
inner: Inner::RocksDB(v),
});
info!(target: LOG, "Started kvs store at {}", path);
v
}
#[cfg(feature = "kv-indxdb")]
s if s.starts_with("indxdb:") => {
info!(target: LOG, "Starting kvs store at {}", path);
let s = s.trim_start_matches("indxdb://");
let s = s.trim_start_matches("indxdb:");
let v = super::indxdb::Datastore::new(s).await.map(|v| Datastore {
inner: Inner::IndxDB(v),
});
info!(target: LOG, "Started kvs store at {}", path);
v
}
#[cfg(feature = "kv-tikv")]
s if s.starts_with("tikv:") => {
info!(target: LOG, "Connecting to kvs store at {}", path);
let s = s.trim_start_matches("tikv://");
let s = s.trim_start_matches("tikv:");
let v = super::tikv::Datastore::new(s).await.map(|v| Datastore {
inner: Inner::TiKV(v),
});
info!(target: LOG, "Connected to kvs store at {}", path);
v
}
#[cfg(feature = "kv-fdb")]
s if s.starts_with("fdb:") => {
info!(target: LOG, "Connecting to kvs store at {}", path);
let s = s.trim_start_matches("fdb://");
let s = s.trim_start_matches("fdb:");
let v = super::fdb::Datastore::new(s).await.map(|v| Datastore {
inner: Inner::FDB(v),
});
info!(target: LOG, "Connected to kvs store at {}", path);
v
}
_ => Err(Error::Ds("Unable to load the specified datastore".into())),
}
}
pub async fn transaction(&self, write: bool, lock: bool) -> Result<Transaction, Error> {
match &self.inner {
#[cfg(feature = "kv-mem")]
Inner::Mem(v) => {
let tx = v.transaction(write, lock).await?;
Ok(Transaction {
inner: super::tx::Inner::Mem(tx),
cache: super::cache::Cache::default(),
})
}
#[cfg(feature = "kv-rocksdb")]
Inner::RocksDB(v) => {
let tx = v.transaction(write, lock).await?;
Ok(Transaction {
inner: super::tx::Inner::RocksDB(tx),
cache: super::cache::Cache::default(),
})
}
#[cfg(feature = "kv-indxdb")]
Inner::IndxDB(v) => {
let tx = v.transaction(write, lock).await?;
Ok(Transaction {
inner: super::tx::Inner::IndxDB(tx),
cache: super::cache::Cache::default(),
})
}
#[cfg(feature = "kv-tikv")]
Inner::TiKV(v) => {
let tx = v.transaction(write, lock).await?;
Ok(Transaction {
inner: super::tx::Inner::TiKV(tx),
cache: super::cache::Cache::default(),
})
}
#[cfg(feature = "kv-fdb")]
Inner::FDB(v) => {
let tx = v.transaction(write, lock).await?;
Ok(Transaction {
inner: super::tx::Inner::FDB(tx),
cache: super::cache::Cache::default(),
})
}
}
}
pub async fn execute(
&self,
txt: &str,
sess: &Session,
vars: Variables,
strict: bool,
) -> Result<Vec<Response>, Error> {
let mut opt = Options::default();
let mut exe = Executor::new(self);
let ctx = Context::default();
let ctx = sess.context(ctx);
let ctx = vars.attach(ctx);
let ast = sql::parse(txt)?;
opt.auth = sess.au.clone();
opt.live = sess.rt;
opt.ns = sess.ns();
opt.db = sess.db();
opt.strict = strict;
exe.execute(ctx, opt, ast).await
}
pub async fn process(
&self,
ast: Query,
sess: &Session,
vars: Variables,
strict: bool,
) -> Result<Vec<Response>, Error> {
let mut opt = Options::default();
let mut exe = Executor::new(self);
let ctx = Context::default();
let ctx = sess.context(ctx);
let ctx = vars.attach(ctx);
opt.auth = sess.au.clone();
opt.live = sess.rt;
opt.ns = sess.ns();
opt.db = sess.db();
opt.strict = strict;
exe.execute(ctx, opt, ast).await
}
pub async fn compute(
&self,
val: Value,
sess: &Session,
vars: Variables,
strict: bool,
) -> Result<Value, Error> {
let txn = self.transaction(val.writeable(), false).await?;
let txn = Arc::new(Mutex::new(txn));
let mut opt = Options::default();
let ctx = Context::default();
let ctx = sess.context(ctx);
let ctx = vars.attach(ctx);
opt.auth = sess.au.clone();
opt.ns = sess.ns();
opt.db = sess.db();
opt.strict = strict;
let res = val.compute(&ctx, &opt, &txn, None).await?;
match val.writeable() {
true => txn.lock().await.commit().await?,
false => txn.lock().await.cancel().await?,
};
Ok(res)
}
pub async fn export(&self, ns: String, db: String, chn: Sender<Vec<u8>>) -> Result<(), Error> {
let mut txn = self.transaction(false, false).await?;
txn.export(&ns, &db, chn).await?;
Ok(())
}
}