use super::{
Content, ContentEntry, ContentError, ContentFolder, ContentFolderRef, ContentFuture,
ContentHandle, ContentMetadata, ContentReader, ContentReaderRef, ContentSink, ContentSinkRef,
DEFAULT_CHUNK_LEN,
};
use std::cell::RefCell;
use std::fs::File;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::rc::Rc;
use std::time::UNIX_EPOCH;
fn io_error(path: &Path, error: std::io::Error) -> ContentError {
match error.kind() {
std::io::ErrorKind::NotFound => ContentError::NotFound(path.display().to_string()),
std::io::ErrorKind::PermissionDenied => {
ContentError::PermissionDenied(path.display().to_string())
}
_ => ContentError::Io(format!("{}: {error}", path.display())),
}
}
fn metadata_for(path: &Path) -> ContentMetadata {
let name = path
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_else(|| path.display().to_string());
let mut metadata = ContentMetadata {
name,
mime_type: None,
len: None,
modified_millis: None,
identifier: path.display().to_string(),
};
if let Ok(stat) = std::fs::metadata(path) {
if stat.is_file() {
metadata.len = Some(stat.len());
}
metadata.modified_millis = stat
.modified()
.ok()
.and_then(|time| time.duration_since(UNIX_EPOCH).ok())
.map(|since| since.as_millis() as u64);
}
metadata
}
pub struct FileContent {
path: PathBuf,
}
impl FileContent {
pub fn new(path: impl Into<PathBuf>) -> Self {
Self { path: path.into() }
}
pub fn path(&self) -> &Path {
&self.path
}
}
impl Content for FileContent {
fn metadata(&self) -> ContentMetadata {
metadata_for(&self.path)
}
fn open(&self) -> ContentFuture<'_, Result<ContentReaderRef, ContentError>> {
let path = self.path.clone();
Box::pin(async move {
let file = File::open(&path).map_err(|error| io_error(&path, error))?;
Ok(Rc::new(FileReader {
path,
file: RefCell::new(Some(file)),
}) as ContentReaderRef)
})
}
fn read_all(&self) -> ContentFuture<'_, Result<Vec<u8>, ContentError>> {
let path = self.path.clone();
Box::pin(async move { std::fs::read(&path).map_err(|error| io_error(&path, error)) })
}
}
pub fn file_content(path: impl Into<PathBuf>) -> ContentHandle {
Rc::new(FileContent::new(path))
}
struct FileReader {
path: PathBuf,
file: RefCell<Option<File>>,
}
impl ContentReader for FileReader {
fn read_chunk(&self) -> ContentFuture<'_, Result<Option<Vec<u8>>, ContentError>> {
Box::pin(async move {
let mut slot = self.file.borrow_mut();
let Some(file) = slot.as_mut() else {
return Ok(None);
};
let mut buffer = vec![0u8; DEFAULT_CHUNK_LEN];
let mut filled = 0;
while filled < buffer.len() {
match file.read(&mut buffer[filled..]) {
Ok(0) => break,
Ok(read) => filled += read,
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(error) => return Err(io_error(&self.path, error)),
}
}
if filled == 0 {
*slot = None;
return Ok(None);
}
buffer.truncate(filled);
Ok(Some(buffer))
})
}
}
pub struct FileFolder {
path: PathBuf,
}
impl FileFolder {
pub fn new(path: impl Into<PathBuf>) -> Self {
Self { path: path.into() }
}
pub fn path(&self) -> &Path {
&self.path
}
}
impl ContentFolder for FileFolder {
fn metadata(&self) -> ContentMetadata {
metadata_for(&self.path)
}
fn entries(&self) -> ContentFuture<'_, Result<Vec<ContentEntry>, ContentError>> {
let path = self.path.clone();
Box::pin(async move {
let listing = std::fs::read_dir(&path).map_err(|error| io_error(&path, error))?;
let mut entries = Vec::new();
for child in listing {
let child = child.map_err(|error| io_error(&path, error))?;
let child_path = child.path();
let is_dir = child
.file_type()
.map(|kind| kind.is_dir())
.unwrap_or_else(|_| child_path.is_dir());
if is_dir {
entries.push(ContentEntry::Folder(
Rc::new(FileFolder::new(child_path)) as ContentFolderRef
));
} else {
entries.push(ContentEntry::File(
Rc::new(FileContent::new(child_path)) as ContentHandle
));
}
}
entries.sort_by_key(|entry| entry.metadata().name);
Ok(entries)
})
}
}
pub fn file_folder(path: impl Into<PathBuf>) -> ContentFolderRef {
Rc::new(FileFolder::new(path))
}
pub struct FileSink {
destination: PathBuf,
staging: PathBuf,
file: RefCell<Option<File>>,
}
impl FileSink {
pub fn create(destination: impl Into<PathBuf>) -> Result<Self, ContentError> {
let destination = destination.into();
if let Some(parent) = destination.parent() {
std::fs::create_dir_all(parent).map_err(|error| io_error(parent, error))?;
}
let staging = staging_path(&destination);
let file = File::create(&staging).map_err(|error| io_error(&staging, error))?;
Ok(Self {
destination,
staging,
file: RefCell::new(Some(file)),
})
}
pub fn handle(self) -> ContentSinkRef {
Rc::new(self)
}
pub fn destination(&self) -> &Path {
&self.destination
}
}
fn staging_path(destination: &Path) -> PathBuf {
let mut name = destination
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_else(|| "content".to_string());
name.push_str(".partial");
destination.with_file_name(name)
}
impl ContentSink for FileSink {
fn write_chunk(&self, bytes: Vec<u8>) -> ContentFuture<'_, Result<(), ContentError>> {
Box::pin(async move {
let mut slot = self.file.borrow_mut();
let file = slot
.as_mut()
.ok_or_else(|| ContentError::Io("sink is already finished".into()))?;
file.write_all(&bytes)
.map_err(|error| io_error(&self.staging, error))
})
}
fn finish(&self) -> ContentFuture<'_, Result<(), ContentError>> {
Box::pin(async move {
let Some(mut file) = self.file.borrow_mut().take() else {
return Ok(());
};
file.flush()
.map_err(|error| io_error(&self.staging, error))?;
file.sync_all()
.map_err(|error| io_error(&self.staging, error))?;
drop(file);
std::fs::rename(&self.staging, &self.destination)
.map_err(|error| io_error(&self.destination, error))
})
}
}
impl Drop for FileSink {
fn drop(&mut self) {
if self.file.borrow().is_some() {
let _ = std::fs::remove_file(&self.staging);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::content::{collect_stream, folder_files, write_all, DEFAULT_CHUNK_LEN};
fn temp_dir(name: &str) -> PathBuf {
crate::test_scratch_dir(&format!("content-{name}"))
}
#[test]
fn a_file_streams_in_chunks_and_reports_its_length() {
let root = temp_dir("chunks");
let path = root.join("payload.bin");
let payload = vec![3u8; DEFAULT_CHUNK_LEN + 5];
std::fs::write(&path, &payload).unwrap();
let content = file_content(&path);
assert_eq!(content.metadata().name, "payload.bin");
assert_eq!(content.metadata().len, Some(payload.len() as u64));
let sizes = pollster::block_on(async {
let reader = content.open().await.unwrap();
let mut sizes = Vec::new();
while let Some(chunk) = reader.read_chunk().await.unwrap() {
sizes.push(chunk.len());
}
sizes
});
assert_eq!(sizes, vec![DEFAULT_CHUNK_LEN, 5]);
assert_eq!(pollster::block_on(content.read_all()).unwrap(), payload);
std::fs::remove_dir_all(&root).unwrap();
}
#[test]
fn a_missing_file_reports_not_found() {
let root = temp_dir("missing");
let content = file_content(root.join("absent.bin"));
assert!(matches!(
pollster::block_on(content.read_all()),
Err(ContentError::NotFound(_))
));
std::fs::remove_dir_all(&root).unwrap();
}
#[test]
fn a_folder_streams_its_whole_tree() {
let root = temp_dir("tree");
std::fs::create_dir_all(root.join("nested")).unwrap();
std::fs::write(root.join("a.txt"), b"a").unwrap();
std::fs::write(root.join("nested/b.txt"), b"b").unwrap();
let stream = folder_files(file_folder(&root));
let mut names: Vec<String> = pollster::block_on(collect_stream(&stream))
.unwrap()
.iter()
.map(|file| file.metadata().name)
.collect();
names.sort();
assert_eq!(names, vec!["a.txt", "b.txt"]);
std::fs::remove_dir_all(&root).unwrap();
}
#[test]
fn a_sink_commits_atomically_and_discards_unfinished_writes() {
let root = temp_dir("sink");
let destination = root.join("out/report.bin");
let sink = FileSink::create(&destination).unwrap().handle();
pollster::block_on(write_all(&sink, b"committed".to_vec())).unwrap();
assert_eq!(std::fs::read(&destination).unwrap(), b"committed");
let abandoned = root.join("out/abandoned.bin");
{
let sink = FileSink::create(&abandoned).unwrap();
pollster::block_on(sink.write_chunk(b"partial".to_vec())).unwrap();
}
assert!(!abandoned.exists());
assert!(!staging_path(&abandoned).exists());
std::fs::remove_dir_all(&root).unwrap();
}
}