first cut of fileset support

diff --git a/src/fileset.rs b/src/fileset.rs
new file mode 100644 (file)
index 0000000..ee5e645
--- /dev/null
+++ b/src/fileset.rs
@@ -0,0 +1,75 @@
+#![allow(dead_code)]
+use memmap::Mmap;
+
+use crate::merger::Merger;
+use crate::reader::Reader;
+use crate::Result;
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+pub struct Fileset {
+    fs_path: PathBuf,
+    last_reload: Instant,
+    reload_interval: Duration,
+    readers: Mutex<HashMap<PathBuf, (Reader<Mmap>, Instant)>>,
+    merger: Mutex<Option<Merger<Reader<Mmap>>>>,
+}
+
+fn load_fs_file(f: impl AsRef<Path>) -> Result<Vec<PathBuf>> {
+    use std::fs::read_to_string;
+    read_to_string(f)
+        .and_then(|s| Ok(s.lines().map(|l| PathBuf::from(l)).collect::<Vec<_>>()))
+        .map_err(|e| e.into())
+}
+
+impl Fileset {
+    pub fn new(fname: impl AsRef<Path>) -> Self {
+        let mut res = Self {
+            fs_path: PathBuf::new().with_file_name(fname.as_ref()),
+            last_reload: Instant::now(),
+            reload_interval: Duration::new(1, 0),
+            readers: Mutex::new(HashMap::new()),
+            merger: Mutex::new(None),
+        };
+        res.force_reload();
+        res
+    }
+
+    fn force_reload(&mut self) {
+        let files = load_fs_file(&self.fs_path).unwrap();
+        let now = Instant::now();
+
+        let mut readers = self.readers.lock().unwrap();
+        if self.last_reload > now {
+            return;
+        }
+        files
+            .iter()
+            .map(|f| match readers.get(f) {
+                Some(_) => None,
+                None => readers.insert(f.to_path_buf(), (crate::reader::from_file(f), now)),
+            })
+            .count(); // must use
+
+        let clean = readers
+            .iter()
+            .filter(|(_, (_, t))| *t < now)
+            .map(|(f, _)| f.clone())
+            .collect::<Vec<_>>();
+        clean.iter().for_each(|f| {
+            readers.remove(f);
+            ()
+        });
+        let mut merger = self.merger.lock().unwrap();
+        merger.replace(Merger::from(readers.iter().map(|(_, (r, _))| r.clone())));
+        self.last_reload = now;
+    }
+
+    fn reload(&mut self) {
+        if Instant::now().duration_since(self.last_reload) > self.reload_interval {
+            self.force_reload();
+        }
+    }
+}

diff --git a/src/lib.rs b/src/lib.rs
index ceb7f4db01373e73330959a09eb336e858acfd1c..4e6183612871f8ac75f9649df132155fb8001da3 100644 (file)
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -14,6 +14,7 @@ pub mod compression;
 pub mod reader;
 pub mod writer;
 
+mod fileset;
 mod metadata;
 
 type Error = Box<dyn std::error::Error>;