settings_test_common/fakes/
camera3_service.rs

1// Copyright 2021 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
5use crate::fakes::service::Service;
6use anyhow::{Error, format_err};
7use fidl::endpoints::ServerEnd;
8use fidl::prelude::*;
9use fidl_fuchsia_camera3::{
10    DeviceRequest, DeviceWatchMuteStateResponder, DeviceWatcherMarker, DeviceWatcherRequest,
11    DeviceWatcherWatchDevicesResponder, WatchDevicesEvent,
12};
13use fuchsia_async::{self as fasync, DurationExt};
14use futures::TryStreamExt;
15use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
16use futures::future::{AbortHandle, Abortable};
17use settings_camera::CAMERA_WATCHER_TIMEOUT;
18use std::cell::{Cell, RefCell};
19use std::collections::BTreeMap;
20use std::rc::Rc;
21use std::sync::atomic::{AtomicBool, Ordering};
22use zx::MonotonicDuration;
23
24/// The last mute state sent on a camera device connection, if any.
25type LastSentMuted = Rc<Cell<Option<bool>>>;
26
27/// Software mute state shared by all camera device connections. `WatchMuteState` follows
28/// hanging-get semantics: the first call on a connection returns immediately, and later calls
29/// return once the state differs from the last value sent on that connection.
30#[derive(Default)]
31struct SwMuteState {
32    muted: Cell<bool>,
33    pending_watchers: RefCell<Vec<(LastSentMuted, DeviceWatchMuteStateResponder)>>,
34    abort_handles: RefCell<Vec<AbortHandle>>,
35}
36
37impl SwMuteState {
38    fn get(&self) -> bool {
39        self.muted.get()
40    }
41
42    fn set(&self, muted: bool) {
43        if self.muted.replace(muted) == muted {
44            return;
45        }
46        for (last_sent, responder) in self.pending_watchers.borrow_mut().drain(..) {
47            last_sent.set(Some(muted));
48            let _ = responder.send(muted, false);
49        }
50    }
51
52    fn watch(&self, last_sent: &LastSentMuted, responder: DeviceWatchMuteStateResponder) {
53        let muted = self.muted.get();
54        if last_sent.get() == Some(muted) {
55            self.pending_watchers.borrow_mut().push((Rc::clone(last_sent), responder));
56        } else {
57            last_sent.set(Some(muted));
58            let _ = responder.send(muted, false);
59        }
60    }
61
62    fn disconnect(&self) {
63        self.pending_watchers.borrow_mut().clear();
64        for handle in self.abort_handles.borrow_mut().drain(..) {
65            handle.abort();
66        }
67    }
68}
69
70/// Device changes shared by all device watcher connections. Once a connection has received a
71/// nonempty device list, later `WatchDevices` calls on it return once devices are added or
72/// removed.
73#[derive(Default)]
74struct DeviceChanges {
75    connections: RefCell<Vec<std::rc::Weak<WatcherConnection>>>,
76}
77
78#[derive(Default)]
79struct WatcherConnection {
80    initial_devices_sent: Cell<bool>,
81    queued: RefCell<Vec<WatchDevicesEvent>>,
82    pending: RefCell<Option<DeviceWatcherWatchDevicesResponder>>,
83}
84
85impl WatcherConnection {
86    fn flush(&self) {
87        if self.queued.borrow().is_empty() {
88            return;
89        }
90        if let Some(responder) = self.pending.borrow_mut().take() {
91            let events: Vec<_> = self.queued.borrow_mut().drain(..).collect();
92            let _ = responder.send(&events);
93        }
94    }
95}
96
97impl DeviceChanges {
98    fn add(&self, event: WatchDevicesEvent) {
99        self.connections.borrow_mut().retain(|connection| {
100            let Some(connection) = connection.upgrade() else {
101                return false;
102            };
103            if connection.initial_devices_sent.get() {
104                connection.queued.borrow_mut().push(event);
105                connection.flush();
106            }
107            true
108        });
109    }
110}
111
112/// Software mute state of each camera device, by id.
113type Devices = Rc<RefCell<BTreeMap<u64, Rc<SwMuteState>>>>;
114
115/// Returns the initial devices, with a single device with id 1.
116fn initial_devices() -> Devices {
117    Rc::new(RefCell::new(BTreeMap::from([(1, Rc::new(SwMuteState::default()))])))
118}
119
120pub struct Camera3Service {
121    devices: Devices,
122    device_changes: Rc<DeviceChanges>,
123    mute_listeners: Rc<RefCell<Vec<UnboundedSender<bool>>>>,
124    // If true, sends the camera device response immediately. If false, sends
125    // an empty device list. If None, delay_camera_device should be provided.
126    has_camera_device: Rc<AtomicBool>,
127    // If true, first sends an empty device list on watch, then delays, then
128    // sends the camera device response.
129    delay_camera_device: Rc<AtomicBool>,
130}
131
132impl Camera3Service {
133    pub fn new(has_camera_device: bool) -> Self {
134        Self {
135            devices: initial_devices(),
136            device_changes: Rc::new(DeviceChanges::default()),
137            mute_listeners: Rc::new(RefCell::new(Vec::new())),
138            has_camera_device: Rc::new(AtomicBool::new(has_camera_device)),
139            delay_camera_device: Rc::new(AtomicBool::new(false)),
140        }
141    }
142
143    pub fn new_delayed_devices(delay_camera_device: bool) -> Self {
144        Self {
145            devices: initial_devices(),
146            device_changes: Rc::new(DeviceChanges::default()),
147            mute_listeners: Rc::new(RefCell::new(Vec::new())),
148            has_camera_device: Rc::new(AtomicBool::new(false)),
149            delay_camera_device: Rc::new(AtomicBool::new(delay_camera_device)),
150        }
151    }
152
153    /// Returns whether all camera devices are software muted.
154    pub fn camera_sw_muted(&self) -> bool {
155        self.devices.borrow().values().all(|device| device.get())
156    }
157
158    /// Returns a receiver that is sent whether all camera devices are software muted whenever a
159    /// `SetSoftwareMuteState` request is handled.
160    pub fn watch_sw_mute_changes(&self) -> UnboundedReceiver<bool> {
161        let (tx, rx) = mpsc::unbounded();
162        self.mute_listeners.borrow_mut().push(tx);
163        rx
164    }
165
166    /// Sets the software mute state of all camera devices as if it were changed by another
167    /// client of the camera, and notifies pending `WatchMuteState` calls.
168    pub fn set_camera_sw_muted(&self, muted: bool) {
169        for device in self.devices.borrow().values() {
170            device.set(muted);
171        }
172    }
173
174    /// Adds a camera device with the given id and software mute state, and reports it as added
175    /// to device watcher connections that have already received the initial device list.
176    pub fn add_camera_device(&self, id: u64, sw_muted: bool) {
177        let device = Rc::new(SwMuteState::default());
178        device.set(sw_muted);
179        if let Some(old_device) = self.devices.borrow_mut().insert(id, device) {
180            old_device.disconnect();
181        }
182        self.device_changes.add(WatchDevicesEvent::Added(id));
183    }
184
185    /// Removes the camera device with the given id, closes its connections, and reports it as
186    /// removed to device watcher connections that have already received the initial device list.
187    pub fn remove_camera_device(&self, id: u64) {
188        if let Some(device) = self.devices.borrow_mut().remove(&id) {
189            device.disconnect();
190            self.device_changes.add(WatchDevicesEvent::Removed(id));
191        }
192    }
193}
194
195impl Service for Camera3Service {
196    fn can_handle_service(&self, service_name: &str) -> bool {
197        service_name == DeviceWatcherMarker::PROTOCOL_NAME
198    }
199
200    fn process_stream(&mut self, service_name: &str, channel: zx::Channel) -> Result<(), Error> {
201        if !self.can_handle_service(service_name) {
202            return Err(format_err!("can't handle service"));
203        }
204
205        let mut device_watcher_stream =
206            ServerEnd::<DeviceWatcherMarker>::new(channel).into_stream();
207
208        let devices = Rc::clone(&self.devices);
209        let mute_listeners = Rc::clone(&self.mute_listeners);
210        let has_camera_device = Rc::clone(&self.has_camera_device);
211        let delay_camera_device = Rc::clone(&self.delay_camera_device);
212        let connection = Rc::new(WatcherConnection::default());
213        self.device_changes.connections.borrow_mut().push(Rc::downgrade(&connection));
214        let mut watch_count = 0;
215        fasync::Task::local(async move {
216            while let Some(req) = device_watcher_stream.try_next().await.unwrap() {
217                // Support future expansion of FIDL.
218                #[allow(unreachable_patterns)]
219                match req {
220                    DeviceWatcherRequest::WatchDevices { responder } => {
221                        if connection.initial_devices_sent.get() {
222                            let _ = connection.pending.borrow_mut().replace(responder);
223                            connection.flush();
224                            continue;
225                        }
226                        let camera_devices: Vec<_> = devices
227                            .borrow()
228                            .keys()
229                            .map(|id| WatchDevicesEvent::Added(*id))
230                            .collect();
231                        let camera_devices = &camera_devices[..];
232                        let empty_devices = &[][..];
233
234                        let devices = if delay_camera_device.load(Ordering::Relaxed) {
235                            if watch_count == 0 {
236                                watch_count += 1;
237                                empty_devices
238                            } else {
239                                let timer = fasync::Timer::new(
240                                    MonotonicDuration::from_millis(CAMERA_WATCHER_TIMEOUT / 2)
241                                        .after_now(),
242                                );
243                                timer.await;
244                                camera_devices
245                            }
246                        } else if has_camera_device.load(Ordering::Relaxed) {
247                            camera_devices
248                        } else {
249                            empty_devices
250                        };
251                        connection.initial_devices_sent.set(!devices.is_empty());
252                        responder.send(devices).expect("Failed to send devices response");
253                    }
254                    DeviceWatcherRequest::ConnectToDevice {
255                        id,
256                        request, // ServerEnd<DeviceMarker>
257                        control_handle: _,
258                    } => {
259                        let mut stream = request.into_stream();
260                        let Some(camera_sw_muted) = devices.borrow().get(&id).cloned() else {
261                            continue;
262                        };
263                        let devices = Rc::clone(&devices);
264                        let mute_listeners = Rc::clone(&mute_listeners);
265                        let (abort_handle, abort_registration) = AbortHandle::new_pair();
266                        camera_sw_muted.abort_handles.borrow_mut().push(abort_handle);
267                        fasync::Task::local(async move {
268                            let _ = Abortable::new(
269                                async move {
270                                    let last_sent_muted = Rc::new(Cell::new(None));
271                                    while let Some(req) = stream.try_next().await.unwrap() {
272                                        // Support future expansion of FIDL.
273                                        match req {
274                                            DeviceRequest::SetSoftwareMuteState {
275                                                muted,
276                                                responder,
277                                            } => {
278                                                camera_sw_muted.set(muted);
279                                                let _ = responder.send();
280                                                let all_muted = devices
281                                                    .borrow()
282                                                    .values()
283                                                    .all(|device| device.get());
284                                                mute_listeners.borrow_mut().retain(|tx| {
285                                                    tx.unbounded_send(all_muted).is_ok()
286                                                });
287                                            }
288                                            DeviceRequest::WatchMuteState { responder } => {
289                                                camera_sw_muted.watch(&last_sent_muted, responder);
290                                            }
291                                            _ => {}
292                                        }
293                                    }
294                                },
295                                abort_registration,
296                            )
297                            .await;
298                        })
299                        .detach();
300                    }
301                    _ => {}
302                }
303            }
304        })
305        .detach();
306
307        Ok(())
308    }
309}
settings_test_common/fakes/camera3_service.rs

settings_test_common/fakes/
camera3_service.rs
