harmos_storage/directory/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Charged Directory Discovery And Native Change Subscription
11
12/* === Modules === */
13
14mod scan;
15mod selection;
16mod watch;
17
18/* === Re-Exports === */
19
20pub use scan::DirectoryReader;
21pub use watch::DirectoryWatcher;
22
23/* === Imports === */
24
25// Standard Library Imports
26use std::path::Path;
27use std::sync::Arc;
28// Third Party Imports
29// Local Imports
30use crate::manager::Inner;
31use crate::{Error, Message, ReadBatch, Result, StorageManager};
32use harmos_domain::model::storage::{DirectoryCompletion, DirectoryEntry, DirectoryEvent, DirectorySelection};
33
34/* === Constants === */
35
36/* === Macros === */
37
38/* === Type Aliases === */
39
40/* === Structs === */
41
42struct Selection {
43    value: DirectorySelection,
44}
45
46/* === Enums === */
47
48/* === Traits === */
49
50/* === Implementations === */
51
52impl<C> StorageManager<C> {
53    pub async fn list_directory(&self, selection: DirectorySelection) -> Result<DirectoryReader> {
54        let context = crate::execution::Context::current();
55        let span = context.sync(|| tracing::debug_span!("storage.directory.list_directory", outcome = "cancelled"));
56        context.result(span, async { DirectoryReader::new(Arc::clone(&self.inner), selection).await }).await
57    }
58
59    pub async fn watch_directory(&self, selection: DirectorySelection, include_existing: bool) -> Result<DirectoryWatcher> {
60        let context = crate::execution::Context::current();
61        let span = context.sync(|| tracing::debug_span!("storage.directory.watch_directory", outcome = "cancelled"));
62        context.result(span, async { DirectoryWatcher::new(Arc::clone(&self.inner), selection, include_existing).await }).await
63    }
64}
65
66/* === Public Functions === */
67
68/* === Private Functions === */
69
70fn item<T: Message>(inner: &Arc<Inner>, value: T) -> Result<ReadBatch<T>> {
71    if value.encoded_size() > inner.settings.response.batch_bytes.get() {
72        return Err(Error::ResourceExhausted { resource: "directory response bytes" });
73    }
74    Ok(ReadBatch::new(value))
75}
76
77async fn selection(mut value: DirectorySelection) -> Result<Arc<Selection>> {
78    let path = value.root.clone();
79    let context = crate::execution::Context::current();
80    value.root =
81        tokio::task::spawn_blocking(move || context.blocking(|| crate::local::validate_local_path(&path))).await.map_err(join_error)??;
82
83    Ok(Arc::new(Selection { value }))
84}
85
86fn relative(root: &Path, path: &Path, recursive: bool) -> Result<Option<String>> {
87    let Ok(relative) = path.strip_prefix(root) else { return Ok(None) };
88    if relative.as_os_str().is_empty() || (!recursive && relative.components().count() != 1) {
89        return Ok(None);
90    }
91    let mut result = String::new();
92    for component in relative.components() {
93        let std::path::Component::Normal(value) = component else { return Ok(None) };
94        let value = value.to_str().ok_or(Error::InvalidInput {
95            field: "directory entry",
96            reason: "path cannot be represented as UTF-8",
97        })?;
98        if !result.is_empty() {
99            result.push('/');
100        }
101        result.push_str(value);
102    }
103    Ok(Some(result))
104}
105
106fn join_error(error: tokio::task::JoinError) -> Error {
107    Error::backend("directory task", crate::BackendFailure::Permanent, error)
108}
109
110/* === Tests === */
harmos_storage/directory/mod.rs

harmos_storage/directory/
mod.rs
