cranpose_services/content/
file.rs

1use std::{
2    cell::RefCell,
3    fs::File,
4    io::{Read, Write},
5    path::{Path, PathBuf},
6    rc::Rc,
7    time::UNIX_EPOCH,
8};
9
10use super::{
11    Content, ContentEntry, ContentError, ContentFolder, ContentFolderRef, ContentFuture,
12    ContentHandle, ContentMetadata, ContentReader, ContentReaderRef, ContentSink, ContentSinkRef,
13    DEFAULT_CHUNK_LEN,
14};
15
16fn io_error(path: &Path, error: std::io::Error) -> ContentError {
17    match error.kind() {
18        std::io::ErrorKind::NotFound => ContentError::NotFound(path.display().to_string()),
19        std::io::ErrorKind::PermissionDenied => {
20            ContentError::PermissionDenied(path.display().to_string())
21        }
22        _ => ContentError::Io(format!("{}: {error}", path.display())),
23    }
24}
25
26fn metadata_for(path: &Path) -> ContentMetadata {
27    let name = path
28        .file_name()
29        .map(|name| name.to_string_lossy().into_owned())
30        .unwrap_or_else(|| path.display().to_string());
31    let mut metadata = ContentMetadata {
32        name,
33        mime_type: None,
34        len: None,
35        modified_millis: None,
36        identifier: path.display().to_string(),
37    };
38    if let Ok(stat) = std::fs::metadata(path) {
39        if stat.is_file() {
40            metadata.len = Some(stat.len());
41        }
42        metadata.modified_millis = stat
43            .modified()
44            .ok()
45            .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
46            .map(|since| since.as_millis() as u64);
47    }
48    metadata
49}
50
51/// A file on the local filesystem.
52pub struct FileContent {
53    path: PathBuf,
54}
55
56impl FileContent {
57    /// Wraps `path` as readable content.
58    pub fn new(path: impl Into<PathBuf>) -> Self {
59        Self { path: path.into() }
60    }
61
62    /// The wrapped path.
63    pub fn path(&self) -> &Path {
64        &self.path
65    }
66}
67
68impl Content for FileContent {
69    fn metadata(&self) -> ContentMetadata {
70        metadata_for(&self.path)
71    }
72
73    fn open(&self) -> ContentFuture<'_, Result<ContentReaderRef, ContentError>> {
74        let path = self.path.clone();
75        Box::pin(async move {
76            let file = File::open(&path).map_err(|error| io_error(&path, error))?;
77            Ok(Rc::new(FileReader {
78                path,
79                file: RefCell::new(Some(file)),
80            }) as ContentReaderRef)
81        })
82    }
83
84    fn read_all(&self) -> ContentFuture<'_, Result<Vec<u8>, ContentError>> {
85        let path = self.path.clone();
86        Box::pin(async move { std::fs::read(&path).map_err(|error| io_error(&path, error)) })
87    }
88}
89
90/// A shared handle to the file at `path`.
91pub fn file_content(path: impl Into<PathBuf>) -> ContentHandle {
92    Rc::new(FileContent::new(path))
93}
94
95struct FileReader {
96    path: PathBuf,
97    file: RefCell<Option<File>>,
98}
99
100impl ContentReader for FileReader {
101    fn read_chunk(&self) -> ContentFuture<'_, Result<Option<Vec<u8>>, ContentError>> {
102        Box::pin(async move {
103            let mut slot = self.file.borrow_mut();
104            let Some(file) = slot.as_mut() else {
105                return Ok(None);
106            };
107            let mut buffer = vec![0u8; DEFAULT_CHUNK_LEN];
108            let mut filled = 0;
109            while filled < buffer.len() {
110                match file.read(&mut buffer[filled..]) {
111                    Ok(0) => break,
112                    Ok(read) => filled += read,
113                    Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
114                    Err(error) => return Err(io_error(&self.path, error)),
115                }
116            }
117            if filled == 0 {
118                *slot = None;
119                return Ok(None);
120            }
121            buffer.truncate(filled);
122            Ok(Some(buffer))
123        })
124    }
125}
126
127/// A directory on the local filesystem.
128pub struct FileFolder {
129    path: PathBuf,
130}
131
132impl FileFolder {
133    /// Wraps `path` as an enumerable folder.
134    pub fn new(path: impl Into<PathBuf>) -> Self {
135        Self { path: path.into() }
136    }
137
138    /// The wrapped path.
139    pub fn path(&self) -> &Path {
140        &self.path
141    }
142}
143
144impl ContentFolder for FileFolder {
145    fn metadata(&self) -> ContentMetadata {
146        metadata_for(&self.path)
147    }
148
149    fn entries(&self) -> ContentFuture<'_, Result<Vec<ContentEntry>, ContentError>> {
150        let path = self.path.clone();
151        Box::pin(async move {
152            let listing = std::fs::read_dir(&path).map_err(|error| io_error(&path, error))?;
153            let mut entries = Vec::new();
154            for child in listing {
155                let child = child.map_err(|error| io_error(&path, error))?;
156                let child_path = child.path();
157                let is_dir = child
158                    .file_type()
159                    .map(|kind| kind.is_dir())
160                    .unwrap_or_else(|_| child_path.is_dir());
161                if is_dir {
162                    entries.push(ContentEntry::Folder(
163                        Rc::new(FileFolder::new(child_path)) as ContentFolderRef
164                    ));
165                } else {
166                    entries.push(ContentEntry::File(
167                        Rc::new(FileContent::new(child_path)) as ContentHandle
168                    ));
169                }
170            }
171            entries.sort_by_key(|entry| entry.metadata().name);
172            Ok(entries)
173        })
174    }
175}
176
177/// A shared handle to the directory at `path`.
178pub fn file_folder(path: impl Into<PathBuf>) -> ContentFolderRef {
179    Rc::new(FileFolder::new(path))
180}
181
182/// A file being written through a temporary sibling, renamed on
183/// [`ContentSink::finish`] so a partial write never replaces a good file.
184pub struct FileSink {
185    destination: PathBuf,
186    staging: PathBuf,
187    file: RefCell<Option<File>>,
188}
189
190impl FileSink {
191    /// Opens a staged writer for `destination`.
192    pub fn create(destination: impl Into<PathBuf>) -> Result<Self, ContentError> {
193        let destination = destination.into();
194        if let Some(parent) = destination.parent() {
195            std::fs::create_dir_all(parent).map_err(|error| io_error(parent, error))?;
196        }
197        let staging = staging_path(&destination);
198        let file = File::create(&staging).map_err(|error| io_error(&staging, error))?;
199        Ok(Self {
200            destination,
201            staging,
202            file: RefCell::new(Some(file)),
203        })
204    }
205
206    /// A shared handle to this sink.
207    pub fn handle(self) -> ContentSinkRef {
208        Rc::new(self)
209    }
210
211    /// The path this sink commits to.
212    pub fn destination(&self) -> &Path {
213        &self.destination
214    }
215}
216
217fn staging_path(destination: &Path) -> PathBuf {
218    let mut name = destination
219        .file_name()
220        .map(|name| name.to_string_lossy().into_owned())
221        .unwrap_or_else(|| "content".to_string());
222    name.push_str(".partial");
223    destination.with_file_name(name)
224}
225
226impl ContentSink for FileSink {
227    fn write_chunk(&self, bytes: Vec<u8>) -> ContentFuture<'_, Result<(), ContentError>> {
228        Box::pin(async move {
229            let mut slot = self.file.borrow_mut();
230            let file = slot
231                .as_mut()
232                .ok_or_else(|| ContentError::Io("sink is already finished".into()))?;
233            file.write_all(&bytes)
234                .map_err(|error| io_error(&self.staging, error))
235        })
236    }
237
238    fn finish(&self) -> ContentFuture<'_, Result<(), ContentError>> {
239        Box::pin(async move {
240            let Some(mut file) = self.file.borrow_mut().take() else {
241                return Ok(());
242            };
243            file.flush()
244                .map_err(|error| io_error(&self.staging, error))?;
245            file.sync_all()
246                .map_err(|error| io_error(&self.staging, error))?;
247            drop(file);
248            std::fs::rename(&self.staging, &self.destination)
249                .map_err(|error| io_error(&self.destination, error))
250        })
251    }
252}
253
254impl Drop for FileSink {
255    fn drop(&mut self) {
256        if self.file.borrow().is_some() {
257            let _ = std::fs::remove_file(&self.staging);
258        }
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265    use crate::content::{DEFAULT_CHUNK_LEN, collect_stream, folder_files, write_all};
266
267    fn temp_dir(name: &str) -> PathBuf {
268        crate::test_scratch_dir(&format!("content-{name}"))
269    }
270
271    #[test]
272    fn a_file_streams_in_chunks_and_reports_its_length() {
273        let root = temp_dir("chunks");
274        let path = root.join("payload.bin");
275        let payload = vec![3u8; DEFAULT_CHUNK_LEN + 5];
276        std::fs::write(&path, &payload).unwrap();
277
278        let content = file_content(&path);
279        assert_eq!(content.metadata().name, "payload.bin");
280        assert_eq!(content.metadata().len, Some(payload.len() as u64));
281
282        let sizes = pollster::block_on(async {
283            let reader = content.open().await.unwrap();
284            let mut sizes = Vec::new();
285            while let Some(chunk) = reader.read_chunk().await.unwrap() {
286                sizes.push(chunk.len());
287            }
288            sizes
289        });
290        assert_eq!(sizes, vec![DEFAULT_CHUNK_LEN, 5]);
291        assert_eq!(pollster::block_on(content.read_all()).unwrap(), payload);
292        std::fs::remove_dir_all(&root).unwrap();
293    }
294
295    #[test]
296    fn a_missing_file_reports_not_found() {
297        let root = temp_dir("missing");
298        let content = file_content(root.join("absent.bin"));
299        assert!(matches!(
300            pollster::block_on(content.read_all()),
301            Err(ContentError::NotFound(_))
302        ));
303        std::fs::remove_dir_all(&root).unwrap();
304    }
305
306    #[test]
307    fn a_folder_streams_its_whole_tree() {
308        let root = temp_dir("tree");
309        std::fs::create_dir_all(root.join("nested")).unwrap();
310        std::fs::write(root.join("a.txt"), b"a").unwrap();
311        std::fs::write(root.join("nested/b.txt"), b"b").unwrap();
312
313        let stream = folder_files(file_folder(&root));
314        let mut names: Vec<String> = pollster::block_on(collect_stream(&stream))
315            .unwrap()
316            .iter()
317            .map(|file| file.metadata().name)
318            .collect();
319        names.sort();
320        assert_eq!(names, vec!["a.txt", "b.txt"]);
321        std::fs::remove_dir_all(&root).unwrap();
322    }
323
324    #[test]
325    fn a_sink_commits_atomically_and_discards_unfinished_writes() {
326        let root = temp_dir("sink");
327        let destination = root.join("out/report.bin");
328
329        let sink = FileSink::create(&destination).unwrap().handle();
330        pollster::block_on(write_all(&sink, b"committed".to_vec())).unwrap();
331        assert_eq!(std::fs::read(&destination).unwrap(), b"committed");
332
333        let abandoned = root.join("out/abandoned.bin");
334        {
335            let sink = FileSink::create(&abandoned).unwrap();
336            pollster::block_on(sink.write_chunk(b"partial".to_vec())).unwrap();
337        }
338        assert!(!abandoned.exists());
339        assert!(!staging_path(&abandoned).exists());
340        std::fs::remove_dir_all(&root).unwrap();
341    }
342}
cranpose_services/content/file.rs

cranpose_services/content/
file.rs
