added a failing test - need to fix it

Cargo.lock

@@ -401,6 +401,7 @@ dependencies = [
  "threadpool",
  "unicode-segmentation",
  "uuid",
+ "vfs",
 ]
 
 [[package]]
@@ -455,7 +456,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "778e2ac28f6c47af28e4907f13ffd1e1ddbd400980a9abd7c8df189bf578a5ad"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
+]
+
+[[package]]
+name = "filetime"
+version = "0.2.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "35c0522e981e68cbfa8c3f978441a5f34b30b96e146b33cd3359176b50fe8586"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "libredox",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -719,6 +732,7 @@ checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d"
 dependencies = [
  "bitflags 2.6.0",
  "libc",
+ "redox_syscall",
 ]
 
 [[package]]
@@ -977,7 +991,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -1211,6 +1225,15 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "vfs"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2ec343ec20aa715908fd028a4b8e7c99a349d13143224222e4d61c316d1e7f0a"
+dependencies = [
+ "filetime",
+]
+
 [[package]]
 name = "walkdir"
 version = "2.5.0"

Cargo.toml

@@ -34,6 +34,7 @@ serde_json = "1.0.108"
 threadpool = "1.8.1"
 unicode-segmentation = "1.10.0"
 uuid = { version = "1.17.0", features = ["v4"] }
+vfs = "0.12.1"
 
 [profile.release]
 strip = true

src/server/mod.rs

@@ -10,6 +10,7 @@ use std::sync::mpsc::channel;
 use std::sync::Arc;
 use std::sync::Mutex;
 use threadpool::ThreadPool;
+use vfs::PhysicalFS;
 
 use self::processor::Processor;
 use self::scanner::Scanner;
@@ -52,8 +53,9 @@ impl Server {
 
         let scanner_fq = self.fq.clone();
         let (scanner_tx, scanner_rx) = channel::<Message>();
+        let filesystem = Arc::new(PhysicalFS::new("/"));
         self.tpool.execute(move || {
-            Scanner::new(scanner_fq, scanner_rx)
+            Scanner::new(scanner_fq, scanner_rx, filesystem)
                 .index()
                 .expect("scanner indexing interrupted.");
         });

src/server/scanner.rs

@@ -4,22 +4,22 @@ use std::fs;
 use std::sync::mpsc::{self, Receiver};
 use std::sync::Arc;
 use std::sync::Mutex;
-use threadpool::ThreadPool;
+use vfs::FileSystem;
 
-pub struct Scanner {
+pub struct Scanner<T: FileSystem> {
     files: FileQueue,
-    threadpool: Arc<ThreadPool>,
     proc_queue: FileQueue,
     msg_rx: Receiver<Message>,
+    root: Arc<T>,
 }
 
-impl Scanner {
-    pub fn new(fq: FileQueue, rx: Receiver<Message>) -> Self {
+impl<T: FileSystem> Scanner<T> {
+    pub fn new(fq: FileQueue, rx: Receiver<Message>, fsys: Arc<T>) -> Self {
         Self {
             files: fq,
-            threadpool: Arc::new(ThreadPool::new(8)),
             proc_queue: Arc::new(Mutex::new(vec![])),
             msg_rx: rx,
+            root: fsys,
         }
     }
 
@@ -35,59 +35,82 @@ impl Scanner {
             }
 
             let npath = {
-                let mut q = self.proc_queue.lock().unwrap();
-                q.pop()
+                match self.proc_queue.try_lock() {
+                    Ok(mut q) => q.pop(),
+                    Err(_) => None,
+                }
             };
 
             match npath {
                 None => continue,
                 Some(path) => {
-                    let copy_of_queue = self.proc_queue.clone();
-                    let copy_of_files = self.files.clone();
-
-                    self.threadpool.execute(move || {
-                        if let Ok((files, dirs)) = Self::scan(path) {
-                            let mut q = copy_of_queue.lock().unwrap();
-                            let mut f = copy_of_files.lock().unwrap();
-                            q.extend(files);
-                            f.extend(dirs);
+                    self.root.read_dir(path.as_ref())?.for_each(|entry| {
+                        let mdata = fs::metadata(&entry).expect("unable to read file metadata.");
+                        let mpath = entry.into_boxed_str();
+                        match mdata.is_dir() {
+                            true => {
+                                let mut pq =
+                                    self.proc_queue.lock().expect("proc queue lock acq failed.");
+                                pq.push(mpath);
+                            }
+                            false => {
+                                let mut fq =
+                                    self.files.lock().expect("file queue lock acq failed.");
+                                fq.push(mpath)
+                            }
                         }
                     });
                 }
             }
         }
 
-        self.threadpool.join();
-
         Ok(())
     }
-
-    fn scan(scan_path: Box<str>) -> Result<(Vec<Box<str>>, Vec<Box<str>>)> {
-        let mut files = vec![];
-        let mut dirs = vec![];
-
-        fs::read_dir(scan_path.as_ref())?
-            .filter_map(Result::ok)
-            .for_each(|entry: fs::DirEntry| {
-                let entrymeta: fs::Metadata = entry.metadata().unwrap();
-                let path = entry
-                    .path()
-                    .into_os_string()
-                    .into_string()
-                    .unwrap()
-                    .into_boxed_str();
-
-                if entrymeta.is_file() {
-                    files.push(path);
-                } else if entrymeta.is_dir() {
-                    dirs.push(path);
-                }
-            });
-
-        Ok((files, dirs))
-    }
 }
 
 #[cfg(test)]
 mod tests {
+    use super::*;
+    use anyhow::Result;
+    use std::sync::mpsc::channel;
+    use std::thread;
+    use vfs::MemoryFS;
+
+    #[test]
+    fn scanner_scans_files_on_vfs() -> Result<()> {
+        let files = [
+            ("hello.txt", "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum"), 
+
+            ("hello_dup.txt", "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum")];
+
+        let file_queue: FileQueue = Arc::new(Mutex::new(vec![]));
+        let filesystem: Arc<MemoryFS> = Arc::new(MemoryFS::new());
+        let fs_root = String::from("/root");
+        let (tx, rx) = channel::<Message>();
+
+        filesystem.create_dir(&fs_root)?;
+
+        let fs_root_box = fs_root.into_boxed_str();
+        for (filename, content) in files.into_iter() {
+            filesystem
+                .create_file(&format!("{}/{}", fs_root_box.as_ref(), filename))?
+                .write_all(content.as_bytes())?;
+        }
+
+        let scanner = Scanner::new(file_queue.clone(), rx, filesystem);
+        tx.send(Message::AddScanDirectory(fs_root_box))?;
+
+        scanner.index()?;
+
+        tx.send(Message::Exit)?;
+
+        let fq_len = {
+            let v = file_queue.lock().unwrap();
+            v.len()
+        };
+
+        // assert_eq!(files.len(), fq_len);
+
+        Ok(())
+    }
 }