refactor(storage-manager): pass Storage in Arc<Mutex<_>>

The replication service needs to access the underlying `dyn Storage` in
a separate task and mutate its internal state.

This commit simply changes the location where the storage is wrapped
inside an `Arc<Mutex<_>>`.

* plugins/zenoh-plugin-storage-manager/src/storages_mgt/mod.rs: wrap
  the `dyn Storage` inside an `Arc<Mutex<_>>` before spawning the
  `StorageService`.
* plugins/zenoh-plugin-storage-manager/src/storages_mgt/service.rs:
  update the signature of the method `start` to accept an
  `Arc<Mutex<Box dyn Storage>>` instead of a `Box<dyn Storage>`.

Signed-off-by: Julien Loudet <[email protected]>

plugins/zenoh-plugin-storage-manager/src/storages_mgt/mod.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use flume::Sender;
+use tokio::sync::Mutex;
 use zenoh::{session::Session, Result as ZResult};
 use zenoh_backend_traits::{config::StorageConfig, VolumeInstance};
 
@@ -47,6 +48,7 @@ pub(crate) async fn create_and_start_storage(
 
     let (tx, rx) = flume::bounded(1);
 
+    let storage = Arc::new(Mutex::new(storage));
     tokio::task::spawn(async move {
         StorageService::start(zenoh.clone(), config, &name, storage, capability, rx).await;
     });

plugins/zenoh-plugin-storage-manager/src/storages_mgt/service.rs

@@ -60,7 +60,7 @@ pub struct StorageService {
     complete: bool,
     name: String,
     strip_prefix: Option<OwnedKeyExpr>,
-    storage: Mutex<Box<dyn zenoh_backend_traits::Storage>>,
+    storage: Arc<Mutex<Box<dyn zenoh_backend_traits::Storage>>>,
     capability: Capability,
     tombstones: Arc<RwLock<KeBoxTree<Timestamp, NonWild, KeyedSetProvider>>>,
     wildcard_updates: Arc<RwLock<KeBoxTree<Update, UnknownWildness, KeyedSetProvider>>>,
@@ -72,7 +72,7 @@ impl StorageService {
         session: Arc<Session>,
         config: StorageConfig,
         name: &str,
-        storage: Box<dyn zenoh_backend_traits::Storage>,
+        storage: Arc<Mutex<Box<dyn zenoh_backend_traits::Storage>>>,
         capability: Capability,
         rx: Receiver<StorageMessage>,
     ) {
@@ -84,7 +84,7 @@ impl StorageService {
             complete: config.complete,
             name: name.to_string(),
             strip_prefix: config.strip_prefix,
-            storage: Mutex::new(storage),
+            storage,
             capability,
             tombstones: Arc::new(RwLock::new(KeBoxTree::default())),
             wildcard_updates: Arc::new(RwLock::new(KeBoxTree::default())),