input_pipeline/
input_pipeline.rs

1// Copyright 2020 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::display_ownership::DisplayOwnership;
6use crate::focus_listener::FocusListener;
7use crate::input_device::{InputEventType, InputPipelineFeatureFlags};
8use crate::input_handler::Handler;
9use crate::{Dispatcher, Incoming, Transport, dispatcher, input_device, input_handler, metrics};
10use anyhow::{Context, Error, format_err};
11use fidl::endpoints;
12use fidl_fuchsia_io as fio;
13use focus_chain_provider::FocusChainProviderPublisher;
14use fuchsia_async as fasync;
15use fuchsia_component::directory::AsRefDirectory;
16use fuchsia_fs::directory::{WatchEvent, Watcher};
17use fuchsia_inspect::NumericProperty;
18use fuchsia_inspect::health::Reporter;
19use fuchsia_sync::Mutex;
20use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
21use futures::future::LocalBoxFuture;
22use futures::{StreamExt, TryStreamExt};
23use itertools::Itertools;
24use metrics_registry::*;
25use sorted_vec_map::SortedVecMap;
26use std::path::PathBuf;
27use std::rc::Rc;
28use std::sync::atomic::{AtomicU32, Ordering};
29use std::sync::{Arc, LazyLock};
30use strum::EnumCount;
31
32/// Use a self incremental u32 unique id for device_id.
33///
34/// device id start from 10 to avoid conflict with default devices in Starnix.
35/// Currently, Starnix using 0 and 1 as default devices' id. Starnix need to
36/// use default devices to deliver events from physical devices until we have
37/// API to expose device changes to UI clients.
38static NEXT_DEVICE_ID: LazyLock<AtomicU32> = LazyLock::new(|| AtomicU32::new(10));
39
40/// Each time this function is invoked, it returns the current value of its
41/// internal counter (serving as a unique id for device_id) and then increments
42/// that counter in preparation for the next call.
43fn get_next_device_id() -> u32 {
44    NEXT_DEVICE_ID.fetch_add(1, Ordering::SeqCst)
45}
46
47type BoxedInputDeviceBinding = Box<dyn input_device::InputDeviceBinding>;
48
49/// An [`InputDeviceBindingMap`] maps an input device to one or more InputDeviceBindings.
50/// It uses unique device id as key.
51pub type InputDeviceBindingMap = Arc<Mutex<SortedVecMap<u32, Vec<BoxedInputDeviceBinding>>>>;
52
53/// An input pipeline assembly.
54///
55/// Represents a partial stage of the input pipeline which accepts inputs through an asynchronous
56/// sender channel, and emits outputs through an asynchronous receiver channel.  Use [new] to
57/// create a new assembly.  Use [add_handler], or [add_all_handlers] to add the input pipeline
58/// handlers to use.  When done, [InputPipeline::new] can be used to make a new input pipeline.
59///
60/// # Implementation notes
61///
62/// Internally, when a new [InputPipelineAssembly] is created with multiple [InputHandler]s, the
63/// handlers are connected together using async queues.  This allows fully streamed processing of
64/// input events, and also allows some pipeline stages to generate events spontaneously, i.e.
65/// without an external stimulus.
66pub struct InputPipelineAssembly {
67    /// The top-level sender: send into this queue to inject an event into the input
68    /// pipeline.
69    sender: UnboundedSender<Vec<input_device::InputEvent>>,
70    /// The bottom-level receiver: any events that fall through the entire pipeline can
71    /// be read from this receiver.
72    receiver: UnboundedReceiver<Vec<input_device::InputEvent>>,
73
74    /// The input handlers that comprise the input pipeline.
75    handlers: Vec<Rc<dyn input_handler::BatchInputHandler>>,
76
77    /// The display ownership watcher task.
78    display_ownership_fut: Option<LocalBoxFuture<'static, ()>>,
79
80    /// The focus listener task.
81    focus_listener_fut: Option<LocalBoxFuture<'static, ()>>,
82
83    /// The metrics logger.
84    metrics_logger: metrics::MetricsLogger,
85}
86
87impl InputPipelineAssembly {
88    /// Create a new but empty [InputPipelineAssembly]. Use [add_handler] or similar
89    /// to add new handlers to it.
90    pub fn new(metrics_logger: metrics::MetricsLogger) -> Self {
91        let (sender, receiver) = mpsc::unbounded();
92        InputPipelineAssembly {
93            sender,
94            receiver,
95            handlers: vec![],
96            metrics_logger,
97            display_ownership_fut: None,
98            focus_listener_fut: None,
99        }
100    }
101
102    /// Adds another [input_handler::BatchInputHandler] into the [InputPipelineAssembly]. The handlers
103    /// are invoked in the order they are added. Returns `Self` for chaining.
104    pub fn add_handler(mut self, handler: Rc<dyn input_handler::BatchInputHandler>) -> Self {
105        self.handlers.push(handler);
106        self
107    }
108
109    /// Adds all handlers into the assembly in the order they appear in `handlers`.
110    pub fn add_all_handlers(self, handlers: Vec<Rc<dyn input_handler::BatchInputHandler>>) -> Self {
111        handlers.into_iter().fold(self, |assembly, handler| assembly.add_handler(handler))
112    }
113
114    pub fn add_display_ownership(
115        mut self,
116        display_ownership_event: zx::Event,
117        input_handlers_node: &fuchsia_inspect::Node,
118    ) -> InputPipelineAssembly {
119        let h = DisplayOwnership::new(
120            display_ownership_event,
121            input_handlers_node,
122            self.metrics_logger.clone(),
123        );
124        let metrics_logger_clone = self.metrics_logger.clone();
125        let h_clone = h.clone();
126        let sender_clone = self.sender.clone();
127        let display_ownership_fut = Box::pin(async move {
128            h_clone.clone().set_handler_healthy();
129            h_clone.clone()
130                .handle_ownership_change(sender_clone)
131                .await
132                .map_err(|e| {
133                    metrics_logger_clone.log_error(
134                        InputPipelineErrorMetricDimensionEvent::InputPipelineDisplayOwnershipIsNotSupposedToTerminate,
135                        std::format!(
136                            "display ownership is not supposed to terminate - this is likely a problem: {:?}", e));
137                        })
138                        .unwrap();
139            h_clone.set_handler_unhealthy("Receive loop terminated for handler: DisplayOwnership");
140        });
141        self.display_ownership_fut = Some(display_ownership_fut);
142        self.add_handler(h)
143    }
144
145    /// Deconstructs the assembly into constituent components, used when constructing
146    /// [InputPipeline].
147    ///
148    /// You should call [catch_unhandled] on the returned [async_channel::Receiver], and
149    /// [run] on the returned [fuchsia_async::Tasks] (or supply own equivalents).
150    fn into_components(
151        self,
152    ) -> (
153        UnboundedSender<Vec<input_device::InputEvent>>,
154        UnboundedReceiver<Vec<input_device::InputEvent>>,
155        Vec<Rc<dyn input_handler::BatchInputHandler>>,
156        metrics::MetricsLogger,
157        Option<LocalBoxFuture<'static, ()>>,
158        Option<LocalBoxFuture<'static, ()>>,
159    ) {
160        (
161            self.sender,
162            self.receiver,
163            self.handlers,
164            self.metrics_logger,
165            self.display_ownership_fut,
166            self.focus_listener_fut,
167        )
168    }
169
170    pub fn add_focus_listener(
171        mut self,
172        incoming: &Incoming,
173        focus_chain_publisher: FocusChainProviderPublisher,
174    ) -> Self {
175        let metrics_logger_clone = self.metrics_logger.clone();
176        let incoming2 = incoming.clone();
177        let focus_listener_fut = Box::pin(async move {
178            if let Ok(mut focus_listener) = FocusListener::new(
179                &incoming2,
180                focus_chain_publisher,
181                metrics_logger_clone,
182            )
183            .map_err(|e| {
184                log::warn!("could not create focus listener, focus will not be dispatched: {:?}", e)
185            }) {
186                // This will await indefinitely and process focus messages in a loop, unless there
187                // is a problem.
188                let _result = focus_listener
189                    .dispatch_focus_changes()
190                    .await
191                    .map(|_| {
192                        log::warn!("dispatch focus loop ended, focus will no longer be dispatched")
193                    })
194                    .map_err(|e| {
195                        panic!("could not dispatch focus changes, this is a fatal error: {:?}", e)
196                    });
197            }
198        });
199        self.focus_listener_fut = Some(focus_listener_fut);
200        self
201    }
202}
203
204/// An [`InputPipeline`] manages input devices and propagates input events through input handlers.
205///
206/// On creation, clients declare what types of input devices an [`InputPipeline`] manages. The
207/// [`InputPipeline`] will continuously detect new input devices of supported type(s).
208///
209/// # Example
210/// ```
211/// let ime_handler =
212///     ImeHandler::new(scene_manager.session.clone(), scene_manager.compositor_id).await?;
213/// let touch_handler = TouchHandler::new(
214///     scene_manager.session.clone(),
215///     scene_manager.compositor_id,
216///     scene_manager.display_size
217/// ).await?;
218///
219/// let assembly = InputPipelineAssembly::new()
220///     .add_handler(Box::new(ime_handler)),
221///     .add_handler(Box::new(touch_handler)),
222/// let input_pipeline = InputPipeline::new(
223///     vec![
224///         input_device::InputDeviceType::Touch,
225///         input_device::InputDeviceType::Keyboard,
226///     ],
227///     assembly,
228/// );
229/// input_pipeline.handle_input_events().await;
230/// ```
231pub struct InputPipeline {
232    /// The entry point into the input handler pipeline. Incoming input events should
233    /// be inserted into this async queue, and the input pipeline will ensure that they
234    /// are propagated through all the input handlers in the appropriate sequence.
235    pipeline_sender: UnboundedSender<Vec<input_device::InputEvent>>,
236
237    /// A clone of this sender is given to every InputDeviceBinding that this pipeline owns.
238    /// Each InputDeviceBinding will send InputEvents to the pipeline through this channel.
239    device_event_sender: UnboundedSender<Vec<input_device::InputEvent>>,
240
241    /// Receives InputEvents from all InputDeviceBindings that this pipeline owns.
242    device_event_receiver: UnboundedReceiver<Vec<input_device::InputEvent>>,
243
244    /// The types of devices this pipeline supports.
245    input_device_types: Vec<input_device::InputDeviceType>,
246
247    /// The InputDeviceBindings bound to this pipeline.
248    input_device_bindings: InputDeviceBindingMap,
249
250    /// This node is bound to the lifetime of this InputPipeline.
251    /// Inspect data will be dumped for this pipeline as long as it exists.
252    inspect_node: fuchsia_inspect::Node,
253
254    /// The metrics logger.
255    metrics_logger: metrics::MetricsLogger,
256
257    /// The feature flags for the input pipeline.
258    pub feature_flags: input_device::InputPipelineFeatureFlags,
259
260    /// Tasks running in the background on the Dispatcher.
261    _tasks: Vec<dispatcher::TaskHandle<()>>,
262
263    /// Tasks running in the background on fuchsia_async dispatcher.
264    _fasync_tasks: Vec<fasync::Task<()>>,
265}
266
267impl InputPipeline {
268    fn new_common(
269        input_device_types: Vec<input_device::InputDeviceType>,
270        assembly: InputPipelineAssembly,
271        inspect_node: fuchsia_inspect::Node,
272        feature_flags: input_device::InputPipelineFeatureFlags,
273    ) -> Self {
274        let (
275            pipeline_sender,
276            receiver,
277            handlers,
278            metrics_logger,
279            display_ownership_fut,
280            focus_listener_fut,
281        ) = assembly.into_components();
282
283        let mut tasks = vec![];
284        let mut fasync_tasks = vec![];
285
286        let mut handlers_count = handlers.len();
287        // TODO: b/469745447 - should use futures::select! instead of spawning tasks.
288        if let Some(fut) = display_ownership_fut {
289            // The displayer ownership handler, like all input handlers, runs on [`crate::Dispatcher`]
290            // which is driver dispatcher in dso mode. The display ownership future must run on
291            // the same dispatcher because the types do not support multithreaded access.
292            tasks.push(Dispatcher::spawn_local(fut));
293            handlers_count += 1;
294        }
295
296        // TODO: b/469745447 - should use futures::select! instead of spawning tasks.
297        if let Some(fut) = focus_listener_fut {
298            fasync_tasks.push(fasync::Task::local(fut));
299            handlers_count += 1;
300        }
301
302        // Add properties to inspect node
303        inspect_node.record_string("supported_input_devices", input_device_types.iter().join(", "));
304        inspect_node.record_uint("handlers_registered", handlers_count as u64);
305        inspect_node.record_uint("handlers_healthy", handlers_count as u64);
306
307        // Initializes all handlers and starts the input pipeline loop.
308        let runner_task = InputPipeline::run(receiver, handlers, metrics_logger.clone());
309        tasks.push(runner_task);
310
311        let (device_event_sender, device_event_receiver) = futures::channel::mpsc::unbounded();
312        let input_device_bindings: InputDeviceBindingMap =
313            Arc::new(Mutex::new(SortedVecMap::new()));
314        InputPipeline {
315            pipeline_sender,
316            device_event_sender,
317            device_event_receiver,
318            input_device_types,
319            input_device_bindings,
320            inspect_node,
321            metrics_logger,
322            feature_flags,
323            _tasks: tasks,
324            _fasync_tasks: fasync_tasks,
325        }
326    }
327
328    /// Creates a new [`InputPipeline`] for integration testing.
329    /// Unlike a production input pipeline, this pipeline will not monitor
330    /// `/dev/class/input-report` for devices.
331    ///
332    /// # Parameters
333    /// - `input_device_types`: The types of devices the new [`InputPipeline`] will support.
334    /// - `assembly`: The input handlers that the [`InputPipeline`] sends InputEvents to.
335    pub fn new_for_test(
336        input_device_types: Vec<input_device::InputDeviceType>,
337        assembly: InputPipelineAssembly,
338    ) -> Self {
339        let inspector = fuchsia_inspect::Inspector::default();
340        let root = inspector.root();
341        let test_node = root.create_child("input_pipeline");
342        Self::new_common(
343            input_device_types,
344            assembly,
345            test_node,
346            input_device::InputPipelineFeatureFlags { enable_merge_touch_events: false },
347        )
348    }
349
350    /// Creates a new [`InputPipeline`] for production use.
351    ///
352    /// # Parameters
353    /// - `input_device_types`: The types of devices the new [`InputPipeline`] will support.
354    /// - `assembly`: The input handlers that the [`InputPipeline`] sends InputEvents to.
355    /// - `inspect_node`: The root node for InputPipeline's Inspect tree
356    pub fn new(
357        incoming: &Incoming,
358        input_device_types: Vec<input_device::InputDeviceType>,
359        assembly: InputPipelineAssembly,
360        inspect_node: fuchsia_inspect::Node,
361        feature_flags: input_device::InputPipelineFeatureFlags,
362        metrics_logger: metrics::MetricsLogger,
363    ) -> Result<Self, Error> {
364        let mut input_pipeline =
365            Self::new_common(input_device_types, assembly, inspect_node, feature_flags);
366        let input_device_types = input_pipeline.input_device_types.clone();
367        let input_event_sender = input_pipeline.device_event_sender.clone();
368        let input_device_bindings = input_pipeline.input_device_bindings.clone();
369        let devices_node = input_pipeline.inspect_node.create_child("input_devices");
370        let feature_flags = input_pipeline.feature_flags.clone();
371        let incoming = incoming.clone();
372        // This intentionally uses the [`fuchsia_async`] task dispatcher instead of
373        // [`crate::Dispatcher`] -- the directory watcher always uses the fuchsia-async dispatcher.
374        // This is fine for performance because the actual event dispatch is still configured to
375        // run on [`crate::Dispatcher`].
376        let watcher_task = fasync::Task::local(async move {
377            // Watches the input device directory for new input devices. Creates new InputDeviceBindings
378            // that send InputEvents to `input_event_receiver`.
379            match async {
380                let (dir_proxy, server) = endpoints::create_proxy::<fio::DirectoryMarker>();
381                incoming
382                    .as_ref_directory()
383                    .open(input_device::INPUT_REPORT_PATH, fio::PERM_READABLE, server.into())
384                    .with_context(|| {
385                        format!("failed to open {}", input_device::INPUT_REPORT_PATH)
386                    })?;
387                let device_watcher =
388                    Watcher::new(&dir_proxy).await.context("failed to create watcher")?;
389                Self::watch_for_devices(
390                    device_watcher,
391                    dir_proxy,
392                    input_device_types,
393                    input_event_sender,
394                    input_device_bindings,
395                    &devices_node,
396                    false, /* break_on_idle */
397                    feature_flags,
398                    metrics_logger.clone(),
399                )
400                .await
401                .context("failed to watch for devices")
402            }
403            .await
404            {
405                Ok(()) => {}
406                Err(err) => {
407                    // This error is usually benign in tests: it means that the setup does not
408                    // support dynamic device discovery. Almost no tests support dynamic
409                    // device discovery, and they also do not need those.
410                    metrics_logger.log_warn(
411                        InputPipelineErrorMetricDimensionEvent::InputPipelineUnableToWatchForNewInputDevices,
412                        std::format!(
413                            "Input pipeline is unable to watch for new input devices: {:?}",
414                            err
415                        ));
416                }
417            }
418        });
419        input_pipeline._fasync_tasks.push(watcher_task);
420
421        Ok(input_pipeline)
422    }
423
424    /// Gets the input device bindings.
425    pub fn input_device_bindings(&self) -> &InputDeviceBindingMap {
426        &self.input_device_bindings
427    }
428
429    /// Gets the input device sender: this is the channel that should be cloned
430    /// and used for injecting events from the drivers into the input pipeline.
431    pub fn input_event_sender(&self) -> &UnboundedSender<Vec<input_device::InputEvent>> {
432        &self.device_event_sender
433    }
434
435    /// Gets a list of input device types supported by this input pipeline.
436    pub fn input_device_types(&self) -> &Vec<input_device::InputDeviceType> {
437        &self.input_device_types
438    }
439
440    /// Forwards all input events into the input pipeline.
441    pub async fn handle_input_events(mut self) {
442        let metrics_logger_clone = self.metrics_logger.clone();
443        while let Some(input_event) = self.device_event_receiver.next().await {
444            if let Err(e) = self.pipeline_sender.unbounded_send(input_event) {
445                metrics_logger_clone.log_error(
446                    InputPipelineErrorMetricDimensionEvent::InputPipelineCouldNotForwardEventFromDriver,
447                    std::format!("could not forward event from driver: {:?}", e));
448            }
449        }
450
451        metrics_logger_clone.log_error(
452            InputPipelineErrorMetricDimensionEvent::InputPipelineStopHandlingEvents,
453            "Input pipeline stopped handling input events.".to_string(),
454        );
455    }
456
457    /// Watches the input report directory for new input devices. Creates InputDeviceBindings
458    /// if new devices match a type in `device_types`.
459    ///
460    /// # Parameters
461    /// - `device_watcher`: Watches the input report directory for new devices.
462    /// - `dir_proxy`: The directory containing InputDevice connections.
463    /// - `device_types`: The types of devices to watch for.
464    /// - `input_event_sender`: The channel new InputDeviceBindings will send InputEvents to.
465    /// - `bindings`: Holds all the InputDeviceBindings
466    /// - `input_devices_node`: The parent node for all device bindings' inspect nodes.
467    /// - `break_on_idle`: If true, stops watching for devices once all existing devices are handled.
468    /// - `metrics_logger`: The metrics logger.
469    ///
470    /// # Errors
471    /// If the input report directory or a file within it cannot be read.
472    async fn watch_for_devices(
473        mut device_watcher: Watcher,
474        dir_proxy: fio::DirectoryProxy,
475        device_types: Vec<input_device::InputDeviceType>,
476        input_event_sender: UnboundedSender<Vec<input_device::InputEvent>>,
477        bindings: InputDeviceBindingMap,
478        input_devices_node: &fuchsia_inspect::Node,
479        break_on_idle: bool,
480        feature_flags: input_device::InputPipelineFeatureFlags,
481        metrics_logger: metrics::MetricsLogger,
482    ) -> Result<(), Error> {
483        // Add non-static properties to inspect node.
484        let devices_discovered = input_devices_node.create_uint("devices_discovered", 0);
485        let devices_connected = input_devices_node.create_uint("devices_connected", 0);
486        while let Some(msg) = device_watcher.try_next().await? {
487            if let Ok(filename) = msg.filename.into_os_string().into_string() {
488                if filename == "." {
489                    continue;
490                }
491
492                let pathbuf = PathBuf::from(filename.clone());
493                match msg.event {
494                    WatchEvent::EXISTING | WatchEvent::ADD_FILE => {
495                        log::info!("found input device {}", filename);
496                        devices_discovered.add(1);
497                        let device_proxy =
498                            input_device::get_device_from_dir_entry_path(&dir_proxy, &pathbuf)?;
499                        add_device_bindings(
500                            &device_types,
501                            &filename,
502                            device_proxy,
503                            &input_event_sender,
504                            &bindings,
505                            get_next_device_id(),
506                            input_devices_node,
507                            Some(&devices_connected),
508                            feature_flags.clone(),
509                            metrics_logger.clone(),
510                        )
511                        .await;
512                    }
513                    WatchEvent::IDLE => {
514                        if break_on_idle {
515                            break;
516                        }
517                    }
518                    _ => (),
519                }
520            }
521        }
522        // Ensure inspect properties persist for debugging if device watch loop ends.
523        input_devices_node.record(devices_discovered);
524        input_devices_node.record(devices_connected);
525        Err(format_err!("Input pipeline stopped watching for new input devices."))
526    }
527
528    /// Handles the incoming InputDeviceRegistryRequestStream.
529    ///
530    /// This method will end when the request stream is closed. If the stream closes with an
531    /// error the error will be returned in the Result.
532    ///
533    /// **NOTE**: Only one stream is handled at a time. https://fxbug.dev/42061078
534    ///
535    /// # Parameters
536    /// - `stream`: The stream of InputDeviceRegistryRequests.
537    /// - `device_types`: The types of devices to watch for.
538    /// - `input_event_sender`: The channel new InputDeviceBindings will send InputEvents to.
539    /// - `bindings`: Holds all the InputDeviceBindings associated with the InputPipeline.
540    /// - `input_devices_node`: The parent node for all injected devices' inspect nodes.
541    /// - `metrics_logger`: The metrics logger.
542    pub async fn handle_input_device_registry_request_stream(
543        mut stream: fidl_fuchsia_input_injection::InputDeviceRegistryRequestStream,
544        device_types: &Vec<input_device::InputDeviceType>,
545        input_event_sender: &UnboundedSender<Vec<input_device::InputEvent>>,
546        bindings: &InputDeviceBindingMap,
547        input_devices_node: &fuchsia_inspect::Node,
548        feature_flags: input_device::InputPipelineFeatureFlags,
549        metrics_logger: metrics::MetricsLogger,
550    ) -> Result<(), Error> {
551        while let Some(request) = stream
552            .try_next()
553            .await
554            .context("Error handling input device registry request stream")?
555        {
556            match request {
557                fidl_fuchsia_input_injection::InputDeviceRegistryRequest::Register {
558                    device,
559                    ..
560                } => {
561                    // Add a binding if the device is a type being tracked
562                    let device = fidl_next::ClientEnd::<
563                        fidl_next_fuchsia_input_report::InputDevice,
564                        zx::Channel,
565                    >::from_untyped(device.into_channel());
566                    let device = Dispatcher::client_from_zx_channel(device);
567                    let device = device.spawn();
568                    let device_id = get_next_device_id();
569
570                    add_device_bindings(
571                        device_types,
572                        &format!("input-device-registry-{}", device_id),
573                        device,
574                        input_event_sender,
575                        bindings,
576                        device_id,
577                        input_devices_node,
578                        None,
579                        feature_flags.clone(),
580                        metrics_logger.clone(),
581                    )
582                    .await;
583                }
584                fidl_fuchsia_input_injection::InputDeviceRegistryRequest::RegisterAndGetDeviceInfo {
585                    device,
586                    responder,
587                    .. } => {
588                    // Add a binding if the device is a type being tracked
589                    let device = fidl_next::ClientEnd::<
590                        fidl_next_fuchsia_input_report::InputDevice,
591                        zx::Channel,
592                    >::from_untyped(device.into_channel());
593                    let device = Dispatcher::client_from_zx_channel(device);
594                    let device = device.spawn();
595                    let device_id = get_next_device_id();
596
597                    add_device_bindings(
598                        device_types,
599                        &format!("input-device-registry-{}", device_id),
600                        device,
601                        input_event_sender,
602                        bindings,
603                        device_id,
604                        input_devices_node,
605                        None,
606                        feature_flags.clone(),
607                        metrics_logger.clone(),
608                    )
609                    .await;
610
611                    responder.send(fidl_fuchsia_input_injection::InputDeviceRegistryRegisterAndGetDeviceInfoResponse{
612                        device_id: Some(device_id),
613                        ..Default::default()
614                    }).expect("Failed to respond to RegisterAndGetDeviceInfo request");
615                }
616            }
617        }
618
619        Ok(())
620    }
621
622    /// Initializes all handlers and starts the input pipeline loop in an asynchronous executor.
623    fn run(
624        mut receiver: UnboundedReceiver<Vec<input_device::InputEvent>>,
625        handlers: Vec<Rc<dyn input_handler::BatchInputHandler>>,
626        metrics_logger: metrics::MetricsLogger,
627    ) -> dispatcher::TaskHandle<()> {
628        Dispatcher::spawn_local(async move {
629            for handler in &handlers {
630                handler.clone().set_handler_healthy();
631            }
632
633            let mut handlers_by_type: [Vec<Rc<dyn input_handler::BatchInputHandler>>;
634                InputEventType::COUNT] = Default::default();
635
636            // TODO: b/478262850 - We can use supported_input_devices to populate this list.
637            let event_types = vec![
638                InputEventType::Keyboard,
639                InputEventType::LightSensor,
640                InputEventType::ConsumerControls,
641                InputEventType::Mouse,
642                InputEventType::TouchScreen,
643                InputEventType::Touchpad,
644                #[cfg(test)]
645                InputEventType::Fake,
646            ];
647
648            for event_type in event_types {
649                let handlers_for_type: Vec<Rc<dyn input_handler::BatchInputHandler>> = handlers
650                    .iter()
651                    .filter(|h| h.interest().contains(&event_type))
652                    .cloned()
653                    .collect();
654                handlers_by_type[event_type as usize] = handlers_for_type;
655            }
656
657            while let Some(events) = receiver.next().await {
658                if events.is_empty() {
659                    continue;
660                }
661
662                let mut groups_seen = 0;
663                let events = events.into_iter().chunk_by(|e| InputEventType::from(&e.device_event));
664                let events = events.into_iter().map(|(k, v)| (k, v.collect::<Vec<_>>()));
665                for (event_type, event_group) in events {
666                    groups_seen += 1;
667                    if groups_seen == 2 {
668                        metrics_logger.log_error(
669                                InputPipelineErrorMetricDimensionEvent::InputFrameContainsMultipleTypesOfEvents,
670                                "it is not recommended to contain multiple types of events in 1 send".to_string(),
671                            );
672                    }
673                    let mut events_in_group = event_group;
674
675                    // Get pre-computed handlers for this event type.
676                    let handlers = &handlers_by_type[event_type as usize];
677
678                    for handler in handlers {
679                        events_in_group =
680                            handler.clone().handle_input_events(events_in_group).await;
681                    }
682
683                    for event in events_in_group {
684                        if event.handled == input_device::Handled::No {
685                            log::warn!("unhandled input event: {:?}", event);
686                        }
687                        if let Some(trace_id) = event.trace_id {
688                            fuchsia_trace::flow_end!(
689                                "input",
690                                "event_in_input_pipeline",
691                                trace_id.into()
692                            );
693                        }
694                    }
695                }
696            }
697            for handler in &handlers {
698                handler.clone().set_handler_unhealthy("Pipeline loop terminated");
699            }
700            panic!("Runner task is not supposed to terminate.")
701        })
702    }
703}
704
705/// Adds `InputDeviceBinding`s to `bindings` for all `device_types` exposed by `device_proxy`.
706///
707/// # Parameters
708/// - `device_types`: The types of devices to watch for.
709/// - `device_proxy`: A proxy to the input device.
710/// - `input_event_sender`: The channel new InputDeviceBindings will send InputEvents to.
711/// - `bindings`: Holds all the InputDeviceBindings associated with the InputPipeline.
712/// - `device_id`: The device id of the associated bindings.
713/// - `input_devices_node`: The parent node for all device bindings' inspect nodes.
714///
715/// # Note
716/// This will create multiple bindings, in the case where
717/// * `device_proxy().get_descriptor()` returns a `fidl_fuchsia_input_report::DeviceDescriptor`
718///   with multiple table fields populated, and
719/// * multiple populated table fields correspond to device types present in `device_types`
720///
721/// This is used, for example, to support the Atlas touchpad. In that case, a single
722/// node in `/dev/class/input-report` provides both a `fuchsia.input.report.MouseDescriptor` and
723/// a `fuchsia.input.report.TouchDescriptor`.
724async fn add_device_bindings(
725    device_types: &Vec<input_device::InputDeviceType>,
726    filename: &String,
727    device_proxy: fidl_next::Client<fidl_next_fuchsia_input_report::InputDevice, Transport>,
728    input_event_sender: &UnboundedSender<Vec<input_device::InputEvent>>,
729    bindings: &InputDeviceBindingMap,
730    device_id: u32,
731    input_devices_node: &fuchsia_inspect::Node,
732    devices_connected: Option<&fuchsia_inspect::UintProperty>,
733    feature_flags: InputPipelineFeatureFlags,
734    metrics_logger: metrics::MetricsLogger,
735) {
736    let mut matched_device_types = vec![];
737    if let Ok(res) = device_proxy.get_descriptor().await {
738        for device_type in device_types {
739            if input_device::is_device_type(&res.descriptor, *device_type).await {
740                matched_device_types.push(device_type);
741                match devices_connected {
742                    Some(dev_connected) => {
743                        let _ = dev_connected.add(1);
744                    }
745                    None => (),
746                };
747            }
748        }
749        if matched_device_types.is_empty() {
750            log::info!(
751                "device {} did not match any supported device types: {:?}",
752                filename,
753                device_types
754            );
755            let device_node = input_devices_node.create_child(format!("{}_Unsupported", filename));
756            let mut health = fuchsia_inspect::health::Node::new(&device_node);
757            health.set_unhealthy("Unsupported device type.");
758            device_node.record(health);
759            input_devices_node.record(device_node);
760            return;
761        }
762    } else {
763        metrics_logger.clone().log_error(
764            InputPipelineErrorMetricDimensionEvent::InputPipelineNoDeviceDescriptor,
765            std::format!("cannot bind device {} without a device descriptor", filename),
766        );
767        return;
768    }
769
770    log::info!(
771        "binding {} to device types: {}",
772        filename,
773        matched_device_types
774            .iter()
775            .fold(String::new(), |device_types_string, device_type| device_types_string
776                + &format!("{:?}, ", device_type))
777    );
778
779    let mut new_bindings: Vec<BoxedInputDeviceBinding> = vec![];
780    for device_type in matched_device_types {
781        // Clone `device_proxy`, so that multiple bindings (e.g. a `MouseBinding` and a
782        // `TouchBinding`) can read data from the same `/dev/class/input-report` node.
783        //
784        // There's no conflict in having multiple bindings read from the same node,
785        // since:
786        // * each binding will create its own `fuchsia.input.report.InputReportsReader`, and
787        // * the device driver will copy each incoming report to each connected reader.
788        //
789        // This does mean that reports from the Atlas touchpad device get read twice
790        // (by a `MouseBinding` and a `TouchBinding`), regardless of whether the device
791        // is operating in mouse mode or touchpad mode.
792        //
793        // This hasn't been an issue because:
794        // * Semantically: things are fine, because each binding discards irrelevant reports.
795        //   (E.g. `MouseBinding` discards anything that isn't a `MouseInputReport`), and
796        // * Performance wise: things are fine, because the data rate of the touchpad is low
797        //   (125 HZ).
798        //
799        // If we add additional cases where bindings share an underlying `input-report` node,
800        // we might consider adding a multiplexing binding, to avoid reading duplicate reports.
801        let proxy = device_proxy.clone();
802        let device_node = input_devices_node.create_child(format!("{}_{}", filename, device_type));
803        match input_device::get_device_binding(
804            *device_type,
805            proxy,
806            device_id,
807            input_event_sender.clone(),
808            device_node,
809            feature_flags.clone(),
810            metrics_logger.clone(),
811        )
812        .await
813        {
814            Ok(binding) => new_bindings.push(binding),
815            Err(e) => {
816                metrics_logger.log_error(
817                    InputPipelineErrorMetricDimensionEvent::InputPipelineFailedToBind,
818                    std::format!("failed to bind {} as {:?}: {}", filename, device_type, e),
819                );
820            }
821        }
822    }
823
824    if !new_bindings.is_empty() {
825        let mut bindings = bindings.lock();
826        if let Some(v) = bindings.get_mut(&device_id) {
827            v.extend(new_bindings);
828        } else {
829            bindings.insert(device_id, new_bindings);
830        }
831    }
832}
833
834#[cfg(test)]
835mod tests {
836    use super::*;
837    use crate::input_device::{InputDeviceBinding, InputEventType};
838    use crate::utils::Position;
839    use crate::{fake_input_device_binding, mouse_binding, observe_fake_events_input_handler};
840    use async_trait::async_trait;
841    use diagnostics_assertions::AnyProperty;
842    use fidl::endpoints::{create_proxy_and_stream, create_request_stream};
843    use fuchsia_async as fasync;
844    use futures::FutureExt;
845    use pretty_assertions::assert_eq;
846    use rand::Rng;
847    use sorted_vec_map::SortedVecSet;
848    use vfs::{pseudo_directory, service as pseudo_fs_service};
849
850    /// Returns the InputEvent sent over `sender`.
851    ///
852    /// # Parameters
853    /// - `sender`: The channel to send the InputEvent over.
854    fn send_input_event(
855        sender: UnboundedSender<Vec<input_device::InputEvent>>,
856    ) -> Vec<input_device::InputEvent> {
857        let mut rng = rand::rng();
858        let offset =
859            Position { x: rng.random_range(0..10) as f32, y: rng.random_range(0..10) as f32 };
860        let input_event = input_device::InputEvent {
861            device_event: input_device::InputDeviceEvent::Mouse(mouse_binding::MouseEvent::new(
862                mouse_binding::MouseLocation::Relative(mouse_binding::RelativeLocation {
863                    counts: Position { x: offset.x, y: offset.y },
864                }),
865                None, /* wheel_delta_v */
866                None, /* wheel_delta_h */
867                mouse_binding::MousePhase::Move,
868                SortedVecSet::new(),
869                SortedVecSet::new(),
870                None, /* is_precision_scroll */
871                None, /* wake_lease */
872            )),
873            device_descriptor: input_device::InputDeviceDescriptor::Mouse(
874                mouse_binding::MouseDeviceDescriptor {
875                    device_id: 1,
876                    absolute_x_range: None,
877                    absolute_y_range: None,
878                    wheel_v_range: None,
879                    wheel_h_range: None,
880                    buttons: None,
881                },
882            ),
883            event_time: zx::MonotonicInstant::get(),
884            handled: input_device::Handled::No,
885            trace_id: None,
886        };
887        match sender.unbounded_send(vec![input_event.clone()]) {
888            Err(_) => assert!(false),
889            _ => {}
890        }
891
892        vec![input_event]
893    }
894
895    /// Returns a MouseDescriptor on an InputDeviceRequest.
896    ///
897    /// # Parameters
898    /// - `input_device_request`: The request to handle.
899    fn handle_input_device_request(
900        input_device_request: fidl_fuchsia_input_report::InputDeviceRequest,
901    ) {
902        match input_device_request {
903            fidl_fuchsia_input_report::InputDeviceRequest::GetDescriptor { responder } => {
904                let _ = responder.send(&fidl_fuchsia_input_report::DeviceDescriptor {
905                    device_information: None,
906                    mouse: Some(fidl_fuchsia_input_report::MouseDescriptor {
907                        input: Some(fidl_fuchsia_input_report::MouseInputDescriptor {
908                            movement_x: None,
909                            movement_y: None,
910                            scroll_v: None,
911                            scroll_h: None,
912                            buttons: Some(vec![0]),
913                            position_x: None,
914                            position_y: None,
915                            ..Default::default()
916                        }),
917                        ..Default::default()
918                    }),
919                    sensor: None,
920                    touch: None,
921                    keyboard: None,
922                    consumer_control: None,
923                    ..Default::default()
924                });
925            }
926            _ => {}
927        }
928    }
929
930    /// Tests that an input pipeline handles events from multiple devices.
931    #[fasync::run_singlethreaded(test)]
932    async fn multiple_devices_single_handler() {
933        // Create two fake device bindings.
934        let (device_event_sender, device_event_receiver) = futures::channel::mpsc::unbounded();
935        let first_device_binding =
936            fake_input_device_binding::FakeInputDeviceBinding::new(device_event_sender.clone());
937        let second_device_binding =
938            fake_input_device_binding::FakeInputDeviceBinding::new(device_event_sender.clone());
939
940        // Create a fake input handler.
941        let (handler_event_sender, mut handler_event_receiver) =
942            futures::channel::mpsc::channel(100);
943        let input_handler = observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
944            handler_event_sender,
945        );
946
947        // Build the input pipeline.
948        let (sender, receiver, handlers, _, _, _) =
949            InputPipelineAssembly::new(metrics::MetricsLogger::default())
950                .add_handler(input_handler)
951                .into_components();
952        let inspector = fuchsia_inspect::Inspector::default();
953        let test_node = inspector.root().create_child("input_pipeline");
954        let input_pipeline = InputPipeline {
955            pipeline_sender: sender,
956            device_event_sender,
957            device_event_receiver,
958            input_device_types: vec![],
959            input_device_bindings: Arc::new(Mutex::new(SortedVecMap::new())),
960            inspect_node: test_node,
961            metrics_logger: metrics::MetricsLogger::default(),
962            feature_flags: input_device::InputPipelineFeatureFlags::default(),
963            _tasks: vec![],
964            _fasync_tasks: vec![],
965        };
966        let _runner_task =
967            InputPipeline::run(receiver, handlers, metrics::MetricsLogger::default());
968
969        // Send an input event from each device.
970        let first_device_events = send_input_event(first_device_binding.input_event_sender());
971        let second_device_events = send_input_event(second_device_binding.input_event_sender());
972
973        // Run the pipeline.
974        let _pipeline_task = fasync::Task::local(async {
975            input_pipeline.handle_input_events().await;
976        });
977
978        // Assert the handler receives the events.
979        let first_handled_event = handler_event_receiver.next().await;
980        assert_eq!(first_handled_event, first_device_events.into_iter().next());
981
982        let second_handled_event = handler_event_receiver.next().await;
983        assert_eq!(second_handled_event, second_device_events.into_iter().next());
984    }
985
986    /// Tests that an input pipeline handles events through multiple input handlers.
987    #[fasync::run_singlethreaded(test)]
988    async fn single_device_multiple_handlers() {
989        // Create two fake device bindings.
990        let (device_event_sender, device_event_receiver) = futures::channel::mpsc::unbounded();
991        let input_device_binding =
992            fake_input_device_binding::FakeInputDeviceBinding::new(device_event_sender.clone());
993
994        // Create two fake input handlers.
995        let (first_handler_event_sender, mut first_handler_event_receiver) =
996            futures::channel::mpsc::channel(100);
997        let first_input_handler =
998            observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
999                first_handler_event_sender,
1000            );
1001        let (second_handler_event_sender, mut second_handler_event_receiver) =
1002            futures::channel::mpsc::channel(100);
1003        let second_input_handler =
1004            observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
1005                second_handler_event_sender,
1006            );
1007
1008        // Build the input pipeline.
1009        let (sender, receiver, handlers, _, _, _) =
1010            InputPipelineAssembly::new(metrics::MetricsLogger::default())
1011                .add_handler(first_input_handler)
1012                .add_handler(second_input_handler)
1013                .into_components();
1014        let inspector = fuchsia_inspect::Inspector::default();
1015        let test_node = inspector.root().create_child("input_pipeline");
1016        let input_pipeline = InputPipeline {
1017            pipeline_sender: sender,
1018            device_event_sender,
1019            device_event_receiver,
1020            input_device_types: vec![],
1021            input_device_bindings: Arc::new(Mutex::new(SortedVecMap::new())),
1022            inspect_node: test_node,
1023            metrics_logger: metrics::MetricsLogger::default(),
1024            feature_flags: input_device::InputPipelineFeatureFlags::default(),
1025            _tasks: vec![],
1026            _fasync_tasks: vec![],
1027        };
1028        let _runner_task =
1029            InputPipeline::run(receiver, handlers, metrics::MetricsLogger::default());
1030
1031        // Send an input event.
1032        let input_events = send_input_event(input_device_binding.input_event_sender());
1033
1034        // Run the pipeline.
1035        let _pipeline_task = fasync::Task::local(async {
1036            input_pipeline.handle_input_events().await;
1037        });
1038
1039        // Assert both handlers receive the event.
1040        let expected_event = input_events.into_iter().next();
1041        let first_handler_event = first_handler_event_receiver.next().await;
1042        assert_eq!(first_handler_event, expected_event);
1043        let second_handler_event = second_handler_event_receiver.next().await;
1044        assert_eq!(second_handler_event, expected_event);
1045    }
1046
1047    /// Tests that a single mouse device binding is created for the one input device in the
1048    /// input report directory.
1049    #[fasync::run_singlethreaded(test)]
1050    async fn watch_devices_one_match_exists() {
1051        // Create a file in a pseudo directory that represents an input device.
1052        let mut count: i8 = 0;
1053        let dir = pseudo_directory! {
1054            "file_name" => pseudo_fs_service::host(
1055                move |mut request_stream: fidl_fuchsia_input_report::InputDeviceRequestStream| {
1056                    async move {
1057                        while count < 3 {
1058                            if let Some(input_device_request) =
1059                                request_stream.try_next().await.unwrap()
1060                            {
1061                                handle_input_device_request(input_device_request);
1062                                count += 1;
1063                            }
1064                        }
1065
1066                    }.boxed()
1067                },
1068            )
1069        };
1070
1071        // Create a Watcher on the pseudo directory.
1072        let dir_proxy_for_watcher = vfs::directory::serve_read_only(
1073            dir.clone(),
1074            vfs::execution_scope::ExecutionScope::new(),
1075        );
1076        let device_watcher = Watcher::new(&dir_proxy_for_watcher).await.unwrap();
1077        // Get a proxy to the pseudo directory for the input pipeline. The input pipeline uses this
1078        // proxy to get connections to input devices.
1079        let dir_proxy_for_pipeline =
1080            vfs::directory::serve_read_only(dir, vfs::execution_scope::ExecutionScope::new());
1081
1082        let (input_event_sender, _input_event_receiver) = futures::channel::mpsc::unbounded();
1083        let bindings: InputDeviceBindingMap = Arc::new(Mutex::new(SortedVecMap::new()));
1084        let supported_device_types = vec![input_device::InputDeviceType::Mouse];
1085
1086        let inspector = fuchsia_inspect::Inspector::default();
1087        let test_node = inspector.root().create_child("input_pipeline");
1088        test_node.record_string(
1089            "supported_input_devices",
1090            supported_device_types.clone().iter().join(", "),
1091        );
1092        let input_devices = test_node.create_child("input_devices");
1093        // Assert that inspect tree is initialized with no devices.
1094        diagnostics_assertions::assert_data_tree!(inspector, root: {
1095            input_pipeline: {
1096                supported_input_devices: "Mouse",
1097                input_devices: {}
1098            }
1099        });
1100
1101        let _ = InputPipeline::watch_for_devices(
1102            device_watcher,
1103            dir_proxy_for_pipeline,
1104            supported_device_types,
1105            input_event_sender,
1106            bindings.clone(),
1107            &input_devices,
1108            true, /* break_on_idle */
1109            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1110            metrics::MetricsLogger::default(),
1111        )
1112        .await;
1113
1114        // Assert that one mouse device with accurate device id was found.
1115        let bindings_map = bindings.lock();
1116        assert_eq!(bindings_map.len(), 1);
1117        let bindings_vector = bindings_map.get(&10);
1118        assert!(bindings_vector.is_some());
1119        assert_eq!(bindings_vector.unwrap().len(), 1);
1120        let boxed_mouse_binding = bindings_vector.unwrap().get(0);
1121        assert!(boxed_mouse_binding.is_some());
1122        assert_eq!(
1123            boxed_mouse_binding.unwrap().get_device_descriptor(),
1124            input_device::InputDeviceDescriptor::Mouse(mouse_binding::MouseDeviceDescriptor {
1125                device_id: 10,
1126                absolute_x_range: None,
1127                absolute_y_range: None,
1128                wheel_v_range: None,
1129                wheel_h_range: None,
1130                buttons: Some(vec![0]),
1131            })
1132        );
1133
1134        // Assert that inspect tree reflects new device discovered and connected.
1135        diagnostics_assertions::assert_data_tree!(inspector, root: {
1136            input_pipeline: {
1137                supported_input_devices: "Mouse",
1138                input_devices: {
1139                    devices_discovered: 1u64,
1140                    devices_connected: 1u64,
1141                    "file_name_Mouse": contains {
1142                        reports_received_count: 0u64,
1143                        reports_filtered_count: 0u64,
1144                        events_generated: 0u64,
1145                        last_received_timestamp_ns: 0u64,
1146                        last_generated_timestamp_ns: 0u64,
1147                        "fuchsia.inspect.Health": {
1148                            status: "OK",
1149                            // Timestamp value is unpredictable and not relevant in this context,
1150                            // so we only assert that the property is present.
1151                            start_timestamp_nanos: AnyProperty
1152                        },
1153                    }
1154                }
1155            }
1156        });
1157    }
1158
1159    /// Tests that no device bindings are created because the input pipeline looks for keyboard devices
1160    /// but only a mouse exists.
1161    #[fasync::run_singlethreaded(test)]
1162    async fn watch_devices_no_matches_exist() {
1163        // Create a file in a pseudo directory that represents an input device.
1164        let mut count: i8 = 0;
1165        let dir = pseudo_directory! {
1166            "file_name" => pseudo_fs_service::host(
1167                move |mut request_stream: fidl_fuchsia_input_report::InputDeviceRequestStream| {
1168                    async move {
1169                        while count < 1 {
1170                            if let Some(input_device_request) =
1171                                request_stream.try_next().await.unwrap()
1172                            {
1173                                handle_input_device_request(input_device_request);
1174                                count += 1;
1175                            }
1176                        }
1177
1178                    }.boxed()
1179                },
1180            )
1181        };
1182
1183        // Create a Watcher on the pseudo directory.
1184        let dir_proxy_for_watcher = vfs::directory::serve_read_only(
1185            dir.clone(),
1186            vfs::execution_scope::ExecutionScope::new(),
1187        );
1188        let device_watcher = Watcher::new(&dir_proxy_for_watcher).await.unwrap();
1189        // Get a proxy to the pseudo directory for the input pipeline. The input pipeline uses this
1190        // proxy to get connections to input devices.
1191        let dir_proxy_for_pipeline =
1192            vfs::directory::serve_read_only(dir, vfs::execution_scope::ExecutionScope::new());
1193
1194        let (input_event_sender, _input_event_receiver) = futures::channel::mpsc::unbounded();
1195        let bindings: InputDeviceBindingMap = Arc::new(Mutex::new(SortedVecMap::new()));
1196        let supported_device_types = vec![input_device::InputDeviceType::Keyboard];
1197
1198        let inspector = fuchsia_inspect::Inspector::default();
1199        let test_node = inspector.root().create_child("input_pipeline");
1200        test_node.record_string(
1201            "supported_input_devices",
1202            supported_device_types.clone().iter().join(", "),
1203        );
1204        let input_devices = test_node.create_child("input_devices");
1205        // Assert that inspect tree is initialized with no devices.
1206        diagnostics_assertions::assert_data_tree!(inspector, root: {
1207            input_pipeline: {
1208                supported_input_devices: "Keyboard",
1209                input_devices: {}
1210            }
1211        });
1212
1213        let _ = InputPipeline::watch_for_devices(
1214            device_watcher,
1215            dir_proxy_for_pipeline,
1216            supported_device_types,
1217            input_event_sender,
1218            bindings.clone(),
1219            &input_devices,
1220            true, /* break_on_idle */
1221            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1222            metrics::MetricsLogger::default(),
1223        )
1224        .await;
1225
1226        // Assert that no devices were found.
1227        let bindings = bindings.lock();
1228        assert_eq!(bindings.len(), 0);
1229
1230        // Assert that inspect tree reflects new device discovered, but not connected.
1231        diagnostics_assertions::assert_data_tree!(inspector, root: {
1232            input_pipeline: {
1233                supported_input_devices: "Keyboard",
1234                input_devices: {
1235                    devices_discovered: 1u64,
1236                    devices_connected: 0u64,
1237                    "file_name_Unsupported": {
1238                        "fuchsia.inspect.Health": {
1239                            status: "UNHEALTHY",
1240                            message: "Unsupported device type.",
1241                            // Timestamp value is unpredictable and not relevant in this context,
1242                            // so we only assert that the property is present.
1243                            start_timestamp_nanos: AnyProperty
1244                        },
1245                    }
1246                }
1247            }
1248        });
1249    }
1250
1251    /// Tests that a single keyboard device binding is created for the input device registered
1252    /// through InputDeviceRegistry.
1253    #[fasync::run_singlethreaded(test)]
1254    async fn handle_input_device_registry_request_stream() {
1255        let (input_device_registry_proxy, input_device_registry_request_stream) =
1256            create_proxy_and_stream::<fidl_fuchsia_input_injection::InputDeviceRegistryMarker>();
1257        let (input_device_client_end, mut input_device_request_stream) =
1258            create_request_stream::<fidl_fuchsia_input_report::InputDeviceMarker>();
1259
1260        let device_types = vec![input_device::InputDeviceType::Mouse];
1261        let (input_event_sender, _input_event_receiver) = futures::channel::mpsc::unbounded();
1262        let bindings: InputDeviceBindingMap = Arc::new(Mutex::new(SortedVecMap::new()));
1263
1264        // Handle input device requests.
1265        let mut count: i8 = 0;
1266        let _task = fasync::Task::local(async move {
1267            // Register a device.
1268            let _ = input_device_registry_proxy.register(input_device_client_end);
1269
1270            while count < 3 {
1271                if let Some(input_device_request) =
1272                    input_device_request_stream.try_next().await.unwrap()
1273                {
1274                    handle_input_device_request(input_device_request);
1275                    count += 1;
1276                }
1277            }
1278
1279            // End handle_input_device_registry_request_stream() by taking the event stream.
1280            input_device_registry_proxy.take_event_stream();
1281        });
1282
1283        let inspector = fuchsia_inspect::Inspector::default();
1284        let test_node = inspector.root().create_child("input_pipeline");
1285
1286        // Start listening for InputDeviceRegistryRequests.
1287        let bindings_clone = bindings.clone();
1288        let _ = InputPipeline::handle_input_device_registry_request_stream(
1289            input_device_registry_request_stream,
1290            &device_types,
1291            &input_event_sender,
1292            &bindings_clone,
1293            &test_node,
1294            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1295            metrics::MetricsLogger::default(),
1296        )
1297        .await;
1298
1299        // Assert that a device was registered.
1300        let bindings = bindings.lock();
1301        assert_eq!(bindings.len(), 1);
1302    }
1303
1304    // Tests that correct properties are added to inspect node when InputPipeline is created.
1305    #[fasync::run_singlethreaded(test)]
1306    async fn check_inspect_node_has_correct_properties() {
1307        let device_types = vec![
1308            input_device::InputDeviceType::Touch,
1309            input_device::InputDeviceType::ConsumerControls,
1310        ];
1311        let inspector = fuchsia_inspect::Inspector::default();
1312        let test_node = inspector.root().create_child("input_pipeline");
1313        // Create fake input handler for assembly
1314        let (fake_handler_event_sender, _fake_handler_event_receiver) =
1315            futures::channel::mpsc::channel(100);
1316        let fake_input_handler =
1317            observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
1318                fake_handler_event_sender,
1319            );
1320        let assembly = InputPipelineAssembly::new(metrics::MetricsLogger::default())
1321            .add_handler(fake_input_handler);
1322        let _test_input_pipeline = InputPipeline::new(
1323            &Incoming::new(),
1324            device_types,
1325            assembly,
1326            test_node,
1327            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1328            metrics::MetricsLogger::default(),
1329        );
1330        diagnostics_assertions::assert_data_tree!(inspector, root: {
1331            input_pipeline: {
1332                supported_input_devices: "Touch, ConsumerControls",
1333                handlers_registered: 1u64,
1334                handlers_healthy: 1u64,
1335                input_devices: {}
1336            }
1337        });
1338    }
1339
1340    struct SpecificInterestFakeHandler {
1341        interest_types: Vec<input_device::InputEventType>,
1342        event_sender: std::cell::RefCell<futures::channel::mpsc::Sender<input_device::InputEvent>>,
1343    }
1344
1345    impl SpecificInterestFakeHandler {
1346        pub fn new(
1347            interest_types: Vec<input_device::InputEventType>,
1348            event_sender: futures::channel::mpsc::Sender<input_device::InputEvent>,
1349        ) -> Rc<Self> {
1350            Rc::new(SpecificInterestFakeHandler {
1351                interest_types,
1352                event_sender: std::cell::RefCell::new(event_sender),
1353            })
1354        }
1355    }
1356
1357    impl Handler for SpecificInterestFakeHandler {
1358        fn set_handler_healthy(self: std::rc::Rc<Self>) {}
1359        fn set_handler_unhealthy(self: std::rc::Rc<Self>, _msg: &str) {}
1360        fn get_name(&self) -> &'static str {
1361            "SpecificInterestFakeHandler"
1362        }
1363
1364        fn interest(&self) -> Vec<input_device::InputEventType> {
1365            self.interest_types.clone()
1366        }
1367    }
1368
1369    #[async_trait(?Send)]
1370    impl input_handler::InputHandler for SpecificInterestFakeHandler {
1371        async fn handle_input_event(
1372            self: Rc<Self>,
1373            input_event: input_device::InputEvent,
1374        ) -> Vec<input_device::InputEvent> {
1375            match self.event_sender.borrow_mut().try_send(input_event.clone()) {
1376                Err(e) => panic!("SpecificInterestFakeHandler failed to send event: {:?}", e),
1377                Ok(_) => {}
1378            }
1379            vec![input_event]
1380        }
1381    }
1382
1383    #[fasync::run_singlethreaded(test)]
1384    async fn run_only_sends_events_to_interested_handlers() {
1385        // Mouse Handler (Specific Interest: Mouse)
1386        let (mouse_sender, mut mouse_receiver) = futures::channel::mpsc::channel(1);
1387        let mouse_handler =
1388            SpecificInterestFakeHandler::new(vec![InputEventType::Mouse], mouse_sender);
1389
1390        // Fake Handler (Specific Interest: Fake)
1391        let (fake_sender, mut fake_receiver) = futures::channel::mpsc::channel(1);
1392        let fake_handler =
1393            SpecificInterestFakeHandler::new(vec![InputEventType::Fake], fake_sender);
1394
1395        let (pipeline_sender, pipeline_receiver, handlers, _, _, _) =
1396            InputPipelineAssembly::new(metrics::MetricsLogger::default())
1397                .add_handler(mouse_handler)
1398                .add_handler(fake_handler)
1399                .into_components();
1400
1401        // Run the pipeline logic
1402        let _runner_task =
1403            InputPipeline::run(pipeline_receiver, handlers, metrics::MetricsLogger::default());
1404
1405        // Create a Fake event
1406        let fake_event = input_device::InputEvent {
1407            device_event: input_device::InputDeviceEvent::Fake,
1408            device_descriptor: input_device::InputDeviceDescriptor::Fake,
1409            event_time: zx::MonotonicInstant::get(),
1410            handled: input_device::Handled::No,
1411            trace_id: None,
1412        };
1413
1414        // Send the Fake event
1415        pipeline_sender.unbounded_send(vec![fake_event.clone()]).expect("failed to send event");
1416
1417        // Verify Fake Handler received it
1418        let received_by_fake = fake_receiver.next().await;
1419        assert_eq!(received_by_fake, Some(fake_event));
1420
1421        // Verify Mouse Handler did NOT receive it
1422        assert!(mouse_receiver.try_next().is_err());
1423    }
1424
1425    fn create_mouse_event(x: f32, y: f32) -> input_device::InputEvent {
1426        input_device::InputEvent {
1427            device_event: input_device::InputDeviceEvent::Mouse(mouse_binding::MouseEvent::new(
1428                mouse_binding::MouseLocation::Relative(mouse_binding::RelativeLocation {
1429                    counts: Position { x, y },
1430                }),
1431                None,
1432                None,
1433                mouse_binding::MousePhase::Move,
1434                SortedVecSet::new(),
1435                SortedVecSet::new(),
1436                None,
1437                None,
1438            )),
1439            device_descriptor: input_device::InputDeviceDescriptor::Mouse(
1440                mouse_binding::MouseDeviceDescriptor {
1441                    device_id: 1,
1442                    absolute_x_range: None,
1443                    absolute_y_range: None,
1444                    wheel_v_range: None,
1445                    wheel_h_range: None,
1446                    buttons: None,
1447                },
1448            ),
1449            event_time: zx::MonotonicInstant::get(),
1450            handled: input_device::Handled::No,
1451            trace_id: None,
1452        }
1453    }
1454
1455    #[fasync::run_singlethreaded(test)]
1456    async fn run_mixed_event_types_dispatched_correctly() {
1457        // Mouse Handler (Specific Interest: Mouse)
1458        let (mouse_sender, mut mouse_receiver) = futures::channel::mpsc::channel(10);
1459        let mouse_handler =
1460            SpecificInterestFakeHandler::new(vec![InputEventType::Mouse], mouse_sender);
1461
1462        // Fake Handler (Specific Interest: Fake)
1463        let (fake_sender, mut fake_receiver) = futures::channel::mpsc::channel(10);
1464        let fake_handler =
1465            SpecificInterestFakeHandler::new(vec![InputEventType::Fake], fake_sender);
1466
1467        let (pipeline_sender, pipeline_receiver, handlers, _, _, _) =
1468            InputPipelineAssembly::new(metrics::MetricsLogger::default())
1469                .add_handler(mouse_handler)
1470                .add_handler(fake_handler)
1471                .into_components();
1472
1473        // Run the pipeline logic
1474        let _runner_task =
1475            InputPipeline::run(pipeline_receiver, handlers, metrics::MetricsLogger::default());
1476
1477        // Create events
1478        let mouse_event_1 = create_mouse_event(1.0, 1.0);
1479        let mouse_event_2 = create_mouse_event(2.0, 2.0);
1480        let mouse_event_3 = create_mouse_event(3.0, 3.0);
1481
1482        let fake_event_1 = input_device::InputEvent {
1483            device_event: input_device::InputDeviceEvent::Fake,
1484            device_descriptor: input_device::InputDeviceDescriptor::Fake,
1485            event_time: zx::MonotonicInstant::get(),
1486            handled: input_device::Handled::No,
1487            trace_id: None,
1488        };
1489
1490        // Send mixed batch: [Mouse, Mouse, Fake, Mouse]
1491        // This should result in 3 chunks: [Mouse, Mouse], [Fake], [Mouse]
1492        let mixed_batch = vec![
1493            mouse_event_1.clone(),
1494            mouse_event_2.clone(),
1495            fake_event_1.clone(),
1496            mouse_event_3.clone(),
1497        ];
1498        pipeline_sender.unbounded_send(mixed_batch).expect("failed to send events");
1499
1500        // Verify Mouse Handler received M1, M2, and then M3
1501        assert_eq!(mouse_receiver.next().await, Some(mouse_event_1));
1502        assert_eq!(mouse_receiver.next().await, Some(mouse_event_2));
1503        assert_eq!(mouse_receiver.next().await, Some(mouse_event_3));
1504
1505        // Verify Fake Handler received F1
1506        assert_eq!(fake_receiver.next().await, Some(fake_event_1));
1507    }
1508}
input_pipeline/input_pipeline.rs

input_pipeline/
input_pipeline.rs
