inotify/
stream.rs

1use std::{
2    io,
3    os::unix::io::{AsRawFd, RawFd},
4    pin::Pin,
5    sync::Arc,
6    task::{Context, Poll},
7};
8
9use futures_core::{ready, Stream};
10use tokio::io::unix::AsyncFd;
11
12use crate::events::{Event, EventOwned};
13use crate::fd_guard::FdGuard;
14use crate::Inotify;
15use crate::util::read_into_buffer;
16use crate::watches::Watches;
17
18/// Stream of inotify events
19///
20/// Allows for streaming events returned by [`Inotify::into_event_stream`].
21#[derive(Debug)]
22pub struct EventStream<T> {
23    fd: AsyncFd<ArcFdGuard>,
24    buffer: T,
25    buffer_pos: usize,
26    unused_bytes: usize,
27}
28
29impl<T> EventStream<T>
30where
31    T: AsMut<[u8]> + AsRef<[u8]>,
32{
33    /// Returns a new `EventStream` associated with the default reactor.
34    pub(crate) fn new(fd: Arc<FdGuard>, buffer: T) -> io::Result<Self> {
35        Ok(EventStream {
36            fd: AsyncFd::new(ArcFdGuard(fd))?,
37            buffer,
38            buffer_pos: 0,
39            unused_bytes: 0,
40        })
41    }
42
43    /// Returns an instance of `Watches` to add and remove watches.
44    /// See [`Watches::add`] and [`Watches::remove`].
45    pub fn watches(&self) -> Watches {
46        Watches::new(self.fd.get_ref().0.clone())
47    }
48
49    /// Consumes the `EventStream` instance and returns an `Inotify` using the original
50    /// file descriptor that was passed from `Inotify` to create the `EventStream`.
51    pub fn into_inotify(self) -> Inotify {
52        Inotify::from_file_descriptor(self.fd.into_inner().0)
53    }
54}
55
56impl<T> Stream for EventStream<T>
57where
58    T: AsMut<[u8]> + AsRef<[u8]>,
59{
60    type Item = io::Result<EventOwned>;
61
62    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
63        // Safety: safe because we never move out of `self_`.
64        let self_ = unsafe { self.get_unchecked_mut() };
65
66        if self_.unused_bytes == 0 {
67            // Nothing usable in buffer. Need to reset and fill buffer.
68            self_.buffer_pos = 0;
69            self_.unused_bytes = ready!(read(&self_.fd, self_.buffer.as_mut(), cx))?;
70        }
71
72        if self_.unused_bytes == 0 {
73            // The previous read returned `0` signalling end-of-file. Let's
74            // signal end-of-stream to the caller.
75            return Poll::Ready(None);
76        }
77
78        // We have bytes in the buffer. inotify doesn't put partial events in
79        // there, and we only take complete events out. That means we have at
80        // least one event in there and can call `from_buffer` to take it out.
81        let (bytes_consumed, event) = Event::from_buffer(
82            Arc::downgrade(&self_.fd.get_ref().0),
83            &self_.buffer.as_ref()[self_.buffer_pos..],
84        );
85        self_.buffer_pos += bytes_consumed;
86        self_.unused_bytes -= bytes_consumed;
87
88        Poll::Ready(Some(Ok(event.to_owned())))
89    }
90}
91
92// Newtype wrapper because AsRawFd isn't implemented for Arc<T> where T: AsRawFd.
93#[derive(Debug)]
94struct ArcFdGuard(Arc<FdGuard>);
95
96impl AsRawFd for ArcFdGuard {
97    fn as_raw_fd(&self) -> RawFd {
98        self.0.as_raw_fd()
99    }
100}
101
102fn read(fd: &AsyncFd<ArcFdGuard>, buffer: &mut [u8], cx: &mut Context) -> Poll<io::Result<usize>> {
103    let mut guard = ready!(fd.poll_read_ready(cx))?;
104    let result = guard.try_io(|_| {
105        let read = read_into_buffer(fd.as_raw_fd(), buffer);
106        if read == -1 {
107            return Err(io::Error::last_os_error());
108        }
109
110        Ok(read as usize)
111    });
112
113    match result {
114        Ok(result) => Poll::Ready(result),
115        Err(_would_block) => {
116            cx.waker().wake_by_ref();
117            Poll::Pending
118        }
119    }
120}
inotify/stream.rs

inotify/
stream.rs
