vfs/directory/watchers/
watcher.rs

1// Copyright 2019 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//! A task that is run to process communication with an individual watcher.
6
7use crate::directory::entry_container::DirectoryWatcher;
8use crate::directory::watchers::event_producers::EventProducer;
9use crate::execution_scope::ExecutionScope;
10
11use flex_fuchsia_io as fio;
12use futures::channel::mpsc::{self, UnboundedSender};
13use futures::{FutureExt, select};
14
15#[derive(Clone)]
16pub struct Controller {
17    mask: fio::WatchMask,
18    messages: UnboundedSender<Vec<u8>>,
19}
20
21impl Controller {
22    /// `done` is not guaranteed to be called if the task failed to start.  It should only happen
23    /// in case the return value is an `Err`.  Unfortunately, there is no way to return the `done`
24    /// object itself, as the [`futures::Spawn::spawn_obj`] does not return the ownership in case
25    /// of a failure.
26    pub(crate) fn new(
27        scope: ExecutionScope,
28        mask: fio::WatchMask,
29        watcher: DirectoryWatcher,
30        done: impl FnOnce() + Send + 'static,
31    ) -> Controller {
32        use futures::StreamExt as _;
33
34        let (sender, mut receiver) = mpsc::unbounded::<Vec<u8>>();
35        let done = CallOnDrop(Some(done));
36        let task = async move {
37            #[cfg(not(feature = "fdomain"))]
38            let mut recv_msg = std::pin::pin!(
39                fuchsia_async::OnSignals::new(
40                    watcher.channel(),
41                    fidl::Signals::CHANNEL_READABLE | fidl::Signals::CHANNEL_PEER_CLOSED,
42                )
43                .fuse()
44            );
45            #[cfg(feature = "fdomain")]
46            let mut recv_msg = watcher.channel().recv_msg().fuse();
47            loop {
48                select! {
49                    command = receiver.next() => match command {
50                        Some(message) => {
51                            #[cfg(not(feature = "fdomain"))]
52                            let result = watcher.channel().write(&*message, &mut []);
53                            #[cfg(feature = "fdomain")]
54                            let result = watcher.channel().write(&*message, std::vec::Vec::new());
55                            if result.is_err() {
56                                break;
57                            }
58                        },
59                        None => break,
60                    },
61                    _ = recv_msg => {
62                        // We do not expect any messages to be received over the watcher connection.
63                        // Should we receive a message we will close the connection to indicate an
64                        // error.  If any error occurs, we also close the connection.  And if the
65                        // connection is closed, we just stop the command processing as well.
66                        break;
67                    },
68                }
69            }
70            // The purpose of this line is to reference `done` within the async closure so the async
71            // closure will take ownership of it. Doing `let done = done;` causes `done` to take up
72            // twice the space in the generated Future.
73            std::mem::drop(done);
74        };
75
76        scope.spawn(task);
77        Controller { mask, messages: sender }
78    }
79
80    /// Sends a buffer to the connected watcher.  `mask` specifies the type of the event the buffer
81    /// is for.  If the watcher mask does not include the event specified by the `mask` then the
82    /// buffer is not sent and `buffer` is not even invoked.
83    pub(crate) fn send_buffer(&self, mask: fio::WatchMask, buffer: impl FnOnce() -> Vec<u8>) {
84        if !self.mask.intersects(mask) {
85            return;
86        }
87
88        if self.messages.unbounded_send(buffer()).is_ok() {
89            return;
90        }
91
92        // An error to send indicates the execution task has been disconnected.  Controller should
93        // always be removed from the watchers list before it is destroyed.  So this is some
94        // logical bug.
95        debug_assert!(false, "Watcher controller failed to send a command to the watcher.");
96    }
97
98    /// Uses a `producer` to generate one or more buffers and send them all to the connected
99    /// watcher.  `producer.mask()` is used to determine the type of the event - in case the
100    /// watcher mask does not specify that it needs to receive this event, then the producer is not
101    /// used and `false` is returned.  If the producers mask and the watcher mask overlap, then
102    /// `true` is returned (even if the producer did not generate a single buffer).
103    pub fn send_event(&self, producer: &mut dyn EventProducer) -> bool {
104        if !self.mask.intersects(producer.mask()) {
105            return false;
106        }
107
108        while producer.prepare_for_next_buffer() {
109            let buffer = producer.buffer();
110            if self.messages.unbounded_send(buffer).is_ok() {
111                continue;
112            }
113
114            // An error to send indicates the execution task has been disconnected.  Controller
115            // should always be removed from the watchers list before it is destroyed.  So this is
116            // some logical bug.
117            debug_assert!(false, "Watcher controller failed to send a command to the watcher.");
118        }
119
120        return true;
121    }
122}
123
124/// Calls the function when this object is dropped.
125struct CallOnDrop<F: FnOnce()>(Option<F>);
126
127impl<F: FnOnce()> Drop for CallOnDrop<F> {
128    fn drop(&mut self) {
129        self.0.take().unwrap()();
130    }
131}
vfs/directory/watchers/watcher.rs

vfs/directory/watchers/
watcher.rs
