|
|
|
//! Handle deletion
|
|
|
|
//!
|
|
|
|
//! # Handling paths
|
|
|
|
//! Deletion of single files is trivial, deletion of directories is not as trivial.
|
|
|
|
//! Other types of filesystem object are ignored.
|
|
|
|
//!
|
|
|
|
//! ## Directories
|
|
|
|
//! With directories the deletion processes all files containing recursively.
|
|
|
|
//! See `restore` for restoration of directories
|
|
|
|
use super::*;
|
|
|
|
use std::{
|
|
|
|
path::{
|
|
|
|
Path,
|
|
|
|
},
|
|
|
|
};
|
|
|
|
use tokio::{
|
|
|
|
fs,
|
|
|
|
};
|
|
|
|
use futures::{
|
|
|
|
prelude::*,
|
|
|
|
future::BoxFuture,
|
|
|
|
};
|
|
|
|
|
|
|
|
/// Process a single file.
|
|
|
|
///
|
|
|
|
/// `path` is known to be a file at this point.
|
|
|
|
async fn process_single(state: Arc<state::State>, path: impl AsRef<Path>) -> eyre::Result<()>
|
|
|
|
{
|
|
|
|
let path = path.as_ref();
|
|
|
|
#[cfg(debug_assertions)] {
|
|
|
|
if !path.is_file() {
|
|
|
|
panic!("process_single() expected a file, but {:?} is not one.", path);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
let _g = state.lock().await;
|
|
|
|
debug!("{:?} Processing", path);
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
fn walk_dir<'a, T, F, P, Fut>(path: P, mut output: F) -> BoxFuture<'static, eyre::Result<Vec<T>>>
|
|
|
|
where F: FnMut(fs::DirEntry) -> Fut + Clone + 'static + Send + Sync,
|
|
|
|
P: AsRef<Path> + Send + Sync + 'static,
|
|
|
|
T: Send + 'static,
|
|
|
|
Fut: Future<Output= T> + Send + Sync,
|
|
|
|
{
|
|
|
|
#[cfg(debug_assertions)] {
|
|
|
|
let path = path.as_ref();
|
|
|
|
if !path.is_dir() {
|
|
|
|
panic!("walk_dir() expected a directory, but {:?} is not one.", path);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
trace!("{:?} Spawning children", path.as_ref());
|
|
|
|
async move {
|
|
|
|
let se_path = || format!("{:?}", path.as_ref()).header("Path was");
|
|
|
|
let mut dir = fs::read_dir(&path).await
|
|
|
|
.wrap_err(eyre!("Failed to read directory contents"))
|
|
|
|
.with_section(se_path)?;
|
|
|
|
let mut voutput: Vec<T> = util::alloc_stream_hint(&dir);
|
|
|
|
let mut workers: Vec<_> = util::alloc_stream_hint(&dir);
|
|
|
|
while let Some(entry) = dir.next_entry().await
|
|
|
|
.wrap_err(eyre!("Failed to read entry from directory contents"))
|
|
|
|
.with_section(se_path)?
|
|
|
|
{
|
|
|
|
let path = entry.path();
|
|
|
|
if path.is_dir() {
|
|
|
|
|
|
|
|
workers.push(tokio::spawn(walk_dir(path, output.clone())));
|
|
|
|
} else if path.is_file() {
|
|
|
|
voutput.push(output(entry).await);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
voutput.extend(future::join_all(workers)
|
|
|
|
.map(|x| x.into_iter()
|
|
|
|
.filter_map(Result::ok).flatten())
|
|
|
|
.await.into_iter()
|
|
|
|
.flatten());
|
|
|
|
Ok(voutput)
|
|
|
|
}.boxed()
|
|
|
|
}
|
|
|
|
|
|
|
|
/// Process this path
|
|
|
|
///
|
|
|
|
/// This will not return until all its children finish too (if any)
|
|
|
|
pub async fn process(state: Arc<state::State>, path: impl AsRef<Path>) -> eyre::Result<()>
|
|
|
|
{
|
|
|
|
let path = path.as_ref();
|
|
|
|
let se_path = || format!("{:?}", path).header("Path was");
|
|
|
|
if path.is_dir() {
|
|
|
|
for res in walk_dir(path.to_owned(), move |file| process_single(Arc::clone(&state), file.path())).await?
|
|
|
|
.into_iter()
|
|
|
|
.filter_map(Result::err)
|
|
|
|
{
|
|
|
|
error!("{:?} Failed to process child: {}", path, res);
|
|
|
|
return Err(res)
|
|
|
|
.wrap_err(eyre!("Failed to process child"))
|
|
|
|
.with_section(se_path);
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
} else if path.is_file() {
|
|
|
|
process_single(state, path).await
|
|
|
|
.wrap_err(eyre!("Processing file failed"))
|
|
|
|
.with_section(se_path)
|
|
|
|
} else {
|
|
|
|
error!("{:?} is not a recognised FS object", path);
|
|
|
|
return Err(eyre!("Unsupported FS object"))
|
|
|
|
.with_section(se_path)
|
|
|
|
}
|
|
|
|
}
|