settings_common/inspect/
event.rs

1// Copyright 2025 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 super::listener_logger::ListenerInspectLogger;
6use crate::service_context::ExternalServiceEvent;
7use anyhow::{Error, anyhow};
8use futures::channel::mpsc::UnboundedSender;
9use std::cell::Cell;
10use std::marker::PhantomData;
11use std::rc::Rc;
12
13#[derive(Clone)]
14pub enum Direction {
15    Request(String),
16    Response(String, ResponseType),
17}
18
19#[derive(Clone)]
20pub struct UsageEvent {
21    pub setting: &'static str,
22    pub request_type: RequestType,
23    pub direction: Direction,
24    pub id: u64,
25}
26
27pub trait Nameable {
28    const NAME: &str;
29}
30
31#[derive(Clone)]
32pub struct UsagePublisher<T> {
33    id_gen: Rc<Cell<u64>>,
34    inspect_tx: UnboundedSender<UsageEvent>,
35    listener_logger: Rc<ListenerInspectLogger>,
36    _phantom: PhantomData<T>,
37}
38
39impl<T> UsagePublisher<T> {
40    pub fn new(
41        inspect_tx: UnboundedSender<UsageEvent>,
42        listener_logger: Rc<ListenerInspectLogger>,
43    ) -> Self {
44        Self { id_gen: Rc::new(Cell::new(0)), inspect_tx, listener_logger, _phantom: PhantomData }
45    }
46}
47
48impl<T> UsagePublisher<T>
49where
50    T: Nameable,
51{
52    pub fn request(&self, request: String, request_type: RequestType) -> UsageResponsePublisher<T> {
53        let id = self.id_gen.get();
54        self.id_gen.set(id.wrapping_add(1));
55        let _ = self.inspect_tx.unbounded_send(UsageEvent {
56            setting: T::NAME,
57            request_type,
58            direction: Direction::Request(request),
59            id,
60        });
61        if let RequestType::Get = request_type {
62            self.listener_logger.add_listener(T::NAME.into());
63        }
64
65        UsageResponsePublisher {
66            id,
67            request_type,
68            inspect_tx: self.inspect_tx.clone(),
69            listener_logger: Rc::clone(&self.listener_logger),
70            sent: false,
71            _phantom: PhantomData,
72        }
73    }
74}
75
76#[derive(Clone)]
77pub struct UsageResponsePublisher<T> {
78    id: u64,
79    request_type: RequestType,
80    inspect_tx: UnboundedSender<UsageEvent>,
81    listener_logger: Rc<ListenerInspectLogger>,
82    sent: bool,
83    _phantom: PhantomData<T>,
84}
85
86impl<T> UsageResponsePublisher<T>
87where
88    T: Nameable,
89{
90    pub fn respond(mut self, response: String, response_type: ResponseType) {
91        let _ = self.inspect_tx.unbounded_send(UsageEvent {
92            setting: T::NAME,
93            request_type: self.request_type,
94            direction: Direction::Response(response, response_type),
95            id: self.id,
96        });
97        if let RequestType::Get = self.request_type {
98            self.listener_logger.remove_listener(T::NAME.into());
99        }
100        self.sent = true;
101    }
102}
103
104impl<T> Drop for UsageResponsePublisher<T> {
105    fn drop(&mut self) {
106        if !self.sent {
107            log::error!("UsageResponsePublisher dropped without sending response");
108        }
109    }
110}
111
112/// A wrapper for hanging get observers that ensures the `UsageResponsePublisher`
113/// is responded to even if the observer is dropped (e.g., due to client disconnect).
114pub struct HangingGetObserver<T, R>
115where
116    T: Nameable,
117{
118    inner: Option<(UsageResponsePublisher<T>, R)>,
119}
120
121impl<T, R> HangingGetObserver<T, R>
122where
123    T: Nameable,
124{
125    pub fn new(publisher: UsageResponsePublisher<T>, responder: R) -> Self {
126        Self { inner: Some((publisher, responder)) }
127    }
128
129    /// Deconstruct the observer to retrieve the publisher and responder.
130    /// This consumes the wrapper and prevents the automatic "Cancelled" response on drop.
131    pub fn into_parts(mut self) -> (UsageResponsePublisher<T>, R) {
132        self.inner.take().expect("inner components present")
133    }
134}
135
136impl<T, R> Drop for HangingGetObserver<T, R>
137where
138    T: Nameable,
139{
140    fn drop(&mut self) {
141        if let Some((publisher, _)) = self.inner.take() {
142            publisher.respond("Cancelled".to_string(), ResponseType::Cancelled);
143        }
144    }
145}
146
147#[derive(Copy, Clone, Debug)]
148pub enum RequestType {
149    Get,
150    Set,
151    Camera,
152    MediaButtons,
153}
154
155#[derive(Debug, Copy, Clone)]
156/// Response type to a request to a setting. Used for accumulating response type
157/// counts for inspect. This should be updated to have a matching error for each
158/// of the controller error variants.
159pub enum ResponseType {
160    OkSome,
161    OkNone,
162    UnimplementedRequest,
163    StorageFailure,
164    InitFailure,
165    RestoreFailure,
166    InvalidArgument,
167    IncompatibleArguments,
168    ExternalFailure,
169    UnhandledType,
170    DeliveryError,
171    UnexpectedError,
172    UndeliverableError,
173    UnsupportedError,
174    CommunicationError,
175    IrrecoverableError,
176    TimeoutError,
177    AlreadySubscribed,
178    MaximumInputDevicesReached,
179    Cancelled,
180}
181
182#[derive(Copy, Clone, Debug)]
183pub struct MediaButtons {
184    pub mic_mute: Option<bool>,
185    pub camera_disable: Option<bool>,
186}
187
188#[derive(Clone)]
189pub struct SettingValuePublisher<T> {
190    tx: UnboundedSender<(&'static str, String)>,
191    _phantom: PhantomData<T>,
192}
193
194impl<T> SettingValuePublisher<T> {
195    pub fn new(tx: UnboundedSender<(&'static str, String)>) -> Self {
196        Self { tx, _phantom: PhantomData }
197    }
198}
199
200impl<T> SettingValuePublisher<T>
201where
202    T: Nameable + std::fmt::Debug,
203{
204    pub fn publish(&self, value: &T) -> Result<(), Error> {
205        self.tx
206            .unbounded_send((T::NAME, format!("{value:?}")))
207            .map_err(|e| anyhow!("Unable to send setting_value update: {e:?}"))
208    }
209}
210
211#[derive(Clone)]
212pub struct ExternalEventPublisher {
213    tx: UnboundedSender<ExternalServiceEvent>,
214}
215
216impl ExternalEventPublisher {
217    pub fn new(tx: UnboundedSender<ExternalServiceEvent>) -> Self {
218        Self { tx }
219    }
220
221    pub fn publish(&self, event: ExternalServiceEvent) -> Result<(), Error> {
222        self.tx
223            .unbounded_send(event)
224            .map_err(|e| anyhow!("Unable to send external event update: {e:?}"))
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231    use crate::inspect::listener_logger::ListenerInspectLogger;
232    use futures::channel::mpsc;
233
234    struct TestSetting;
235    impl Nameable for TestSetting {
236        const NAME: &'static str = "TestSetting";
237    }
238
239    #[test]
240    fn test_observer_into_parts() {
241        let (tx, _rx) = mpsc::unbounded();
242        let logger = Rc::new(ListenerInspectLogger::new());
243        let publisher = UsagePublisher::<TestSetting>::new(tx, logger);
244        let response_publisher = publisher.request("Watch".to_string(), RequestType::Get);
245        let dummy_responder = 42;
246
247        let observer = HangingGetObserver::new(response_publisher, dummy_responder);
248        let (pub_out, resp_out) = observer.into_parts();
249        assert_eq!(resp_out, 42);
250
251        // Mark sent so drop doesn't log error
252        pub_out.respond("Ok".to_string(), ResponseType::OkSome);
253    }
254
255    #[test]
256    fn test_observer_drop_cancels() {
257        let (tx, mut rx) = mpsc::unbounded();
258        let logger = Rc::new(ListenerInspectLogger::new());
259        let publisher = UsagePublisher::<TestSetting>::new(tx, logger);
260        let response_publisher = publisher.request("Watch".to_string(), RequestType::Get);
261
262        let observer = HangingGetObserver::new(response_publisher, ());
263        drop(observer);
264
265        // Verify request event was sent
266        let req_event = rx.try_next().unwrap().unwrap();
267        assert_eq!(req_event.setting, "TestSetting");
268        assert!(matches!(req_event.direction, Direction::Request(_)));
269
270        // Verify response event was sent on drop with Cancelled status
271        let resp_event = rx.try_next().unwrap().unwrap();
272        assert_eq!(resp_event.setting, "TestSetting");
273        if let Direction::Response(msg, status) = resp_event.direction {
274            assert_eq!(msg, "Cancelled");
275            assert!(matches!(status, ResponseType::Cancelled));
276        } else {
277            panic!("Expected Direction::Response");
278        }
279    }
280}
settings_common/inspect/event.rs

settings_common/inspect/
event.rs
