Add API to delay and combine inotify events.

Cargo.lock

@@ -907,6 +907,7 @@ dependencies = [
  "mio 0.6.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio-uds 0.6.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-macros 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 

Cargo.toml

@@ -10,7 +10,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 dirs = "2.0"
 byteorder = "1.3"
-tokio = { version = "0.2", features = ["fs", "io-util", "macros", "net", "stream", "sync", "tcp"] }
+tokio = { version = "0.2", features = ["fs", "io-util", "macros", "net", "stream", "sync", "tcp", "time"] }
 futures = "0.3.1"
 futures-tokio-compat = { git = "https://github.com/mgottschlag/futures-tokio-compat.git" }
 async-tungstenite = { version = "0.3", features = ["tokio-runtime"] }

src/file_event_delay.rs

@@ -0,0 +1,47 @@
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+use futures::stream::Stream;
+
+/// A stream which delays inotify events and combines them whenever possible.
+///
+/// The type tries to reduce the number of unnecessary events (e.g., a file which was created and
+/// then immediately removed again, or a large number of modification events for the same file) and
+/// tries to reconstruct additional information (e.g., if a file was moved from one directory to
+/// another, in which case inotify cannot detect the move).
+pub struct FileEventDelay<T>
+where
+    T: Stream<Item = io::Result<inotify::EventOwned>>,
+{
+    input: T,
+    min_delay: Duration,
+}
+
+impl<T> FileEventDelay<T>
+where
+    T: Stream<Item = io::Result<inotify::EventOwned>>,
+{
+    pub fn new(input: T, min_delay: Duration) -> Self {
+        Self { input, min_delay }
+    }
+}
+
+impl<T> Stream for FileEventDelay<T>
+where
+    T: Stream<Item = io::Result<inotify::EventOwned>>,
+{
+    type Item = io::Result<FileEvent>;
+
+    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // TODO: Two arrays, fill one array with inotify events. Whenever a periodic timer elapses,
+        // parse contents of the other array and clear it, and swap the two arrays.
+        // TODO
+        panic!("Not yet implemented.");
+    }
+}
+
+pub enum FileEvent {
+    // TODO
+}

src/file_system_watcher.rs

@@ -1,12 +1,14 @@
 use std::collections::VecDeque;
 use std::sync::Arc;
 use std::path::Path;
+use std::time::Duration;
 
 use inotify::{Inotify, WatchMask};
 use tokio::sync::{mpsc, oneshot, Mutex};
 use tokio::task;
 
 use super::{Error, FileTree};
+use super::file_event_delay::FileEventDelay;
 
 pub struct FileSystemWatcher {
     state_send: mpsc::Sender<(StateChange, oneshot::Sender<()>)>,
@@ -79,7 +81,8 @@ impl FileSystemWatcher {
         new_directories.push_back(file_tree.lock().await.root_path().to_owned());
 
         let mut inotify_buffer = InotifyBuffer { data: [0; 1024] };
-        let _inotify_stream = inotify.event_stream(&mut inotify_buffer).unwrap();
+        let inotify_stream = inotify.event_stream(&mut inotify_buffer).unwrap();
+        let _file_events = FileEventDelay::new(inotify_stream, Duration::from_secs(3));
         //let watches_by_path = BTreeMap::new();
         //let paths_by_watch = HashMap::new();
 

src/lib.rs

@@ -1,6 +1,7 @@
 use std::io;
 
 mod database;
+mod file_event_delay;
 mod file_system_watcher;
 mod file_tree;