settings_intl/
intl_fidl_handler.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
5use super::intl_controller::{IntlController, IntlError, Request};
6use super::types::IntlInfo;
7use async_utils::hanging_get::server;
8use fidl_fuchsia_settings::{
9    Error as SettingsError, IntlRequest, IntlRequestStream, IntlSettings, IntlWatchResponder,
10};
11use fuchsia_async as fasync;
12use futures::StreamExt;
13use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
14use futures::channel::oneshot;
15use settings_common::inspect::event::{
16    HangingGetObserver, RequestType, ResponseType, UsagePublisher,
17};
18
19pub(super) type SubscriberObject = HangingGetObserver<IntlInfo, IntlWatchResponder>;
20type HangingGetFn = fn(&IntlInfo, SubscriberObject) -> bool;
21pub(super) type HangingGet = server::HangingGet<IntlInfo, SubscriberObject, HangingGetFn>;
22pub(super) type Publisher = server::Publisher<IntlInfo, SubscriberObject, HangingGetFn>;
23pub(super) type Subscriber = server::Subscriber<IntlInfo, SubscriberObject, HangingGetFn>;
24
25pub struct IntlFidlHandler {
26    hanging_get: HangingGet,
27    controller_tx: UnboundedSender<Request>,
28    usage_publisher: UsagePublisher<IntlInfo>,
29}
30
31impl IntlFidlHandler {
32    pub(crate) fn new(
33        intl_controller: &mut IntlController,
34        usage_publisher: UsagePublisher<IntlInfo>,
35        initial_value: IntlInfo,
36    ) -> (Self, UnboundedReceiver<Request>) {
37        let hanging_get = HangingGet::new(initial_value, Self::hanging_get);
38        intl_controller.register_publisher(hanging_get.new_publisher());
39        let (controller_tx, controller_rx) = mpsc::unbounded();
40        (Self { hanging_get, controller_tx, usage_publisher }, controller_rx)
41    }
42
43    fn hanging_get(info: &IntlInfo, observer: SubscriberObject) -> bool {
44        let (usage_responder, responder) = observer.into_parts();
45        usage_responder.respond(format!("{info:?}"), ResponseType::OkSome);
46        if let Err(e) = responder.send(&IntlSettings::from(info.clone())) {
47            log::warn!("Failed to respond to watch request: {e:?}");
48            return false;
49        }
50        true
51    }
52
53    pub fn handle_stream(&mut self, mut stream: IntlRequestStream) {
54        let request_handler = RequestHandler {
55            subscriber: self.hanging_get.new_subscriber(),
56            controller_tx: self.controller_tx.clone(),
57            usage_publisher: self.usage_publisher.clone(),
58        };
59        fasync::Task::local(async move {
60            while let Some(Ok(request)) = stream.next().await {
61                request_handler.handle_request(request).await;
62            }
63        })
64        .detach();
65    }
66}
67
68#[derive(Debug)]
69enum HandlerError {
70    AlreadySubscribed,
71    ControllerStopped,
72    Controller(IntlError),
73}
74
75impl From<&HandlerError> for ResponseType {
76    fn from(error: &HandlerError) -> Self {
77        match error {
78            HandlerError::AlreadySubscribed => ResponseType::AlreadySubscribed,
79            HandlerError::ControllerStopped => ResponseType::UnexpectedError,
80            HandlerError::Controller(e) => ResponseType::from(e),
81        }
82    }
83}
84
85struct RequestHandler {
86    subscriber: Subscriber,
87    controller_tx: UnboundedSender<Request>,
88    usage_publisher: UsagePublisher<IntlInfo>,
89}
90
91impl RequestHandler {
92    async fn handle_request(&self, request: IntlRequest) {
93        match request {
94            IntlRequest::Watch { responder } => {
95                let usage_res = self.usage_publisher.request("Watch".to_string(), RequestType::Get);
96                let observer = HangingGetObserver::new(usage_res, responder);
97                if let Err(observer) = self.subscriber.register2(observer) {
98                    let e = HandlerError::AlreadySubscribed;
99                    let (usage_res, responder) = observer.into_parts();
100                    usage_res.respond(format!("Err({e:?})"), ResponseType::from(&e));
101                    drop(responder);
102                }
103            }
104            IntlRequest::Set { settings, responder } => {
105                let usage_res = self
106                    .usage_publisher
107                    .request(format!("Set{{settings:{settings:?}}}"), RequestType::Set);
108                if let Err(e) = self.set(settings).await {
109                    usage_res.respond(format!("Err({e:?}"), ResponseType::from(&e));
110                    let _ = responder.send(Err(SettingsError::Failed));
111                } else {
112                    usage_res.respond("Ok(())".to_string(), ResponseType::OkNone);
113                    let _ = responder.send(Ok(()));
114                }
115            }
116        }
117    }
118
119    async fn set(&self, settings: IntlSettings) -> Result<(), HandlerError> {
120        let (set_tx, set_rx) = oneshot::channel();
121        self.controller_tx
122            .unbounded_send(Request::Set(IntlInfo::from(settings), set_tx))
123            .map_err(|_| HandlerError::ControllerStopped)?;
124        set_rx
125            .await
126            .map_err(|_| HandlerError::ControllerStopped)
127            .and_then(|res| res.map_err(HandlerError::Controller))
128    }
129}
settings_intl/intl_fidl_handler.rs

settings_intl/
intl_fidl_handler.rs
