async_walkdir/
lib.rs

1// Copyright 2020 Ririsoft <riri@ririsoft.com>
2// Copyright 2024 Jordan Danford <jordandanford@gmail.com>
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! An utility for walking through a directory asynchronously and recursively.
17//!
18//! Based on [async-fs](https://docs.rs/async-fs) and [blocking](https://docs.rs/blocking),
19//! it uses a thread pool to handle blocking IOs. Please refere to those crates for the rationale.
20//! This crate is compatible with any async runtime based on [futures 0.3](https://docs.rs/futures-core),
21//! which includes [tokio](https://docs.rs/tokio), [async-std](https://docs.rs/async-std) and [smol](https://docs.rs/smol).
22//!
23//! Symbolic links are walked through but they are not followed.
24//!
25//! # Example
26//!
27//! Recursively traverse a directory:
28//!
29//! ```
30//! use async_walkdir::WalkDir;
31//! use futures_lite::future::block_on;
32//! use futures_lite::stream::StreamExt;
33//!
34//! block_on(async {
35//!     let mut entries = WalkDir::new("my_directory");
36//!     loop {
37//!         match entries.next().await {
38//!             Some(Ok(entry)) => println!("file: {}", entry.path().display()),
39//!             Some(Err(e)) => {
40//!                 eprintln!("error: {}", e);
41//!                 break;
42//!             }
43//!             None => break,
44//!         }
45//!     }
46//! });
47//! ```
48//!
49//! Do not recurse through directories whose name starts with '.':
50//!
51//! ```
52//! use async_walkdir::{Filtering, WalkDir};
53//! use futures_lite::future::block_on;
54//! use futures_lite::stream::StreamExt;
55//!
56//! block_on(async {
57//!     let mut entries = WalkDir::new("my_directory").filter(|entry| async move {
58//!         if let Some(true) = entry
59//!             .path()
60//!             .file_name()
61//!             .map(|f| f.to_string_lossy().starts_with('.'))
62//!         {
63//!             return Filtering::IgnoreDir;
64//!         }
65//!         Filtering::Continue
66//!     });
67//!
68//!     loop {
69//!         match entries.next().await {
70//!             Some(Ok(entry)) => println!("file: {}", entry.path().display()),
71//!             Some(Err(e)) => {
72//!                 eprintln!("error: {}", e);
73//!                 break;
74//!             }
75//!             None => break,
76//!         }
77//!     }
78//! });
79//! ```
80
81#![forbid(unsafe_code)]
82#![deny(missing_docs)]
83
84mod error;
85
86use std::future::Future;
87use std::path::{Path, PathBuf};
88use std::pin::Pin;
89use std::task::{Context, Poll};
90
91use async_fs::{read_dir, ReadDir};
92use futures_lite::future::Boxed as BoxedFut;
93use futures_lite::stream::{self, Stream, StreamExt};
94
95#[doc(no_inline)]
96pub use async_fs::DirEntry;
97
98pub use error::Error;
99use error::InnerError;
100
101/// A specialized [`Result`][`std::result::Result`] type.
102pub type Result<T> = std::result::Result<T, Error>;
103
104type BoxStream = futures_lite::stream::Boxed<Result<DirEntry>>;
105
106/// A `Stream` of `DirEntry` generated from recursively traversing
107/// a directory.
108///
109/// Entries are returned without a specific ordering. The top most root directory
110/// is not returned but child directories are.
111///
112/// # Panics
113///
114/// Panics if the directories depth overflows `usize`.
115pub struct WalkDir {
116    root: PathBuf,
117    entries: BoxStream,
118}
119
120/// Sets the filtering behavior.
121#[derive(Debug, PartialEq, Eq)]
122pub enum Filtering {
123    /// Ignore the current entry.
124    Ignore,
125    /// Ignore the current entry and, if a directory,
126    /// do not traverse its childs.
127    IgnoreDir,
128    /// Continue the normal processing.
129    Continue,
130}
131
132impl WalkDir {
133    /// Returns a new `Walkdir` starting at `root`.
134    pub fn new(root: impl AsRef<Path>) -> Self {
135        Self {
136            root: root.as_ref().to_owned(),
137            entries: walk_dir(
138                root,
139                None::<Box<dyn FnMut(DirEntry) -> BoxedFut<Filtering> + Send>>,
140            ),
141        }
142    }
143
144    /// Filter entries.
145    pub fn filter<F, Fut>(self, f: F) -> Self
146    where
147        F: FnMut(DirEntry) -> Fut + Send + 'static,
148        Fut: Future<Output = Filtering> + Send,
149    {
150        let root = self.root.clone();
151        Self {
152            root: self.root,
153            entries: walk_dir(root, Some(f)),
154        }
155    }
156}
157
158impl Stream for WalkDir {
159    type Item = Result<DirEntry>;
160
161    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
162        let entries = Pin::new(&mut self.entries);
163        entries.poll_next(cx)
164    }
165}
166
167fn walk_dir<F, Fut>(root: impl AsRef<Path>, filter: Option<F>) -> BoxStream
168where
169    F: FnMut(DirEntry) -> Fut + Send + 'static,
170    Fut: Future<Output = Filtering> + Send,
171{
172    stream::unfold(
173        State::Start((root.as_ref().to_owned(), filter)),
174        move |state| async move {
175            match state {
176                State::Start((root, filter)) => match read_dir(&root).await {
177                    Err(source) => Some((
178                        Err(InnerError::Io { path: root, source }.into()),
179                        State::Done,
180                    )),
181                    Ok(rd) => walk(vec![(root, rd)], filter).await,
182                },
183                State::Walk((dirs, filter)) => walk(dirs, filter).await,
184                State::Done => None,
185            }
186        },
187    )
188    .boxed()
189}
190
191enum State<F> {
192    Start((PathBuf, Option<F>)),
193    Walk((Vec<(PathBuf, ReadDir)>, Option<F>)),
194    Done,
195}
196
197type UnfoldState<F> = (Result<DirEntry>, State<F>);
198
199// Iterative on purpose: filtered-out entries must not grow the stack (see issue #13).
200async fn walk<F, Fut>(
201    mut dirs: Vec<(PathBuf, ReadDir)>,
202    mut filter: Option<F>,
203) -> Option<UnfoldState<F>>
204where
205    F: FnMut(DirEntry) -> Fut + Send + 'static,
206    Fut: Future<Output = Filtering> + Send,
207{
208    loop {
209        let (path, dir) = dirs.last_mut()?;
210        let entry = match dir.next().await {
211            Some(Ok(entry)) => entry,
212            Some(Err(source)) => return io_error(path.to_path_buf(), source, dirs, filter),
213            None => {
214                dirs.pop();
215                continue;
216            }
217        };
218        let ft = match entry.file_type().await {
219            Ok(ft) => ft,
220            Err(source) => return io_error(entry.path(), source, dirs, filter),
221        };
222        let filtering = match filter.as_mut() {
223            Some(filter) => filter(entry.clone()).await,
224            None => Filtering::Continue,
225        };
226        if ft.is_dir() && filtering != Filtering::IgnoreDir {
227            let path = entry.path();
228            let rd = match read_dir(&path).await {
229                Ok(rd) => rd,
230                Err(source) => return io_error(path, source, dirs, filter),
231            };
232            dirs.push((path, rd));
233        }
234        if filtering == Filtering::Continue {
235            return Some((Ok(entry), State::Walk((dirs, filter))));
236        }
237    }
238}
239
240fn io_error<F>(
241    path: PathBuf,
242    source: std::io::Error,
243    dirs: Vec<(PathBuf, ReadDir)>,
244    filter: Option<F>,
245) -> Option<UnfoldState<F>> {
246    let err = InnerError::Io { path, source }.into();
247    Some((Err(err), State::Walk((dirs, filter))))
248}
249
250#[cfg(test)]
251mod tests {
252    use std::io::{ErrorKind, Result};
253
254    use futures_lite::future::block_on;
255    use futures_lite::stream::StreamExt;
256
257    use super::{Filtering, WalkDir};
258
259    #[test]
260    fn walk_dir_empty() -> Result<()> {
261        block_on(async {
262            let root = tempfile::tempdir()?;
263            let mut wd = WalkDir::new(root.path());
264            assert!(wd.next().await.is_none());
265            Ok(())
266        })
267    }
268
269    #[test]
270    fn walk_dir_not_exist() {
271        block_on(async {
272            let mut wd = WalkDir::new("foobar");
273            match wd.next().await.unwrap() {
274                Err(e) => {
275                    assert_eq!(wd.root, e.path().unwrap());
276                    assert_eq!(e.io().unwrap().kind(), ErrorKind::NotFound);
277                    assert_eq!(e.into_io().unwrap().kind(), ErrorKind::NotFound);
278                }
279                _ => panic!("want IO error"),
280            }
281            assert!(wd.next().await.is_none());
282        })
283    }
284
285    #[test]
286    fn walk_dir_read_dir_error() -> Result<()> {
287        block_on(async {
288            let root = tempfile::tempdir()?;
289            let d1 = root.path().join("d1");
290            async_fs::create_dir_all(&d1).await?;
291            // Removing the directory from the filter makes the subsequent read_dir fail.
292            let mut wd = WalkDir::new(root.path()).filter(|entry| async move {
293                async_fs::remove_dir(entry.path()).await.unwrap();
294                Filtering::Continue
295            });
296            match wd.next().await.unwrap() {
297                Err(e) => {
298                    assert_eq!(e.path().unwrap(), d1.as_path());
299                    assert_eq!(e.io().unwrap().kind(), ErrorKind::NotFound);
300                }
301                _ => panic!("want IO error"),
302            }
303            assert!(wd.next().await.is_none());
304            Ok(())
305        })
306    }
307
308    #[test]
309    fn into_io_error() {
310        block_on(async {
311            let mut wd = WalkDir::new("foobar");
312            match wd.next().await.unwrap() {
313                Err(e) => {
314                    let e: std::io::Error = e.into();
315                    assert_eq!(e.kind(), ErrorKind::NotFound);
316                }
317                _ => panic!("want IO error"),
318            }
319        })
320    }
321
322    #[test]
323    fn walk_dir_files() -> Result<()> {
324        block_on(async {
325            let root = tempfile::tempdir()?;
326            let f1 = root.path().join("f1.txt");
327            let d1 = root.path().join("d1");
328            let f2 = d1.join("f2.txt");
329            let d2 = d1.join("d2");
330            let f3 = d2.join("f3.txt");
331
332            async_fs::create_dir_all(&d2).await?;
333            async_fs::write(&f1, []).await?;
334            async_fs::write(&f2, []).await?;
335            async_fs::write(&f3, []).await?;
336
337            let want = vec![
338                d1.to_owned(),
339                d2.to_owned(),
340                f3.to_owned(),
341                f2.to_owned(),
342                f1.to_owned(),
343            ];
344            let mut wd = WalkDir::new(root.path());
345
346            let mut got = Vec::new();
347            while let Some(entry) = wd.next().await {
348                let entry = entry.unwrap();
349                got.push(entry.path());
350            }
351            got.sort();
352            assert_eq!(got, want);
353
354            Ok(())
355        })
356    }
357
358    #[test]
359    fn filter_dirs() -> Result<()> {
360        block_on(async {
361            let root = tempfile::tempdir()?;
362            let f1 = root.path().join("f1.txt");
363            let d1 = root.path().join("d1");
364            let f2 = d1.join("f2.txt");
365            let d2 = d1.join("d2");
366            let f3 = d2.join("f3.txt");
367
368            async_fs::create_dir_all(&d2).await?;
369            async_fs::write(&f1, []).await?;
370            async_fs::write(&f2, []).await?;
371            async_fs::write(&f3, []).await?;
372
373            let want = vec![f3.to_owned(), f2.to_owned(), f1.to_owned()];
374
375            let mut wd = WalkDir::new(root.path()).filter(|entry| async move {
376                match entry.file_type().await {
377                    Ok(ft) if ft.is_dir() => Filtering::Ignore,
378                    _ => Filtering::Continue,
379                }
380            });
381
382            let mut got = Vec::new();
383            while let Some(entry) = wd.next().await {
384                let entry = entry.unwrap();
385                got.push(entry.path());
386            }
387            got.sort();
388            assert_eq!(got, want);
389
390            Ok(())
391        })
392    }
393
394    #[test]
395    fn filter_dirs_no_traverse() -> Result<()> {
396        block_on(async {
397            let root = tempfile::tempdir()?;
398            let f1 = root.path().join("f1.txt");
399            let d1 = root.path().join("d1");
400            let f2 = d1.join("f2.txt");
401            let d2 = d1.join("d2");
402            let f3 = d2.join("f3.txt");
403
404            async_fs::create_dir_all(&d2).await?;
405            async_fs::write(&f1, []).await?;
406            async_fs::write(&f2, []).await?;
407            async_fs::write(&f3, []).await?;
408
409            let want = vec![d1, f2.to_owned(), f1.to_owned()];
410
411            let mut wd = WalkDir::new(root.path()).filter(move |entry| {
412                let d2 = d2.clone();
413                async move {
414                    if entry.path() == d2 {
415                        Filtering::IgnoreDir
416                    } else {
417                        Filtering::Continue
418                    }
419                }
420            });
421
422            let mut got = Vec::new();
423            while let Some(entry) = wd.next().await {
424                let entry = entry.unwrap();
425                got.push(entry.path());
426            }
427            got.sort();
428            assert_eq!(got, want);
429
430            Ok(())
431        })
432    }
433}
434
435#[cfg(all(unix, test))]
436mod test_unix {
437    use async_fs::unix::PermissionsExt;
438    use std::io::Result;
439
440    use futures_lite::future::block_on;
441    use futures_lite::stream::StreamExt;
442
443    use super::{Filtering, WalkDir};
444
445    #[test]
446    fn filter_ignore_dir_does_not_read_it() -> Result<()> {
447        block_on(async {
448            let root = tempfile::tempdir()?;
449            let d1 = root.path().join("d1");
450            async_fs::create_dir_all(&d1).await?;
451            let mut perms = async_fs::metadata(&d1).await?.permissions();
452            perms.set_mode(0o222);
453            async_fs::set_permissions(&d1, perms).await?;
454            let mut wd = WalkDir::new(&root).filter(|_| async { Filtering::IgnoreDir });
455            assert!(wd.next().await.is_none());
456            Ok(())
457        })
458    }
459
460    #[test]
461    fn walk_dir_error_path() -> Result<()> {
462        block_on(async {
463            let root = tempfile::tempdir()?;
464            let d1 = root.path().join("d1");
465            async_fs::create_dir_all(&d1).await?;
466            let mut perms = async_fs::metadata(&d1).await?.permissions();
467            perms.set_mode(0o222);
468            async_fs::set_permissions(&d1, perms).await?;
469            let mut wd = WalkDir::new(&root);
470            match wd.next().await.unwrap() {
471                Err(e) => assert_eq!(e.path().unwrap(), d1.as_path()),
472                _ => panic!("want IO error"),
473            }
474            Ok(())
475        })
476    }
477}
async_walkdir/lib.rs

async_walkdir/
lib.rs
