kellnr_storage/
fs_storage.rs

1use std::fs::DirBuilder;
2
3use async_trait::async_trait;
4use bytes::Bytes;
5use futures_util::StreamExt;
6use object_store::local::LocalFileSystem;
7use object_store::path::Path;
8use object_store::{ObjectStore, ObjectStoreExt, PutMode};
9
10use crate::storage::{ObjectWithMeta, Storage};
11use crate::storage_error::StorageError;
12
13pub struct FSStorage(LocalFileSystem);
14
15#[async_trait]
16impl Storage for FSStorage {
17    async fn get(&self, key: &str) -> Result<Bytes, StorageError> {
18        self.storage()
19            .get(&Path::from(key))
20            .await?
21            .bytes()
22            .await
23            .map_err(StorageError::from)
24    }
25
26    async fn get_with_meta(&self, key: &str) -> Result<ObjectWithMeta, StorageError> {
27        let result = self.storage().get(&Path::from(key)).await?;
28        let e_tag = result.meta.e_tag.clone();
29        let last_modified = result.meta.last_modified;
30        let bytes = result.bytes().await?;
31        Ok(ObjectWithMeta {
32            bytes,
33            e_tag,
34            last_modified,
35        })
36    }
37
38    async fn put(&self, key: &str, object: Bytes) -> Result<(), StorageError> {
39        self.storage()
40            .put_opts(&Path::from(key), object.into(), PutMode::Create.into())
41            .await?;
42        Ok(())
43    }
44
45    async fn put_overwrite(&self, key: &str, object: Bytes) -> Result<(), StorageError> {
46        self.storage()
47            .put_opts(&Path::from(key), object.into(), PutMode::Overwrite.into())
48            .await?;
49        Ok(())
50    }
51
52    async fn delete(&self, key: &str) -> Result<(), StorageError> {
53        self.storage().delete(&Path::from(key)).await?;
54        Ok(())
55    }
56
57    async fn exists(&self, key: &str) -> Result<bool, StorageError> {
58        self.storage()
59            .head(&Path::from(key))
60            .await
61            .map(|_| true)
62            .or_else(|e| match e {
63                object_store::Error::NotFound { .. } => Ok(false),
64                _ => Err(StorageError::from(e)),
65            })
66    }
67
68    async fn size(&self, key: &str) -> Result<u64, StorageError> {
69        Ok(self.storage().head(&Path::from(key)).await?.size)
70    }
71
72    async fn list(&self, prefix: &str) -> Result<Vec<String>, StorageError> {
73        let prefix = if prefix.is_empty() {
74            None
75        } else {
76            Some(Path::from(prefix))
77        };
78        let mut stream = self.storage().list(prefix.as_ref());
79        let mut keys = Vec::new();
80        while let Some(meta) = stream.next().await {
81            keys.push(meta?.location.to_string());
82        }
83        Ok(keys)
84    }
85}
86
87impl FSStorage {
88    pub fn new(crate_folder: &str) -> Result<Self, StorageError> {
89        let path = std::path::Path::new(crate_folder);
90        if !path.exists() {
91            DirBuilder::new()
92                .recursive(true)
93                .create(crate_folder)
94                .map_err(|e| StorageError::CreateBinPath(path.to_path_buf(), e))?;
95        }
96        let client = LocalFileSystem::new_with_prefix(path)?;
97        Ok(Self(client))
98    }
99
100    fn storage(&self) -> &LocalFileSystem {
101        &self.0
102    }
103}
kellnr_storage/fs_storage.rs

kellnr_storage/
fs_storage.rs
