harmos-storage 0.1.7

Harmos storage: durable journals, snapshots and Arrow-backed stream storage.
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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */

// Charged Directory Discovery And Native Change Subscription

/* === Modules === */

mod scan;
mod selection;
mod watch;

/* === Re-Exports === */

pub use scan::DirectoryReader;
pub use watch::DirectoryWatcher;

/* === Imports === */

// Standard Library Imports
use std::path::Path;
use std::sync::Arc;
// Third Party Imports
// Local Imports
use crate::manager::Inner;
use crate::{Error, Message, ReadBatch, Result, StorageManager};
use harmos_domain::model::storage::{DirectoryCompletion, DirectoryEntry, DirectoryEvent, DirectorySelection};

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

struct Selection {
    value: DirectorySelection,
}

/* === Enums === */

/* === Traits === */

/* === Implementations === */

impl<C> StorageManager<C> {
    pub async fn list_directory(&self, selection: DirectorySelection) -> Result<DirectoryReader> {
        let context = crate::execution::Context::current();
        let span = context.sync(|| tracing::debug_span!("storage.directory.list_directory", outcome = "cancelled"));
        context.result(span, async { DirectoryReader::new(Arc::clone(&self.inner), selection).await }).await
    }

    pub async fn watch_directory(&self, selection: DirectorySelection, include_existing: bool) -> Result<DirectoryWatcher> {
        let context = crate::execution::Context::current();
        let span = context.sync(|| tracing::debug_span!("storage.directory.watch_directory", outcome = "cancelled"));
        context.result(span, async { DirectoryWatcher::new(Arc::clone(&self.inner), selection, include_existing).await }).await
    }
}

/* === Public Functions === */

/* === Private Functions === */

fn item<T: Message>(inner: &Arc<Inner>, value: T) -> Result<ReadBatch<T>> {
    if value.encoded_size() > inner.settings.response.batch_bytes.get() {
        return Err(Error::ResourceExhausted { resource: "directory response bytes" });
    }
    Ok(ReadBatch::new(value))
}

async fn selection(mut value: DirectorySelection) -> Result<Arc<Selection>> {
    let path = value.root.clone();
    let context = crate::execution::Context::current();
    value.root =
        tokio::task::spawn_blocking(move || context.blocking(|| crate::local::validate_local_path(&path))).await.map_err(join_error)??;

    Ok(Arc::new(Selection { value }))
}

fn relative(root: &Path, path: &Path, recursive: bool) -> Result<Option<String>> {
    let Ok(relative) = path.strip_prefix(root) else { return Ok(None) };
    if relative.as_os_str().is_empty() || (!recursive && relative.components().count() != 1) {
        return Ok(None);
    }
    let mut result = String::new();
    for component in relative.components() {
        let std::path::Component::Normal(value) = component else { return Ok(None) };
        let value = value.to_str().ok_or(Error::InvalidInput {
            field: "directory entry",
            reason: "path cannot be represented as UTF-8",
        })?;
        if !result.is_empty() {
            result.push('/');
        }
        result.push_str(value);
    }
    Ok(Some(result))
}

fn join_error(error: tokio::task::JoinError) -> Error {
    Error::backend("directory task", crate::BackendFailure::Permanent, error)
}

/* === Tests === */