More code to delay file system events.

The code to process the events is still missing.

src/file_event_delay.rs

@@ -1,9 +1,12 @@
+use std::collections::VecDeque;
 use std::io;
+use std::mem;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
 use futures::stream::Stream;
+use tokio::time::{self, Interval};
 
 /// A stream which delays inotify events and combines them whenever possible.
 ///
@@ -15,8 +18,18 @@ pub struct FileEventDelay<T>
 where
     T: Stream<Item = io::Result<inotify::EventOwned>>,
 {
-    input: T,
+    input: Pin<Box<T>>,
+
     min_delay: Duration,
+    timer: Option<Pin<Box<Interval>>>,
+
+    /// Inotify events which are queued for processing.
+    ///
+    /// Whenever the timer expires, the events from the second array are processed and the content
+    /// of the first array is moved to the second. This way, each event is in the queue for a
+    /// duration of at least `min_delay`.
+    event_queue: (Vec<inotify::EventOwned>, Vec<inotify::EventOwned>),
+    processed_events: VecDeque<FileEvent>,
 }
 
 impl<T> FileEventDelay<T>
@@ -24,7 +37,21 @@ where
     T: Stream<Item = io::Result<inotify::EventOwned>>,
 {
     pub fn new(input: T, min_delay: Duration) -> Self {
-        Self { input, min_delay }
+        Self {
+            input: Box::pin(input),
+            min_delay,
+            timer: None,
+            event_queue: (Vec::new(), Vec::new()),
+            processed_events: VecDeque::new(),
+        }
+    }
+
+    fn process_events(&mut self) {
+        // TODO
+
+        // Swap the arrays. This code assumes that the second array is now empty.
+        // TODO: Check the assumption.
+        mem::swap(&mut self.event_queue.0, &mut self.event_queue.1);
     }
 }
 
@@ -34,11 +61,50 @@ where
 {
     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.");
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // Safe, as we will not move self_.
+        let self_ = unsafe { self.get_unchecked_mut() };
+
+        if self_.processed_events.is_empty() {
+            // Fill the first array with incoming inotify events.
+            while let Poll::Ready(inotify_event) = Pin::as_mut(&mut self_.input).poll_next(cx) {
+                if inotify_event.is_some() {
+                    self_.event_queue.0.push(inotify_event.unwrap().unwrap());
+                } else {
+                    // TODO: Stream exhausted. Can this ever happen?
+                    return Poll::Ready(None);
+                }
+            }
+
+            // If the first array is not empty, start a timer for processing if no timer is already
+            // running.
+            if !self_.event_queue.0.is_empty() && self_.timer.is_none() {
+                self_.timer = Some(Box::pin(time::interval(self_.min_delay)));
+            }
+
+            // If the timer elapsed, process the events from the second array and move the content
+            // of the first array to the second.
+            if let Poll::Ready(Some(_)) =
+                Pin::as_mut(&mut self_.timer.as_mut().unwrap()).poll_next(cx)
+            {
+                self_.process_events();
+
+                // If no unprocessed events are available, stop the timer to reduce CPU
+                // consumption.
+                if self_.event_queue.1.is_empty() {
+                    self_.timer = None;
+                }
+                // Return the first event if possible.
+                if let Some(next_event) = self_.processed_events.pop_front() {
+                    return Poll::Ready(Some(Ok(next_event)));
+                }
+            }
+
+            Poll::Pending
+        } else {
+            let next_event = self_.processed_events.pop_front().unwrap();
+            Poll::Ready(Some(Ok(next_event)))
+        }
     }
 }