zenoh-plugin-storage-manager 0.11.0

The zenoh storages plugin.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

//
// Copyright (c) 2023 ZettaScale Technology
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//
// Contributors:
//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
//
use super::storages_mgt::*;
use flume::Sender;
use std::sync::Arc;
use zenoh::prelude::r#async::*;
use zenoh::Session;
use zenoh_backend_traits::config::StorageConfig;
use zenoh_backend_traits::{Capability, VolumeInstance};
use zenoh_result::ZResult;

pub struct StoreIntercept {
    pub storage: Box<dyn zenoh_backend_traits::Storage>,
    pub capability: Capability,
    pub in_interceptor: Option<Arc<dyn Fn(Sample) -> Sample + Send + Sync>>,
    pub out_interceptor: Option<Arc<dyn Fn(Sample) -> Sample + Send + Sync>>,
}

pub(crate) async fn create_and_start_storage(
    admin_key: String,
    config: StorageConfig,
    backend: &VolumeInstance,
    in_interceptor: Option<Arc<dyn Fn(Sample) -> Sample + Send + Sync>>,
    out_interceptor: Option<Arc<dyn Fn(Sample) -> Sample + Send + Sync>>,
    zenoh: Arc<Session>,
) -> ZResult<Sender<StorageMessage>> {
    tracing::trace!("Create storage '{}'", &admin_key);
    let capability = backend.get_capability();
    let storage = backend.create_storage(config.clone()).await?;
    let store_intercept = StoreIntercept {
        storage,
        capability,
        in_interceptor,
        out_interceptor,
    };

    start_storage(store_intercept, config, admin_key, zenoh).await
}