added write_transaction in broker_store

p2p-broker/src/broker_store_overlay.rs

@@ -73,32 +73,33 @@ impl<'a> Overlay<'a> {
         if acc.exists() {
             return Err(StorageError::BackendError);
         }
-        store.put(
+        store.write_transaction(&|tx| {
-            Self::PREFIX,
+            tx.put(
-            &to_vec(&id)?,
-            Some(Self::SECRET),
-            to_vec(&secret)?,
-        )?;
-        if repo.is_some() {
-            store.put(
                 Self::PREFIX,
                 &to_vec(&id)?,
-                Some(Self::REPO),
+                Some(Self::SECRET),
-                to_vec(&repo.unwrap())?,
+                &to_vec(&secret)?,
             )?;
-            //TODO if failure, should remove the previously added SECRET property
+            if repo.is_some() {
-        }
+                tx.put(
-        let meta = OverlayMeta {
+                    Self::PREFIX,
-            users: 1,
+                    &to_vec(&id)?,
-            last_used: now_timestamp(),
+                    Some(Self::REPO),
-        };
+                    & to_vec(&repo.unwrap())?,
-        store.put(
+                )?;
-            Self::PREFIX,
+            }
-            &to_vec(&id)?,
+            let meta = OverlayMeta {
-            Some(Self::META),
+                users: 1,
-            to_vec(&meta)?,
+                last_used: now_timestamp(),
-        )?;
+            };
-        //TODO if failure, should remove the previously added SECRET and REPO properties
+            tx.put(
+                Self::PREFIX,
+                &to_vec(&id)?,
+                Some(Self::META),
+                &to_vec(&meta)?,
+            )?;
+            Ok(())
+        })?;
         Ok(acc)
     }
     pub fn exists(&self) -> bool {

p2p-broker/src/broker_store_peer.rs

@@ -74,18 +74,21 @@ impl<'a> Peer<'a> {
         if acc.exists() {
             return Err(StorageError::BackendError);
         }
-        store.put(
+        store.write_transaction(&|tx| {
-            Self::PREFIX,
+            tx.put(
-            &to_vec(&id)?,
+                Self::PREFIX,
-            Some(Self::VERSION),
+                &to_vec(&id)?,
-            to_vec(&advert.version())?,
+                Some(Self::VERSION),
-        )?;
+                & to_vec(&advert.version())?,
-        store.put(
+            )?;
-            Self::PREFIX,
+            tx.put(
-            &to_vec(&id)?,
+                Self::PREFIX,
-            Some(Self::ADVERT),
+                &to_vec(&id)?,
-            to_vec(&advert)?,
+                Some(Self::ADVERT),
-        )?;
+                & to_vec(&advert)?,
+            )?;
+            Ok(())
+        })?;
         Ok(acc)
     }
     pub fn exists(&self) -> bool {

p2p-broker/src/broker_store_topic.rs

@@ -50,7 +50,7 @@ impl<'a> Topic<'a> {
         }
         Ok(opening)
     }
-    pub fn create(id: &TopicId, store: &'a dyn BrokerStore) -> Result<Topic<'a>, StorageError> {
+    pub fn create(id: &TopicId, store: &'a mut dyn BrokerStore) -> Result<Topic<'a>, StorageError> {
         let acc = Topic {
             id: id.clone(),
             store,

p2p-repo/src/broker_store.rs

@@ -8,7 +8,45 @@
 
 use crate::store::{StorageError};
 
-pub trait BrokerStore {
+pub trait WriteTransaction : ReadTransaction {
+
+    /// Save a property value to the store.
+    fn put(
+        &mut self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: &Vec<u8>,
+    ) -> Result<(), StorageError>;
+
+    /// Replace the property of a key (single value) to the store.
+    fn replace(
+        &mut self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: &Vec<u8>,
+    ) -> Result<(), StorageError>;
+
+    /// Delete a property from the store.
+    fn del(&mut self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<(), StorageError>;
+
+    /// Delete all properties of a key from the store.
+    fn del_all(&mut self, prefix: u8, key: &Vec<u8>, all_suffixes: &[u8]) -> Result<(), StorageError>;
+
+    /// Delete a specific value for a property from the store.
+    fn del_property_value(
+        &mut self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: &Vec<u8>,
+    ) -> Result<(), StorageError>;
+
+}
+
+pub trait ReadTransaction {
+
     /// Load a property from the store.
     fn get(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<Vec<u8>, StorageError>;
 
@@ -29,6 +67,12 @@ pub trait BrokerStore {
         value: Vec<u8>,
     ) -> Result<(), StorageError>;
 
+}
+
+pub trait BrokerStore : ReadTransaction {
+
+    fn write_transaction(&self, method: & dyn Fn(&mut dyn WriteTransaction) -> Result<(), StorageError> ) -> Result<(), StorageError> ;
+
     /// Save a property value to the store.
     fn put(
         &self,
@@ -61,4 +105,6 @@ pub trait BrokerStore {
         suffix: Option<u8>,
         value: Vec<u8>,
     ) -> Result<(), StorageError>;
+
+
 }

p2p-stores-lmdb/src/broker_store.rs

@@ -15,6 +15,7 @@ use p2p_repo::utils::*;
 use debug_print::*;
 use std::path::Path;
 use std::path::PathBuf;
+use std::sync::RwLockReadGuard;
 use std::sync::{Arc, RwLock};
 
 use rkv::backend::{
@@ -28,6 +29,165 @@ use rkv::{
 use serde::{Deserialize, Serialize};
 use serde_bare::error::Error;
 
+
+pub struct LmdbTransaction<'a> {
+
+    store: &'a LmdbBrokerStore,
+    writer: Option<Writer<LmdbRwTransaction<'a>>>,
+
+}
+
+impl<'a> LmdbTransaction<'a> {
+
+    fn commit(&mut self) {
+        self.writer.take().unwrap().commit().unwrap();
+    }
+
+}
+
+impl<'a> ReadTransaction for LmdbTransaction<'a> {
+    /// Load a single value property from the store.
+    fn get(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<Vec<u8>, StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+        
+        let mut iter = self
+            .store.main_store
+            .get(self.writer.as_ref().unwrap(), property)
+            .map_err(|e| StorageError::BackendError)?;
+        match iter.next() {
+            Some(Ok(val)) => Ok(val.1.to_bytes().unwrap()),
+            Some(Err(_e)) => Err(StorageError::BackendError),
+            None => Err(StorageError::NotFound),
+        }
+    }
+
+    /// Load all the values of a property from the store.
+    fn get_all(
+        &self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+    ) -> Result<Vec<Vec<u8>>, StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+       
+        let mut iter = self
+            .store.main_store
+            .get(self.writer.as_ref().unwrap(), property)
+            .map_err(|e| StorageError::BackendError)?;
+        let mut vector: Vec<Vec<u8>> = vec![];
+        while let res = iter.next() {
+            vector.push(match res {
+                Some(Ok(val)) => val.1.to_bytes().unwrap(),
+                Some(Err(_e)) => return Err(StorageError::BackendError),
+                None => {
+                    break;
+                }
+            });
+        }
+        Ok(vector)
+    }
+
+    /// Check if a specific value exists for a property from the store.
+    fn has_property_value(
+        &self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: Vec<u8>,
+    ) -> Result<(), StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+
+        let exists = self
+            .store.main_store
+            .get_key_value(self.writer.as_ref().unwrap(), property, &Value::Blob(value.as_slice()))
+            .map_err(|e| StorageError::BackendError)?;
+        if exists {
+            Ok(())
+        } else {
+            Err(StorageError::NotFound)
+        }
+    }
+
+}
+
+impl<'a> WriteTransaction for LmdbTransaction<'a> {
+
+    /// Save a property value to the store.
+    fn put(
+        &mut self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: &Vec<u8>,
+    ) -> Result<(), StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+        self.store.main_store
+            .put(self.writer.as_mut().unwrap(), property, &Value::Blob(value.as_slice()))
+            .map_err(|e| StorageError::BackendError)?;
+
+        Ok(())
+    }
+
+    /// Replace the property of a key (single value) to the store.
+    fn replace(
+        &mut self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: &Vec<u8>,
+    ) -> Result<(), StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+
+        self.store.main_store
+            .delete_all(self.writer.as_mut().unwrap(), property.clone())
+            .map_err(|e| StorageError::BackendError)?;
+
+        self.store.main_store
+            .put(self.writer.as_mut().unwrap(), property, &Value::Blob(value.as_slice()))
+            .map_err(|e| StorageError::BackendError)?;
+
+        Ok(())
+    }
+
+    /// Delete a property from the store.
+    fn del(&mut self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<(), StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+        self.store.main_store
+            .delete_all(self.writer.as_mut().unwrap(), property)
+            .map_err(|e| StorageError::BackendError)?;
+
+        Ok(())
+    }
+
+    /// Delete a specific value for a property from the store.
+    fn del_property_value(
+        &mut self,
+        prefix: u8,
+        key: &Vec<u8>,
+        suffix: Option<u8>,
+        value: &Vec<u8>,
+    ) -> Result<(), StorageError> {
+        let property = LmdbBrokerStore::compute_property(prefix, key, suffix);
+        self.store.main_store
+            .delete(self.writer.as_mut().unwrap(), property, &Value::Blob(value.as_slice()))
+            .map_err(|e| StorageError::BackendError)?;
+
+        Ok(())
+    }
+
+    /// Delete all properties of a key from the store.
+    fn del_all(&mut self, prefix: u8, key: &Vec<u8>, all_suffixes: &[u8]) -> Result<(), StorageError> {
+        for suffix in all_suffixes {
+            self.del(prefix, key, Some(*suffix))?;
+        }
+        if all_suffixes.is_empty() {
+            self.del(prefix, key, None)?;
+        }
+        Ok(())
+    }
+}
+
+
 pub struct LmdbBrokerStore {
     /// the main store where all the properties of keys are stored
     main_store: MultiStore<LmdbDatabase>,
@@ -37,7 +197,8 @@ pub struct LmdbBrokerStore {
     path: String,
 }
 
-impl BrokerStore for LmdbBrokerStore {
+impl ReadTransaction for LmdbBrokerStore {
+
     /// Load a single value property from the store.
     fn get(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<Vec<u8>, StorageError> {
         let property = Self::compute_property(prefix, key, suffix);
@@ -103,6 +264,29 @@ impl BrokerStore for LmdbBrokerStore {
         }
     }
 
+
+}
+
+impl BrokerStore for LmdbBrokerStore {
+
+    fn write_transaction(&self,  method: & dyn Fn(&mut dyn WriteTransaction) -> Result<(), StorageError> )-> Result<(), StorageError> {
+
+        let lock = self.environment.read().unwrap();
+        let writer = lock.write().unwrap();
+        
+        let mut transaction = LmdbTransaction {
+            store: self,
+            writer: Some(writer),
+        };
+        let res = method(&mut transaction);
+        
+        if res.is_ok() {
+            transaction.commit();
+        }
+        res
+
+    }
+
     /// Save a property value to the store.
     fn put(
         &self,
@@ -111,16 +295,10 @@ impl BrokerStore for LmdbBrokerStore {
         suffix: Option<u8>,
         value: Vec<u8>,
     ) -> Result<(), StorageError> {
-        let property = Self::compute_property(prefix, key, suffix);
-        let lock = self.environment.read().unwrap();
-        let mut writer = lock.write().unwrap();
-        self.main_store
-            .put(&mut writer, property, &Value::Blob(value.as_slice()))
-            .map_err(|e| StorageError::BackendError)?;
 
-        writer.commit().unwrap();
+        self.write_transaction(&|tx| {
-
+            tx.put(prefix,key,suffix,&value)
-        Ok(())
+        })
     }
 
     /// Replace the property of a key (single value) to the store.
@@ -131,34 +309,17 @@ impl BrokerStore for LmdbBrokerStore {
         suffix: Option<u8>,
         value: Vec<u8>,
     ) -> Result<(), StorageError> {
-        let property = Self::compute_property(prefix, key, suffix);
-        let lock = self.environment.read().unwrap();
-        let mut writer = lock.write().unwrap();
-        self.main_store
-            .delete_all(&mut writer, property.clone())
-            .map_err(|e| StorageError::BackendError)?;
-
-        self.main_store
-            .put(&mut writer, property, &Value::Blob(value.as_slice()))
-            .map_err(|e| StorageError::BackendError)?;
 
-        writer.commit().unwrap();
+        self.write_transaction(&|tx| {
-
+            tx.replace(prefix,key,suffix,&value)
-        Ok(())
+        })
     }
 
     /// Delete a property from the store.
     fn del(&self, prefix: u8, key: &Vec<u8>, suffix: Option<u8>) -> Result<(), StorageError> {
-        let property = Self::compute_property(prefix, key, suffix);
+        self.write_transaction(&|tx| {
-        let lock = self.environment.read().unwrap();
+            tx.del(prefix,key,suffix)
-        let mut writer = lock.write().unwrap();
+        })
-        self.main_store
-            .delete_all(&mut writer, property)
-            .map_err(|e| StorageError::BackendError)?;
-
-        writer.commit().unwrap();
-
-        Ok(())
     }
 
     /// Delete a specific value for a property from the store.
@@ -169,16 +330,9 @@ impl BrokerStore for LmdbBrokerStore {
         suffix: Option<u8>,
         value: Vec<u8>,
     ) -> Result<(), StorageError> {
-        let property = Self::compute_property(prefix, key, suffix);
+        self.write_transaction(&|tx| {
-        let lock = self.environment.read().unwrap();
+            tx.del_property_value(prefix,key,suffix, &value)
-        let mut writer = lock.write().unwrap();
+        })
-        self.main_store
-            .delete(&mut writer, property, &Value::Blob(value.as_slice()))
-            .map_err(|e| StorageError::BackendError)?;
-
-        writer.commit().unwrap();
-
-        Ok(())
     }
 
     /// Delete all properties of a key from the store.