From dd3a29d9e9e8118b9f303915c2fabc1a3594280a Mon Sep 17 00:00:00 2001 From: Chris Mikkelson Date: Fri, 16 Aug 2024 10:47:59 -0500 Subject: [PATCH] first cut of fileset support --- src/fileset.rs | 75 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 1 + 2 files changed, 76 insertions(+) create mode 100644 src/fileset.rs diff --git a/src/fileset.rs b/src/fileset.rs new file mode 100644 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, Instant)>>, + merger: Mutex>>>, +} + +fn load_fs_file(f: impl AsRef) -> Result> { + use std::fs::read_to_string; + read_to_string(f) + .and_then(|s| Ok(s.lines().map(|l| PathBuf::from(l)).collect::>())) + .map_err(|e| e.into()) +} + +impl Fileset { + pub fn new(fname: impl AsRef) -> 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::>(); + 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 ceb7f4d..4e61836 100644 --- 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; -- 2.50.1