fuchsia_fs/directory/
watcher.rs

1// Copyright 2018 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Stream-based Fuchsia VFS directory watcher
6
7#![deny(missing_docs)]
8
9use flex_client::{MessageBuf, ProxyHasDomain};
10use flex_fuchsia_io as fio;
11use futures::stream::{FusedStream, Stream};
12use std::ffi::OsStr;
13use std::os::unix::ffi::OsStrExt;
14use std::path::PathBuf;
15use std::pin::Pin;
16use std::task::{Context, Poll};
17use thiserror::Error;
18
19#[cfg(not(feature = "fdomain"))]
20use fuchsia_async as fasync;
21
22#[derive(Debug, Error, Clone)]
23#[allow(missing_docs)]
24pub enum WatcherCreateError {
25    #[error("while sending watch request: {0}")]
26    SendWatchRequest(#[source] fidl::Error),
27
28    #[error("watch failed with status: {0}")]
29    WatchError(#[source] zx_status::Status),
30
31    #[error("while converting client end to fasync channel: {0}")]
32    ChannelConversion(#[source] zx_status::Status),
33}
34
35#[derive(Debug, Error)]
36#[cfg_attr(not(feature = "fdomain"), derive(Eq, PartialEq))]
37#[allow(missing_docs)]
38pub enum WatcherStreamError {
39    #[cfg(not(feature = "fdomain"))]
40    #[error("read from watch channel failed with status: {0}")]
41    ChannelRead(#[from] zx_status::Status),
42    #[cfg(feature = "fdomain")]
43    #[error("read from watch channel failed: {0}")]
44    ChannelRead(#[from] flex_client::Error),
45}
46
47/// Describes the type of event that occurred in the directory being watched.
48#[repr(C)]
49#[derive(Copy, Clone, Eq, PartialEq)]
50pub struct WatchEvent(fio::WatchEvent);
51
52impl WatchEvent {
53    /// The directory being watched has been deleted. The name returned for this event
54    /// will be `.` (dot), as it is referring to the directory itself.
55    pub const DELETED: Self = Self(fio::WatchEvent::Deleted);
56    /// A file was added.
57    pub const ADD_FILE: Self = Self(fio::WatchEvent::Added);
58    /// A file was removed.
59    pub const REMOVE_FILE: Self = Self(fio::WatchEvent::Removed);
60    /// A file existed at the time the Watcher was created.
61    pub const EXISTING: Self = Self(fio::WatchEvent::Existing);
62    /// All existing files have been enumerated.
63    pub const IDLE: Self = Self(fio::WatchEvent::Idle);
64
65    const fn assoc_const_name(&self) -> &'static str {
66        match self.0 {
67            fio::WatchEvent::Deleted => "DELETED",
68            fio::WatchEvent::Added => "ADD_FILE",
69            fio::WatchEvent::Removed => "REMOVE_FILE",
70            fio::WatchEvent::Existing => "EXISTING",
71            fio::WatchEvent::Idle => "IDLE",
72        }
73    }
74}
75
76impl std::fmt::Debug for WatchEvent {
77    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        write!(f, "WatchEvent({})", self.assoc_const_name())
79    }
80}
81
82/// A message containing a `WatchEvent` and the filename (relative to the directory being watched)
83/// that triggered the event.
84#[derive(Debug, Eq, PartialEq)]
85pub struct WatchMessage {
86    /// The event that occurred.
87    pub event: WatchEvent,
88    /// The filename that triggered the message.
89    pub filename: PathBuf,
90}
91
92#[derive(Debug, Eq, PartialEq)]
93enum WatcherState {
94    Watching,
95    TerminateOnNextPoll,
96    Terminated,
97}
98
99/// Provides a Stream of WatchMessages corresponding to filesystem events for a given directory.
100/// After receiving an error, the stream will return the error, and then will terminate. After it's
101/// terminated, the stream is fused and will continue to return None when polled.
102#[derive(Debug)]
103#[must_use = "futures/streams must be polled"]
104pub struct Watcher {
105    ch: flex_client::AsyncChannel,
106    // If idx >= buf.bytes().len(), you must call reset_buf() before get_next_msg().
107    buf: MessageBuf,
108    idx: usize,
109    state: WatcherState,
110}
111
112impl Unpin for Watcher {}
113
114impl Watcher {
115    /// Creates a new `Watcher` for the directory given by `dir`.
116    pub async fn new(dir: &fio::DirectoryProxy) -> Result<Watcher, WatcherCreateError> {
117        Self::new_with_mask(dir, fio::WatchMask::all()).await
118    }
119
120    /// Creates a new `Watcher` for the directory given by `dir`, only returning events specified
121    /// by `mask`.
122    pub async fn new_with_mask(
123        dir: &fio::DirectoryProxy,
124        mask: fio::WatchMask,
125    ) -> Result<Watcher, WatcherCreateError> {
126        let (client_end, server_end) = dir.domain().create_endpoints();
127        let options = 0u32;
128        let status = dir
129            .watch(mask, options, server_end)
130            .await
131            .map_err(WatcherCreateError::SendWatchRequest)?;
132        zx_status::Status::ok(status).map_err(WatcherCreateError::WatchError)?;
133        let mut buf = MessageBuf::new();
134        buf.ensure_capacity_bytes(fio::MAX_BUF as usize);
135        Ok(Watcher {
136            #[cfg(not(feature = "fdomain"))]
137            ch: fasync::Channel::from_channel(client_end.into_channel()),
138            #[cfg(feature = "fdomain")]
139            ch: client_end.into_channel(),
140            buf,
141            idx: 0,
142            state: WatcherState::Watching,
143        })
144    }
145
146    fn reset_buf(&mut self) {
147        self.idx = 0;
148        self.buf.clear();
149    }
150
151    fn get_next_msg(&mut self) -> WatchMessage {
152        assert!(self.idx < self.buf.bytes().len());
153        let next_msg = VfsWatchMsg::from_raw(&self.buf.bytes()[self.idx..])
154            .expect("Invalid buffer received by Watcher!");
155        self.idx += next_msg.len();
156
157        let mut pathbuf = PathBuf::new();
158        pathbuf.push(OsStr::from_bytes(next_msg.name()));
159        let event = next_msg.event();
160        WatchMessage { event, filename: pathbuf }
161    }
162}
163
164impl FusedStream for Watcher {
165    fn is_terminated(&self) -> bool {
166        self.state == WatcherState::Terminated
167    }
168}
169
170impl Stream for Watcher {
171    type Item = Result<WatchMessage, WatcherStreamError>;
172
173    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
174        let this = &mut *self;
175        // Once this stream has hit an error, it's likely unrecoverable at this level and should be
176        // closed. Clients can attempt to recover by creating a new Watcher.
177        if this.state == WatcherState::TerminateOnNextPoll {
178            this.state = WatcherState::Terminated;
179        }
180        if this.state == WatcherState::Terminated {
181            return Poll::Ready(None);
182        }
183        if this.idx >= this.buf.bytes().len() {
184            this.reset_buf();
185        }
186        if this.idx == 0 {
187            match this.ch.recv_from(cx, &mut this.buf) {
188                Poll::Ready(Ok(())) => {}
189                Poll::Ready(Err(e)) => {
190                    self.state = WatcherState::TerminateOnNextPoll;
191                    return Poll::Ready(Some(Err(e.into())));
192                }
193                Poll::Pending => return Poll::Pending,
194            }
195        }
196        Poll::Ready(Some(Ok(this.get_next_msg())))
197    }
198}
199
200#[repr(C)]
201#[derive(Default)]
202struct IncompleteArrayField<T>(::std::marker::PhantomData<T>);
203impl<T> IncompleteArrayField<T> {
204    #[inline]
205    pub unsafe fn as_ptr(&self) -> *const T {
206        unsafe { ::std::mem::transmute(self) }
207    }
208    #[inline]
209    pub unsafe fn as_slice(&self, len: usize) -> &[T] {
210        unsafe { ::std::slice::from_raw_parts(self.as_ptr(), len) }
211    }
212}
213impl<T> ::std::fmt::Debug for IncompleteArrayField<T> {
214    fn fmt(&self, fmt: &mut ::std::fmt::Formatter<'_>) -> ::std::fmt::Result {
215        fmt.write_str("IncompleteArrayField")
216    }
217}
218
219#[repr(C)]
220#[derive(Debug)]
221struct vfs_watch_msg_t {
222    event: fio::WatchEvent,
223    len: u8,
224    name: IncompleteArrayField<u8>,
225}
226
227#[derive(Debug)]
228struct VfsWatchMsg<'a> {
229    inner: &'a vfs_watch_msg_t,
230}
231
232impl<'a> VfsWatchMsg<'a> {
233    fn from_raw(buf: &'a [u8]) -> Option<VfsWatchMsg<'a>> {
234        if buf.len() < ::std::mem::size_of::<vfs_watch_msg_t>() {
235            return None;
236        }
237        // This is safe as long as the buffer is at least as large as a vfs_watch_msg_t, which we
238        // just verified. Further, we verify that the buffer has enough bytes to hold the
239        // "incomplete array field" member.
240        let m = unsafe { VfsWatchMsg { inner: &*(buf.as_ptr() as *const vfs_watch_msg_t) } };
241        if buf.len() < ::std::mem::size_of::<vfs_watch_msg_t>() + m.namelen() {
242            return None;
243        }
244        Some(m)
245    }
246
247    fn len(&self) -> usize {
248        ::std::mem::size_of::<vfs_watch_msg_t>() + self.namelen()
249    }
250
251    fn event(&self) -> WatchEvent {
252        WatchEvent(self.inner.event)
253    }
254
255    fn namelen(&self) -> usize {
256        self.inner.len as usize
257    }
258
259    fn name(&self) -> &'a [u8] {
260        // This is safe because we verified during construction that the inner name field has at
261        // least namelen() bytes in it.
262        unsafe { self.inner.name.as_slice(self.namelen()) }
263    }
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269    use assert_matches::assert_matches;
270    use fuchsia_async::{DurationExt, TimeoutExt};
271
272    use futures::prelude::*;
273    use std::fmt::Debug;
274    use std::fs::File;
275    use std::path::Path;
276    use std::sync::Arc;
277    use tempfile::tempdir;
278    use vfs::ObjectRequestRef;
279    use vfs::directory::dirents_sink;
280    use vfs::directory::entry::{EntryInfo, GetEntryInfo};
281    use vfs::directory::entry_container::{Directory, DirectoryWatcher};
282    use vfs::directory::immutable::connection::ImmutableConnection;
283    use vfs::directory::traversal_position::TraversalPosition;
284    use vfs::execution_scope::ExecutionScope;
285    use vfs::node::Node;
286
287    fn one_step<'a, S, OK, ERR>(s: &'a mut S) -> impl Future<Output = OK> + 'a
288    where
289        S: Stream<Item = Result<OK, ERR>> + Unpin,
290        ERR: Debug,
291    {
292        let f = s.next();
293        let f = f.on_timeout(zx::MonotonicDuration::from_millis(500).after_now(), || {
294            panic!("timeout waiting for watcher")
295        });
296        f.map(|next| {
297            next.expect("the stream yielded no next item")
298                .unwrap_or_else(|e| panic!("Error waiting for watcher: {:?}", e))
299        })
300    }
301
302    #[fuchsia::test]
303    async fn test_existing() {
304        let tmp_dir = tempdir().unwrap();
305        let _ = File::create(tmp_dir.path().join("file1")).unwrap();
306
307        let dir = crate::directory::open_in_namespace(
308            tmp_dir.path().to_str().unwrap(),
309            fio::PERM_READABLE,
310        )
311        .unwrap();
312        let mut w = Watcher::new(&dir).await.unwrap();
313
314        let msg = one_step(&mut w).await;
315        assert_eq!(WatchEvent::EXISTING, msg.event);
316        assert_eq!(Path::new("."), msg.filename);
317
318        let msg = one_step(&mut w).await;
319        assert_eq!(WatchEvent::EXISTING, msg.event);
320        assert_eq!(Path::new("file1"), msg.filename);
321
322        let msg = one_step(&mut w).await;
323        assert_eq!(WatchEvent::IDLE, msg.event);
324    }
325
326    #[fuchsia::test]
327    async fn test_add() {
328        let tmp_dir = tempdir().unwrap();
329
330        let dir = crate::directory::open_in_namespace(
331            tmp_dir.path().to_str().unwrap(),
332            fio::PERM_READABLE,
333        )
334        .unwrap();
335        let mut w = Watcher::new(&dir).await.unwrap();
336
337        loop {
338            let msg = one_step(&mut w).await;
339            match msg.event {
340                WatchEvent::EXISTING => continue,
341                WatchEvent::IDLE => break,
342                _ => panic!("Unexpected watch event!"),
343            }
344        }
345
346        let _ = File::create(tmp_dir.path().join("file1")).unwrap();
347        let msg = one_step(&mut w).await;
348        assert_eq!(WatchEvent::ADD_FILE, msg.event);
349        assert_eq!(Path::new("file1"), msg.filename);
350    }
351
352    #[fuchsia::test]
353    async fn test_remove() {
354        let tmp_dir = tempdir().unwrap();
355
356        let filename = "file1";
357        let filepath = tmp_dir.path().join(filename);
358        let _ = File::create(&filepath).unwrap();
359
360        let dir = crate::directory::open_in_namespace(
361            tmp_dir.path().to_str().unwrap(),
362            fio::PERM_READABLE,
363        )
364        .unwrap();
365        let mut w = Watcher::new(&dir).await.unwrap();
366
367        loop {
368            let msg = one_step(&mut w).await;
369            match msg.event {
370                WatchEvent::EXISTING => continue,
371                WatchEvent::IDLE => break,
372                _ => panic!("Unexpected watch event!"),
373            }
374        }
375
376        ::std::fs::remove_file(&filepath).unwrap();
377        let msg = one_step(&mut w).await;
378        assert_eq!(WatchEvent::REMOVE_FILE, msg.event);
379        assert_eq!(Path::new(filename), msg.filename);
380    }
381
382    struct MockDirectory;
383
384    impl MockDirectory {
385        fn new() -> Arc<Self> {
386            Arc::new(Self)
387        }
388    }
389
390    impl GetEntryInfo for MockDirectory {
391        fn entry_info(&self) -> EntryInfo {
392            EntryInfo::new(fio::INO_UNKNOWN, fio::DirentType::Directory)
393        }
394    }
395
396    impl Node for MockDirectory {
397        async fn get_attributes(
398            &self,
399            _query: fio::NodeAttributesQuery,
400        ) -> Result<fio::NodeAttributes2, zx::Status> {
401            unimplemented!();
402        }
403
404        fn close(self: Arc<Self>) {}
405    }
406
407    impl Directory for MockDirectory {
408        fn open(
409            self: Arc<Self>,
410            scope: ExecutionScope,
411            _path: vfs::path::Path,
412            flags: fio::Flags,
413            object_request: ObjectRequestRef<'_>,
414        ) -> Result<(), zx::Status> {
415            object_request.take().create_connection_sync::<ImmutableConnection<_>, _>(
416                scope,
417                self.clone(),
418                flags,
419            );
420            Ok(())
421        }
422
423        async fn read_dirents(
424            &self,
425            _pos: &TraversalPosition,
426            _sink: Box<dyn dirents_sink::Sink>,
427        ) -> Result<(TraversalPosition, Box<dyn dirents_sink::Sealed>), zx::Status> {
428            unimplemented!("Not implemented");
429        }
430
431        fn register_watcher(
432            self: Arc<Self>,
433            _scope: ExecutionScope,
434            _mask: fio::WatchMask,
435            _watcher: DirectoryWatcher,
436        ) -> Result<(), zx::Status> {
437            // Don't do anything, just throw out the watcher, which should close the channel, to
438            // generate a PEER_CLOSED error.
439            Ok(())
440        }
441
442        fn unregister_watcher(self: Arc<Self>, _key: usize) {
443            unimplemented!("Not implemented");
444        }
445    }
446
447    #[fuchsia::test]
448    async fn test_error() {
449        let test_dir = MockDirectory::new();
450        let client = vfs::directory::serve_read_only(test_dir, ExecutionScope::new());
451        let mut w = Watcher::new(&client).await.unwrap();
452        let msg = w.next().await.expect("the stream yielded no next item");
453        assert!(!w.is_terminated());
454        assert_matches!(msg, Err(WatcherStreamError::ChannelRead(zx::Status::PEER_CLOSED)));
455        assert!(!w.is_terminated());
456        assert_matches!(w.next().await, None);
457        assert!(w.is_terminated());
458    }
459}
fuchsia_fs/directory/watcher.rs

fuchsia_fs/directory/
watcher.rs
