vfs/directory/watchers/
watcher.rs

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

vfs/directory/watchers/
watcher.rs
