aboutsummaryrefslogtreecommitdiff
path: root/src/storage/in_memory.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/storage/in_memory.rs')
-rw-r--r--src/storage/in_memory.rs24
1 files changed, 16 insertions, 8 deletions
diff --git a/src/storage/in_memory.rs b/src/storage/in_memory.rs
index 7d8d108..b1f0508 100644
--- a/src/storage/in_memory.rs
+++ b/src/storage/in_memory.rs
@@ -122,13 +122,14 @@ fn prefix_last_bound(prefix: &str) -> Bound<String> {
#[async_trait]
impl IStore for MemStore {
async fn row_fetch<'a>(&self, select: &Selector<'a>) -> Result<Vec<RowVal>, StorageError> {
+ tracing::trace!(select=%select, command="row_fetch");
let store = self.row.read().or(Err(StorageError::Internal))?;
match select {
Selector::Range { shard, sort_begin, sort_end } => {
Ok(store
.get(*shard)
- .ok_or(StorageError::NotFound)?
+ .unwrap_or(&BTreeMap::new())
.range((Included(sort_begin.to_string()), Excluded(sort_end.to_string())))
.map(|(k, v)| v.to_row_val(RowRef::new(shard, k)))
.collect::<Vec<_>>())
@@ -136,12 +137,10 @@ impl IStore for MemStore {
Selector::List(rlist) => {
let mut acc = vec![];
for row_ref in rlist {
- let intval = store
- .get(&row_ref.uid.shard)
- .ok_or(StorageError::NotFound)?
- .get(&row_ref.uid.sort)
- .ok_or(StorageError::NotFound)?;
- acc.push(intval.to_row_val(row_ref.clone()));
+ let maybe_intval = store.get(&row_ref.uid.shard).map(|v| v.get(&row_ref.uid.sort)).flatten();
+ if let Some(intval) = maybe_intval {
+ acc.push(intval.to_row_val(row_ref.clone()));
+ }
}
Ok(acc)
},
@@ -150,7 +149,7 @@ impl IStore for MemStore {
Ok(store
.get(*shard)
- .ok_or(StorageError::NotFound)?
+ .unwrap_or(&BTreeMap::new())
.range((Included(sort_prefix.to_string()), last_bound))
.map(|(k, v)| v.to_row_val(RowRef::new(shard, k)))
.collect::<Vec<_>>())
@@ -167,6 +166,7 @@ impl IStore for MemStore {
}
async fn row_rm_single(&self, entry: &RowRef) -> Result<(), StorageError> {
+ tracing::trace!(entry=%entry, command="row_rm_single");
let mut store = self.row.write().or(Err(StorageError::Internal))?;
let shard = &entry.uid.shard;
let sort = &entry.uid.sort;
@@ -190,6 +190,7 @@ impl IStore for MemStore {
}
async fn row_rm<'a>(&self, select: &Selector<'a>) -> Result<(), StorageError> {
+ tracing::trace!(select=%select, command="row_rm");
//@FIXME not efficient at all...
let values = self.row_fetch(select).await?;
@@ -200,6 +201,7 @@ impl IStore for MemStore {
}
async fn row_insert(&self, values: Vec<RowVal>) -> Result<(), StorageError> {
+ tracing::trace!(entries=%values.iter().map(|v| v.row_ref.to_string()).collect::<Vec<_>>().join(","), command="row_insert");
let mut store = self.row.write().or(Err(StorageError::Internal))?;
for v in values.into_iter() {
let shard = v.row_ref.uid.shard;
@@ -228,6 +230,7 @@ impl IStore for MemStore {
Ok(())
}
async fn row_poll(&self, value: &RowRef) -> Result<RowVal, StorageError> {
+ tracing::trace!(entry=%value, command="row_poll");
let shard = &value.uid.shard;
let sort = &value.uid.sort;
let cauz = match value.causality.as_ref().map(|v| v.parse::<u64>()) {
@@ -253,10 +256,12 @@ impl IStore for MemStore {
}
async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result<BlobVal, StorageError> {
+ tracing::trace!(entry=%blob_ref, command="blob_fetch");
let store = self.blob.read().or(Err(StorageError::Internal))?;
store.get(&blob_ref.0).ok_or(StorageError::NotFound).map(|v| v.to_blob_val(blob_ref))
}
async fn blob_insert(&self, blob_val: &BlobVal) -> Result<(), StorageError> {
+ tracing::trace!(entry=%blob_val.blob_ref, command="blob_insert");
let mut store = self.blob.write().or(Err(StorageError::Internal))?;
let entry = store.entry(blob_val.blob_ref.0.clone()).or_default();
entry.data = blob_val.value.clone();
@@ -264,18 +269,21 @@ impl IStore for MemStore {
Ok(())
}
async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<(), StorageError> {
+ tracing::trace!(src=%src, dst=%dst, command="blob_copy");
let mut store = self.blob.write().or(Err(StorageError::Internal))?;
let blob_src = store.entry(src.0.clone()).or_default().clone();
store.insert(dst.0.clone(), blob_src);
Ok(())
}
async fn blob_list(&self, prefix: &str) -> Result<Vec<BlobRef>, StorageError> {
+ tracing::trace!(prefix=prefix, command="blob_list");
let store = self.blob.read().or(Err(StorageError::Internal))?;
let last_bound = prefix_last_bound(prefix);
let blist = store.range((Included(prefix.to_string()), last_bound)).map(|(k, _)| BlobRef(k.to_string())).collect::<Vec<_>>();
Ok(blist)
}
async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError> {
+ tracing::trace!(entry=%blob_ref, command="blob_rm");
let mut store = self.blob.write().or(Err(StorageError::Internal))?;
store.remove(&blob_ref.0);
Ok(())