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

fuchsia_fs/directory/
watcher.rs
