1 use std::{
2 io,
3 os::unix::io::{AsRawFd, RawFd},
4 pin::Pin,
5 sync::Arc,
6 task::{Context, Poll},
7 };
8
9 use futures_core::{ready, Stream};
10 use tokio::io::unix::AsyncFd;
11
12 use crate::events::{Event, EventOwned};
13 use crate::fd_guard::FdGuard;
14 use crate::Inotify;
15 use crate::util::read_into_buffer;
16 use crate::watches::Watches;
17
18 /// Stream of inotify events
19 ///
20 /// Allows for streaming events returned by [`Inotify::into_event_stream`].
21 #[derive(Debug)]
22 pub struct EventStream<T> {
23 fd: AsyncFd<ArcFdGuard>,
24 buffer: T,
25 buffer_pos: usize,
26 unused_bytes: usize,
27 }
28
29 impl<T> EventStream<T>
30 where
31 T: AsMut<[u8]> + AsRef<[u8]>,
32 {
33 /// Returns a new `EventStream` associated with the default reactor.
new(fd: Arc<FdGuard>, buffer: T) -> io::Result<Self>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`].
watches(&self) -> Watches45 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`.
into_inotify(self) -> Inotify51 pub fn into_inotify(self) -> Inotify {
52 Inotify::from_file_descriptor(self.fd.into_inner().0)
53 }
54 }
55
56 impl<T> Stream for EventStream<T>
57 where
58 T: AsMut<[u8]> + AsRef<[u8]>,
59 {
60 type Item = io::Result<EventOwned>;
61
poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>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)]
94 struct ArcFdGuard(Arc<FdGuard>);
95
96 impl AsRawFd for ArcFdGuard {
as_raw_fd(&self) -> RawFd97 fn as_raw_fd(&self) -> RawFd {
98 self.0.as_raw_fd()
99 }
100 }
101
read(fd: &AsyncFd<ArcFdGuard>, buffer: &mut [u8], cx: &mut Context) -> Poll<io::Result<usize>>102 fn 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 }
121