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.file_name().map_or_else(
28        || path.display().to_string(),
29        |name| name.to_string_lossy().into_owned(),
30    );
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_or_else(|_| child_path.is_dir(), |kind| kind.is_dir());
160                if is_dir {
161                    entries.push(ContentEntry::Folder(
162                        Rc::new(FileFolder::new(child_path)) as ContentFolderRef
163                    ));
164                } else {
165                    entries.push(ContentEntry::File(
166                        Rc::new(FileContent::new(child_path)) as ContentHandle
167                    ));
168                }
169            }
170            entries.sort_by_key(|entry| entry.metadata().name);
171            Ok(entries)
172        })
173    }
174}
175
176/// A shared handle to the directory at `path`.
177pub fn file_folder(path: impl Into<PathBuf>) -> ContentFolderRef {
178    Rc::new(FileFolder::new(path))
179}
180
181/// A file being written through a temporary sibling, renamed on
182/// [`ContentSink::finish`] so a partial write never replaces a good file.
183pub struct FileSink {
184    destination: PathBuf,
185    staging: PathBuf,
186    file: RefCell<Option<File>>,
187}
188
189impl FileSink {
190    /// Opens a staged writer for `destination`.
191    pub fn create(destination: impl Into<PathBuf>) -> Result<Self, ContentError> {
192        let destination = destination.into();
193        if let Some(parent) = destination.parent() {
194            std::fs::create_dir_all(parent).map_err(|error| io_error(parent, error))?;
195        }
196        let staging = staging_path(&destination);
197        let file = File::create(&staging).map_err(|error| io_error(&staging, error))?;
198        Ok(Self {
199            destination,
200            staging,
201            file: RefCell::new(Some(file)),
202        })
203    }
204
205    /// A shared handle to this sink.
206    pub fn handle(self) -> ContentSinkRef {
207        Rc::new(self)
208    }
209
210    /// The path this sink commits to.
211    pub fn destination(&self) -> &Path {
212        &self.destination
213    }
214}
215
216fn staging_path(destination: &Path) -> PathBuf {
217    let mut name = destination.file_name().map_or_else(
218        || "content".to_string(),
219        |name| name.to_string_lossy().into_owned(),
220    );
221    name.push_str(".partial");
222    destination.with_file_name(name)
223}
224
225impl ContentSink for FileSink {
226    fn write_chunk(&self, bytes: Vec<u8>) -> ContentFuture<'_, Result<(), ContentError>> {
227        Box::pin(async move {
228            let mut slot = self.file.borrow_mut();
229            let file = slot
230                .as_mut()
231                .ok_or_else(|| ContentError::Io("sink is already finished".into()))?;
232            file.write_all(&bytes)
233                .map_err(|error| io_error(&self.staging, error))
234        })
235    }
236
237    fn finish(&self) -> ContentFuture<'_, Result<(), ContentError>> {
238        Box::pin(async move {
239            let Some(mut file) = self.file.borrow_mut().take() else {
240                return Ok(());
241            };
242            file.flush()
243                .map_err(|error| io_error(&self.staging, error))?;
244            file.sync_all()
245                .map_err(|error| io_error(&self.staging, error))?;
246            drop(file);
247            std::fs::rename(&self.staging, &self.destination)
248                .map_err(|error| io_error(&self.destination, error))
249        })
250    }
251}
252
253impl Drop for FileSink {
254    fn drop(&mut self) {
255        if self.file.borrow().is_some() {
256            let _ = std::fs::remove_file(&self.staging);
257        }
258    }
259}
260
261#[cfg(test)]
262mod tests {
263    use super::*;
264    use crate::content::{DEFAULT_CHUNK_LEN, collect_stream, folder_files, write_all};
265
266    fn temp_dir(name: &str) -> PathBuf {
267        crate::test_scratch_dir(&format!("content-{name}"))
268    }
269
270    #[test]
271    fn a_file_streams_in_chunks_and_reports_its_length() {
272        let root = temp_dir("chunks");
273        let path = root.join("payload.bin");
274        let payload = vec![3u8; DEFAULT_CHUNK_LEN + 5];
275        std::fs::write(&path, &payload).unwrap();
276
277        let content = file_content(&path);
278        assert_eq!(content.metadata().name, "payload.bin");
279        assert_eq!(content.metadata().len, Some(payload.len() as u64));
280
281        let sizes = pollster::block_on(async {
282            let reader = content.open().await.unwrap();
283            let mut sizes = Vec::new();
284            while let Some(chunk) = reader.read_chunk().await.unwrap() {
285                sizes.push(chunk.len());
286            }
287            sizes
288        });
289        assert_eq!(sizes, vec![DEFAULT_CHUNK_LEN, 5]);
290        assert_eq!(pollster::block_on(content.read_all()).unwrap(), payload);
291        std::fs::remove_dir_all(&root).unwrap();
292    }
293
294    #[test]
295    fn a_missing_file_reports_not_found() {
296        let root = temp_dir("missing");
297        let content = file_content(root.join("absent.bin"));
298        assert!(matches!(
299            pollster::block_on(content.read_all()),
300            Err(ContentError::NotFound(_))
301        ));
302        std::fs::remove_dir_all(&root).unwrap();
303    }
304
305    #[test]
306    fn a_folder_streams_its_whole_tree() {
307        let root = temp_dir("tree");
308        std::fs::create_dir_all(root.join("nested")).unwrap();
309        std::fs::write(root.join("a.txt"), b"a").unwrap();
310        std::fs::write(root.join("nested/b.txt"), b"b").unwrap();
311
312        let stream = folder_files(file_folder(&root));
313        let mut names: Vec<String> = pollster::block_on(collect_stream(&stream))
314            .unwrap()
315            .iter()
316            .map(|file| file.metadata().name)
317            .collect();
318        names.sort();
319        assert_eq!(names, vec!["a.txt", "b.txt"]);
320        std::fs::remove_dir_all(&root).unwrap();
321    }
322
323    #[test]
324    fn a_sink_commits_atomically_and_discards_unfinished_writes() {
325        let root = temp_dir("sink");
326        let destination = root.join("out/report.bin");
327
328        let sink = FileSink::create(&destination).unwrap().handle();
329        pollster::block_on(write_all(&sink, b"committed".to_vec())).unwrap();
330        assert_eq!(std::fs::read(&destination).unwrap(), b"committed");
331
332        let abandoned = root.join("out/abandoned.bin");
333        {
334            let sink = FileSink::create(&abandoned).unwrap();
335            pollster::block_on(sink.write_chunk(b"partial".to_vec())).unwrap();
336        }
337        assert!(!abandoned.exists());
338        assert!(!staging_path(&abandoned).exists());
339        std::fs::remove_dir_all(&root).unwrap();
340    }
341}
cranpose_services/content/file.rs

cranpose_services/content/
file.rs
