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

fuchsia_fs/directory/
watcher.rs
