cranpose_services/content/
file.rs

1mod folder_reference;
2mod stream;
3
4use std::{
5    fs::File,
6    io::Write,
7    path::{Path, PathBuf},
8    rc::Rc,
9    sync::Arc,
10    time::UNIX_EPOCH,
11};
12
13use super::{
14    Content, ContentEntry, ContentError, ContentFolder, ContentFolderRef, ContentFuture,
15    ContentHandle, ContentMetadata, ContentReaderRef, ContentSinkRef, ContentStreamRef,
16    threaded::{BlockingSink, open_threaded, run_threaded, threaded_sink},
17};
18
19fn io_error(path: &Path, error: std::io::Error) -> ContentError {
20    match error.kind() {
21        std::io::ErrorKind::NotFound => ContentError::NotFound(path.display().to_string()),
22        std::io::ErrorKind::PermissionDenied => {
23            ContentError::PermissionDenied(path.display().to_string())
24        }
25        _ => ContentError::Io(format!("{}: {error}", path.display())),
26    }
27}
28
29fn metadata_for(path: &Path) -> ContentMetadata {
30    let name = path.file_name().map_or_else(
31        || path.display().to_string(),
32        |name| name.to_string_lossy().into_owned(),
33    );
34    let mut metadata = ContentMetadata {
35        name,
36        mime_type: None,
37        len: None,
38        modified_millis: None,
39        identifier: path.display().to_string(),
40    };
41    if let Ok(stat) = std::fs::metadata(path) {
42        if stat.is_file() {
43            metadata.len = Some(stat.len());
44        }
45        metadata.modified_millis = stat
46            .modified()
47            .ok()
48            .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
49            .map(|since| since.as_millis() as u64);
50    }
51    metadata
52}
53
54struct PathInfo {
55    path: PathBuf,
56    metadata: Option<ContentMetadata>,
57}
58
59impl PathInfo {
60    fn new(path: impl Into<PathBuf>) -> Self {
61        Self {
62            path: path.into(),
63            metadata: None,
64        }
65    }
66
67    fn metadata(&self) -> ContentMetadata {
68        self.metadata
69            .clone()
70            .unwrap_or_else(|| metadata_for(&self.path))
71    }
72}
73
74/// A file on the local filesystem.
75pub struct FileContent {
76    source: PathInfo,
77    scope: Option<Arc<folder_reference::FolderReference>>,
78}
79
80impl FileContent {
81    /// Wraps `path` as readable content.
82    pub fn new(path: impl Into<PathBuf>) -> Self {
83        Self {
84            source: PathInfo::new(path),
85            scope: None,
86        }
87    }
88
89    /// The wrapped path.
90    pub fn path(&self) -> &Path {
91        &self.source.path
92    }
93}
94
95impl Content for FileContent {
96    fn metadata(&self) -> ContentMetadata {
97        self.source.metadata()
98    }
99
100    fn media_uri(&self) -> Result<String, ContentError> {
101        if let Some(scope) = &self.scope {
102            return scope.reference(&self.source.path);
103        }
104        crate::media_source::media_reference(&self.source.path)
105            .map_err(|error| io_error(&self.source.path, error))
106    }
107
108    fn media_uri_async(&self) -> ContentFuture<'_, Result<String, ContentError>> {
109        let path = self.source.path.clone();
110        let scope = self.scope.clone();
111        run_threaded(move || match scope {
112            Some(scope) => scope.reference(&path),
113            None => {
114                crate::media_source::media_reference(&path).map_err(|error| io_error(&path, error))
115            }
116        })
117    }
118
119    fn open(&self) -> ContentFuture<'_, Result<ContentReaderRef, ContentError>> {
120        let opened = self.source.path.clone();
121        let path = self.source.path.clone();
122        open_threaded(
123            move || File::open(&opened).map_err(|error| io_error(&opened, error)),
124            move |error| io_error(&path, error),
125        )
126    }
127
128    fn read_all(&self) -> ContentFuture<'_, Result<Vec<u8>, ContentError>> {
129        let path = self.source.path.clone();
130        run_threaded(move || std::fs::read(&path).map_err(|error| io_error(&path, error)))
131    }
132}
133
134/// A shared handle to the file at `path`.
135pub fn file_content(path: impl Into<PathBuf>) -> ContentHandle {
136    Rc::new(FileContent::new(path))
137}
138
139/// A directory on the local filesystem.
140pub struct FileFolder {
141    root: PathInfo,
142}
143
144impl FileFolder {
145    /// Wraps `path` as an enumerable folder.
146    pub fn new(path: impl Into<PathBuf>) -> Self {
147        Self {
148            root: PathInfo::new(path),
149        }
150    }
151
152    /// The wrapped path.
153    pub fn path(&self) -> &Path {
154        &self.root.path
155    }
156}
157
158impl ContentFolder for FileFolder {
159    fn metadata(&self) -> ContentMetadata {
160        self.root.metadata()
161    }
162
163    fn entries(&self) -> ContentFuture<'_, Result<Vec<ContentEntry>, ContentError>> {
164        let path = self.root.path.clone();
165        Box::pin(async move {
166            let children = run_threaded(move || {
167                let listing = std::fs::read_dir(&path).map_err(|error| io_error(&path, error))?;
168                let mut entries = Vec::new();
169                for child in listing {
170                    let child = child.map_err(|error| io_error(&path, error))?;
171                    let child_path = child.path();
172                    let kind = child
173                        .file_type()
174                        .map_err(|error| io_error(&child_path, error))?;
175                    let metadata = metadata_for(&child_path);
176                    entries.push((child_path, kind.is_dir(), metadata));
177                }
178                entries.sort_by(|a, b| a.2.name.cmp(&b.2.name));
179                Ok(entries)
180            })
181            .await?;
182            Ok(children
183                .into_iter()
184                .map(|(path, directory, metadata)| {
185                    if directory {
186                        ContentEntry::Folder(Rc::new(Self {
187                            root: PathInfo {
188                                path,
189                                metadata: Some(metadata),
190                            },
191                        }))
192                    } else {
193                        ContentEntry::File(Rc::new(FileContent {
194                            scope: None,
195                            source: PathInfo {
196                                path,
197                                metadata: Some(metadata),
198                            },
199                        }))
200                    }
201                })
202                .collect())
203        })
204    }
205
206    fn stream_files(&self) -> Option<ContentStreamRef> {
207        Some(stream::open(self.root.path.clone()))
208    }
209}
210
211/// A shared handle to the directory at `path`.
212pub fn file_folder(path: impl Into<PathBuf>) -> ContentFolderRef {
213    Rc::new(FileFolder::new(path))
214}
215
216/// A file being written through a temporary sibling, renamed on
217/// [`ContentSink::finish`](super::ContentSink::finish) so a partial write
218/// never replaces a good file. Its [`handle`](FileSink::handle) writes on a
219/// thread of its own.
220pub struct FileSink {
221    destination: PathBuf,
222    staging: PathBuf,
223    file: Option<File>,
224}
225
226impl FileSink {
227    /// Opens a staged writer for `destination`.
228    pub fn create(destination: impl Into<PathBuf>) -> Result<Self, ContentError> {
229        let destination = destination.into();
230        if let Some(parent) = destination.parent() {
231            std::fs::create_dir_all(parent).map_err(|error| io_error(parent, error))?;
232        }
233        let staging = staging_path(&destination);
234        let file = File::create(&staging).map_err(|error| io_error(&staging, error))?;
235        Ok(Self {
236            destination,
237            staging,
238            file: Some(file),
239        })
240    }
241
242    /// A shared handle to this sink, which writes on a thread of its own.
243    pub fn handle(self) -> ContentSinkRef {
244        threaded_sink(self)
245    }
246
247    /// The path this sink commits to.
248    pub fn destination(&self) -> &Path {
249        &self.destination
250    }
251}
252
253fn staging_path(destination: &Path) -> PathBuf {
254    let mut name = destination.file_name().map_or_else(
255        || "content".to_string(),
256        |name| name.to_string_lossy().into_owned(),
257    );
258    name.push_str(".partial");
259    destination.with_file_name(name)
260}
261
262impl BlockingSink for FileSink {
263    fn write(&mut self, bytes: &[u8]) -> Result<(), ContentError> {
264        let file = self
265            .file
266            .as_mut()
267            .ok_or_else(|| ContentError::Io("sink is already finished".into()))?;
268        file.write_all(bytes)
269            .map_err(|error| io_error(&self.staging, error))
270    }
271
272    fn finish(&mut self) -> Result<(), ContentError> {
273        let Some(mut file) = self.file.take() else {
274            return Ok(());
275        };
276        file.flush()
277            .map_err(|error| io_error(&self.staging, error))?;
278        file.sync_all()
279            .map_err(|error| io_error(&self.staging, error))?;
280        drop(file);
281        std::fs::rename(&self.staging, &self.destination)
282            .map_err(|error| io_error(&self.destination, error))
283    }
284}
285
286impl Drop for FileSink {
287    fn drop(&mut self) {
288        if self.file.is_some() {
289            let _ = std::fs::remove_file(&self.staging);
290        }
291    }
292}
293
294#[cfg(test)]
295#[path = "tests/file_tests.rs"]
296mod tests;
cranpose_services/content/file.rs

cranpose_services/content/
file.rs
