Repository: hannobraun/inotify-rs (language: rust)

Issue:
`Inotify::event_stream` is error-prone
`Inotify::event_stream` enabled concurrent use of one `Inotify` instance and an arbitrary number of `EventStream` instances, which is error-prone. It should be replaced with an `into_event_stream` method that consumes the `Inotify`. Ideally, there should also be a matching method on `EventStream`, to convert back to `Inotify`.

See https://github.com/hannobraun/inotify-rs/issues/112#issuecomment-850352465 and https://github.com/hannobraun/inotify-rs/issues/112#issuecomment-850418244.


Discussion / hints:
The proposed solution makes sense to me. Between `into_event_stream()` versus passing a `&mut Inotify` to EventStream, the `into` option makes it clearer what the EventStream struct's API is, and relieves the caller of having to juggle two references (Inotify and EventStream). New APIs can be duplicated on both Inotify and EventStream if they both need access, like `.watches()`.

Originally though, my thought was "why can't inotify-rs handle multiple EventStreams, like a PubSub pattern?" But I found there's not a drop-in solution for that. People have attempted such "forking" combinators for Rust futures, but there's not a stable, wide-consensus implementation yet (see [Rust futures streams GitHub issues](https://github.com/rust-lang/futures-rs/issues?q=is%3Aissue+is%3Aopen+label%3AA-stream+)). The [Tokio streams tutorial](https://tokio.rs/tokio/tutorial/streams) even delegates to Redis for PubSub functionality.

So in conclusion, `Inotify.into_event_stream()` seems the right solution and in line with what the Rust ecosystem currently provides. Downstream projects could implement their own multi-consumer solutions on top of the EventStream if needed.

Fix patch:
diff --git a/examples/stream.rs b/examples/stream.rs
index 03047c1..ca2282f 100644
--- a/examples/stream.rs
+++ b/examples/stream.rs
@@ -14,7 +14,7 @@ use tempfile::TempDir;
 
 #[tokio::main]
 async fn main() -> Result<(), io::Error> {
-    let mut inotify = Inotify::init()
+    let inotify = Inotify::init()
         .expect("Failed to initialize inotify");
 
     let dir = TempDir::new()?;
@@ -29,7 +29,7 @@ async fn main() -> Result<(), io::Error> {
     });
 
     let mut buffer = [0; 1024];
-    let mut stream = inotify.event_stream(&mut buffer)?;
+    let mut stream = inotify.into_event_stream(&mut buffer)?;
 
     while let Some(event_or_error) = stream.next().await {
         println!("event: {:?}", event_or_error?);
diff --git a/src/inotify.rs b/src/inotify.rs
index f4a3c40..f85fb1c 100644
--- a/src/inotify.rs
+++ b/src/inotify.rs
@@ -277,19 +277,43 @@ impl Inotify {
         Ok(Events::new(Arc::downgrade(&self.fd), buffer, num_bytes))
     }
 
-    /// Create a stream which collects events
+    /// Deprecated: use `into_event_stream()` instead, which enforces a single `Stream` and predictable reads.
+    /// Using this method to create multiple `EventStream` instances from one `Inotify` is unsupported,
+    /// as they will contend over one event source and each produce unpredictable stream contents.
+    #[deprecated = "use `into_event_stream()` instead, which enforces a single Stream and predictable reads"]
+    #[cfg(feature = "stream")]
+    pub fn event_stream<T>(&mut self, buffer: T)
+        -> io::Result<EventStream<T>>
+    where
+        T: AsMut<[u8]> + AsRef<[u8]>,
+    {
+        EventStream::new(self.fd.clone(), buffer)
+    }
+
+    /// Create a stream which collects events. Consumes the `Inotify` instance.
     ///
     /// Returns a `Stream` over all events that are available. This stream is an
     /// infinite source of events.
     ///
     /// An internal buffer which can hold the largest possible event is used.
     #[cfg(feature = "stream")]
-    pub fn event_stream<T>(&mut self, buffer: T)
+    pub fn into_event_stream<T>(self, buffer: T)
         -> io::Result<EventStream<T>>
     where
         T: AsMut<[u8]> + AsRef<[u8]>,
     {
-        EventStream::new(self.fd.clone(), buffer)
+        EventStream::new(self.fd, buffer)
+    }
+
+    /// Creates an `Inotify` instance using the file descriptor which was originally
+    /// initialized in `Inotify::init`. This is intended to be used to transform an
+    /// `EventStream` back into an `Inotify`. Do not attempt to clone `Inotify` with this.
+    #[cfg(feature = "stream")]
+    pub(crate) fn from_file_descriptor(fd: Arc<FdGuard>) -> Self
+    {
+        Inotify {
+            fd,
+        }
     }
 
     /// Closes the inotify instance
diff --git a/src/stream.rs b/src/stream.rs
index aaad5c7..65a2e3c 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -11,14 +11,15 @@ use tokio::io::unix::AsyncFd;
 
 use crate::events::{Event, EventOwned};
 use crate::fd_guard::FdGuard;
+use crate::Inotify;
 use crate::util::read_into_buffer;
 use crate::watches::Watches;
 
 /// Stream of inotify events
 ///
-/// Allows for streaming events returned by [`Inotify::event_stream`].
+/// Allows for streaming events returned by [`Inotify::into_event_stream`].
 ///
-/// [`Inotify::event_stream`]: struct.Inotify.html#method.event_stream
+/// [`Inotify::into_event_stream`]: struct.Inotify.html#method.into_event_stream
 #[derive(Debug)]
 pub struct EventStream<T> {
     fd: AsyncFd<ArcFdGuard>,
@@ -49,6 +50,12 @@ where
     pub fn watches(&self) -> Watches {
         Watches::new(self.fd.get_ref().0.clone())
     }
+
+    /// Consumes the `EventStream` instance and returns an `Inotify` using the original
+    /// file descriptor that was passed from `Inotify` to create the `EventStream`.
+    pub fn into_inotify(self) -> Inotify {
+        Inotify::from_file_descriptor(self.fd.into_inner().0)
+    }
 }
 
 impl<T> Stream for EventStream<T>


Test patch:
diff --git a/tests/main.rs b/tests/main.rs
index 181371a..9c5b476 100644
--- a/tests/main.rs
+++ b/tests/main.rs
@@ -49,8 +49,16 @@ async fn it_should_watch_a_file_async() {
     let mut testdir = TestDir::new();
     let (path, mut file) = testdir.new_file();
 
-    let mut inotify = Inotify::init().unwrap();
-    let watch = inotify.watches().add(&path, WatchMask::MODIFY).unwrap();
+    let inotify = Inotify::init().unwrap();
+
+    // Hold ownership of `watches` for this test, so that the underlying file descriptor has
+    // at least one reference to keep it alive, and we can inspect the WatchDescriptors below.
+    // Otherwise the `Weak<FdGuard>` contained in the WatchDescriptors will be invalidated
+    // when `inotify` is consumed by `into_event_stream()` and the EventStream is dropped
+    // during `await`.
+    let mut watches = inotify.watches();
+
+    let watch = watches.add(&path, WatchMask::MODIFY).unwrap();
 
     write_to(&mut file);
 
@@ -58,7 +66,7 @@ async fn it_should_watch_a_file_async() {
 
     use futures_util::StreamExt;
     let events = inotify
-        .event_stream(&mut buffer[..])
+        .into_event_stream(&mut buffer[..])
         .unwrap()
         .take(1)
         .collect::<Vec<_>>()
@@ -80,14 +88,20 @@ async fn it_should_watch_a_file_from_eventstream_watches() {
     let mut testdir = TestDir::new();
     let (path, mut file) = testdir.new_file();
 
-    let mut inotify = Inotify::init().unwrap();
+    let inotify = Inotify::init().unwrap();
 
     let mut buffer = [0; 1024];
 
     use futures_util::StreamExt;
-    let stream = inotify.event_stream(&mut buffer[..]).unwrap();
+    let stream = inotify.into_event_stream(&mut buffer[..]).unwrap();
+
+    // Hold ownership of `watches` for this test, so that the underlying file descriptor has
+    // at least one reference to keep it alive, and we can inspect the WatchDescriptors below.
+    // Otherwise the `Weak<FdGuard>` contained in the WatchDescriptors will be invalidated
+    // when `stream` is dropped during `await`.
+    let mut watches = stream.watches();
 
-    let watch = stream.watches().add(&path, WatchMask::MODIFY).unwrap();
+    let watch = watches.add(&path, WatchMask::MODIFY).unwrap();
     write_to(&mut file);
 
     let events = stream
@@ -105,6 +119,32 @@ async fn it_should_watch_a_file_from_eventstream_watches() {
     assert!(num_events > 0);
 }
 
+#[cfg(feature = "stream")]
+#[tokio::test]
+async fn it_should_watch_a_file_after_converting_back_from_eventstream() {
+    let mut testdir = TestDir::new();
+    let (path, mut file) = testdir.new_file();
+
+    let inotify = Inotify::init().unwrap();
+
+    let mut buffer = [0; 1024];
+    let stream = inotify.into_event_stream(&mut buffer[..]).unwrap();
+    let mut inotify = stream.into_inotify();
+
+    let watch = inotify.watches().add(&path, WatchMask::MODIFY).unwrap();
+
+    write_to(&mut file);
+
+    let events = inotify.read_events_blocking(&mut buffer).unwrap();
+
+    let mut num_events = 0;
+    for event in events {
+        assert_eq!(watch, event.wd);
+        num_events += 1;
+    }
+    assert!(num_events > 0);
+}
+
 #[test]
 fn it_should_return_immediately_if_no_events_are_available() {
     let mut inotify = Inotify::init().unwrap();