zenoh-plugin-storage-manager 1.0.0-beta.2

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
50
51
52
53

//
// 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 std::sync::Arc;

use zenoh::{session::Session, Result as ZResult};
use zenoh_backend_traits::config::StorageConfig;

pub use super::replica::{Replica, StorageService};

pub enum StorageMessage {
    Stop,
    GetStatus(tokio::sync::mpsc::Sender<serde_json::Value>),
}

pub(crate) async fn start_storage(
    store_intercept: super::StoreIntercept,
    config: StorageConfig,
    admin_key: String,
    zenoh: Arc<Session>,
) -> ZResult<flume::Sender<StorageMessage>> {
    // Ex: @/390CEC11A1E34977A1C609A35BC015E6/router/status/plugins/storage_manager/storages/demo1 -> 390CEC11A1E34977A1C609A35BC015E6/demo1 (/<type> needed????)
    let parts: Vec<&str> = admin_key.split('/').collect();
    let uuid = parts[2];
    let storage_name = parts[7];
    let name = format!("{uuid}/{storage_name}");

    tracing::trace!("Start storage '{}' on keyexpr '{}'", name, config.key_expr);

    let (tx, rx) = flume::bounded(1);

    tokio::task::spawn(async move {
        // If a configuration for replica is present, we initialize a replica, else only a storage service
        // A replica contains a storage service and all metadata required for anti-entropy
        if config.replica_config.is_some() {
            Replica::start(zenoh.clone(), store_intercept, config, &name, rx).await;
        } else {
            StorageService::start(zenoh.clone(), config, &name, store_intercept, rx, None).await;
        }
    });

    Ok(tx)
}