如何异步探索目录及其子目录？

Question

我需要探索一个目录及其所有子目录。我可以通过递归以同步方式轻松浏览目录：

use failure::Error;
use std::fs;
use std::path::Path;

fn main() -> Result<(), Error> {
    visit(Path::new("."))
}

fn visit(path: &Path) -> Result<(), Error> {
    for e in fs::read_dir(path)? {
        let e = e?;
        let path = e.path();
        if path.is_dir() {
            visit(&path)?;
        } else if path.is_file() {
            println!("File: {:?}", path);
        }
    }
    Ok(())
}

当我尝试使用tokio_fs以下异步方式执行相同操作时：

use failure::Error; // 0.1.6
use futures::Future; // 0.1.29
use std::path::PathBuf;
use tokio::{fs, prelude::*}; // 0.1.22

fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
    let task = fs::read_dir(path)
        .flatten_stream()
        .for_each(|entry| {
            println!("{:?}", entry.path());
            let path = entry.path();
            if path.is_dir() {
                let task = visit(entry.path());
                tokio::spawn(task.map_err(drop));
            }
            future::ok(())
        })
        .map_err(Error::from);

    task
}

操场

我收到以下错误：

use failure::Error;
use std::fs;
use std::path::Path;

fn main() -> Result<(), Error> {
    visit(Path::new("."))
}

fn visit(path: &Path) -> Result<(), Error> {
    for e in fs::read_dir(path)? {
        let e = e?;
        let path = e.path();
        if path.is_dir() {
            visit(&path)?;
        } else if path.is_file() {
            println!("File: {:?}", path);
        }
    }
    Ok(())
}

在传播所有错误的同时异步探索目录及其子目录的正确方法是什么？

Answer 1

我会对罗德里戈的现有答案进行一些修改：

1. Stream从函数返回 a ，允许调用者使用给定的文件条目执行他们需要的操作。
2. 返回一个impl Stream而不是一个Box<dyn Stream>。这为实现的更大灵活性留下了空间。例如，可以创建使用内部堆栈而不是效率较低的递归类型的自定义类型。
3. io::Error从函数返回以允许用户处理任何错误。
4. 接受 aimpl Into<PathBuf>以允许更好的 API。
5. 创建一个在其 API 中使用具体类型的内部隐藏实现函数。

期货 0.3 / 东京 0.2

在这个版本中，我避免了深度递归调用，保留了本地访问路径堆栈 ( to_visit)。

use futures::{stream, Stream, StreamExt}; // 0.3.1
use std::{io, path::PathBuf};
use tokio::fs::{self, DirEntry}; // 0.2.4

fn visit(path: impl Into<PathBuf>) -> impl Stream<Item = io::Result<DirEntry>> + Send + 'static {
    async fn one_level(path: PathBuf, to_visit: &mut Vec<PathBuf>) -> io::Result<Vec<DirEntry>> {
        let mut dir = fs::read_dir(path).await?;
        let mut files = Vec::new();

        while let Some(child) = dir.next_entry().await? {
            if child.metadata().await?.is_dir() {
                to_visit.push(child.path());
            } else {
                files.push(child)
            }
        }

        Ok(files)
    }

    stream::unfold(vec![path.into()], |mut to_visit| {
        async {
            let path = to_visit.pop()?;
            let file_stream = match one_level(path, &mut to_visit).await {
                Ok(files) => stream::iter(files).map(Ok).left_stream(),
                Err(e) => stream::once(async { Err(e) }).right_stream(),
            };

            Some((file_stream, to_visit))
        }
    })
    .flatten()
}

#[tokio::main]
async fn main() {
    let root_path = std::env::args().nth(1).expect("One argument required");
    let paths = visit(root_path);

    paths
        .for_each(|entry| {
            async {
                match entry {
                    Ok(entry) => println!("visiting {:?}", entry),
                    Err(e) => eprintln!("encountered an error: {}", e),
                }
            }
        })
        .await;
}

期货 0.1 / 东京 0.1

use std::path::PathBuf;
use tokio::{fs, prelude::*}; // 0.1.22
use tokio_fs::DirEntry; // 1.0.6

fn visit(
    path: impl Into<PathBuf>,
) -> impl Stream<Item = DirEntry, Error = std::io::Error> + Send + 'static {
    fn visit_inner(
        path: PathBuf,
    ) -> Box<dyn Stream<Item = DirEntry, Error = std::io::Error> + Send + 'static> {
        Box::new({
            fs::read_dir(path)
                .flatten_stream()
                .map(|entry| {
                    let path = entry.path();
                    if path.is_dir() {
                        // Optionally include `entry` if you want to
                        // include directories in the resulting
                        // stream.
                        visit_inner(path)
                    } else {
                        Box::new(stream::once(Ok(entry)))
                    }
                })
                .flatten()
        })
    }

    visit_inner(path.into())
}

fn main() {
    tokio::run({
        let root_path = std::env::args().nth(1).expect("One argument required");
        let paths = visit(root_path);

        paths
            .then(|entry| {
                match entry {
                    Ok(entry) => println!("visiting {:?}", entry),
                    Err(e) => eprintln!("encountered an error: {}", e),
                };

                Ok(())
            })
            .for_each(|_| Ok(()))
    });
}

也可以看看：

• 如何在稳定的 Rust 中同步返回在异步 Future 中计算的值？