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::Path;
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(async move {
293                fut.await;
294                panic!("display_ownership_fut exited unexpectedly, which compromises device state tracking. Terminating to avoid inconsistent state.");
295            }));
296            handlers_count += 1;
297        }
298
299        // TODO: b/469745447 - should use futures::select! instead of spawning tasks.
300        if let Some(fut) = focus_listener_fut {
301            fasync_tasks.push(fasync::Task::local(async move {
302                fut.await;
303                panic!("focus_listener_fut exited unexpectedly, which breaks input routing. Terminating to avoid inconsistent state.");
304            }));
305            handlers_count += 1;
306        }
307
308        // Add properties to inspect node
309        inspect_node.record_string("supported_input_devices", input_device_types.iter().join(", "));
310        inspect_node.record_uint("handlers_registered", handlers_count as u64);
311        inspect_node.record_uint("handlers_healthy", handlers_count as u64);
312
313        // Initializes all handlers and starts the input pipeline loop.
314        let runner_task = InputPipeline::run(receiver, handlers, metrics_logger.clone());
315        tasks.push(runner_task);
316
317        let (device_event_sender, device_event_receiver) = futures::channel::mpsc::unbounded();
318        let input_device_bindings: InputDeviceBindingMap =
319            Arc::new(Mutex::new(SortedVecMap::new()));
320        InputPipeline {
321            pipeline_sender,
322            device_event_sender,
323            device_event_receiver,
324            input_device_types,
325            input_device_bindings,
326            inspect_node,
327            metrics_logger,
328            feature_flags,
329            _tasks: tasks,
330            _fasync_tasks: fasync_tasks,
331        }
332    }
333
334    /// Creates a new [`InputPipeline`] for integration testing.
335    /// Unlike a production input pipeline, this pipeline will not monitor
336    /// `/svc/fuchsia.input.report.Service` for devices.
337    ///
338    /// # Parameters
339    /// - `input_device_types`: The types of devices the new [`InputPipeline`] will support.
340    /// - `assembly`: The input handlers that the [`InputPipeline`] sends InputEvents to.
341    pub fn new_for_test(
342        input_device_types: Vec<input_device::InputDeviceType>,
343        assembly: InputPipelineAssembly,
344    ) -> Self {
345        let inspector = fuchsia_inspect::Inspector::default();
346        let root = inspector.root();
347        let test_node = root.create_child("input_pipeline");
348        Self::new_common(
349            input_device_types,
350            assembly,
351            test_node,
352            input_device::InputPipelineFeatureFlags { enable_merge_touch_events: false },
353        )
354    }
355
356    /// Creates a new [`InputPipeline`] for production use.
357    ///
358    /// # Parameters
359    /// - `input_device_types`: The types of devices the new [`InputPipeline`] will support.
360    /// - `assembly`: The input handlers that the [`InputPipeline`] sends InputEvents to.
361    /// - `inspect_node`: The root node for InputPipeline's Inspect tree
362    pub fn new(
363        incoming: &Incoming,
364        input_device_types: Vec<input_device::InputDeviceType>,
365        assembly: InputPipelineAssembly,
366        inspect_node: fuchsia_inspect::Node,
367        feature_flags: input_device::InputPipelineFeatureFlags,
368        metrics_logger: metrics::MetricsLogger,
369    ) -> Result<Self, Error> {
370        let mut input_pipeline =
371            Self::new_common(input_device_types, assembly, inspect_node, feature_flags);
372        let input_device_types = input_pipeline.input_device_types.clone();
373        let input_event_sender = input_pipeline.device_event_sender.clone();
374        let input_device_bindings = input_pipeline.input_device_bindings.clone();
375        let devices_node = input_pipeline.inspect_node.create_child("input_devices");
376        let feature_flags = input_pipeline.feature_flags.clone();
377        let incoming = incoming.clone();
378        // This intentionally uses the [`fuchsia_async`] task dispatcher instead of
379        // [`crate::Dispatcher`] -- the directory watcher always uses the fuchsia-async dispatcher.
380        // This is fine for performance because the actual event dispatch is still configured to
381        // run on [`crate::Dispatcher`].
382        let watcher_task = fasync::Task::local(async move {
383            // Watches the input device directory for new input devices. Creates new InputDeviceBindings
384            // that send InputEvents to `input_event_receiver`.
385            match async {
386                let (dir_proxy, server) = endpoints::create_proxy::<fio::DirectoryMarker>();
387                incoming
388                    .as_ref_directory()
389                    .open(input_device::INPUT_REPORT_PATH, fio::PERM_READABLE, server.into())
390                    .with_context(|| {
391                        format!("failed to open {}", input_device::INPUT_REPORT_PATH)
392                    })?;
393                let device_watcher =
394                    Watcher::new(&dir_proxy).await.context("failed to create watcher")?;
395                Self::watch_for_devices(
396                    device_watcher,
397                    dir_proxy,
398                    &input_device_types,
399                    input_event_sender,
400                    input_device_bindings,
401                    &devices_node,
402                    false, /* break_on_idle */
403                    feature_flags,
404                    metrics_logger.clone(),
405                )
406                .await
407                .context("failed to watch for devices")
408            }
409            .await
410            {
411                Ok(()) => {}
412                Err(err) => {
413                    // This error is usually benign in tests: it means that the setup does not
414                    // support dynamic device discovery. Almost no tests support dynamic
415                    // device discovery, and they also do not need those.
416                    metrics_logger.log_warn(
417                        InputPipelineErrorMetricDimensionEvent::InputPipelineUnableToWatchForNewInputDevices,
418                        std::format!(
419                            "Input pipeline is unable to watch for new input devices: {:?}",
420                            err
421                        ));
422                }
423            }
424        });
425        input_pipeline._fasync_tasks.push(watcher_task);
426
427        Ok(input_pipeline)
428    }
429
430    /// Gets the input device bindings.
431    pub fn input_device_bindings(&self) -> &InputDeviceBindingMap {
432        &self.input_device_bindings
433    }
434
435    /// Gets the input device sender: this is the channel that should be cloned
436    /// and used for injecting events from the drivers into the input pipeline.
437    pub fn input_event_sender(&self) -> &UnboundedSender<Vec<input_device::InputEvent>> {
438        &self.device_event_sender
439    }
440
441    /// Gets a list of input device types supported by this input pipeline.
442    pub fn input_device_types(&self) -> &[input_device::InputDeviceType] {
443        &self.input_device_types
444    }
445
446    /// Forwards all input events into the input pipeline.
447    pub async fn handle_input_events(mut self) {
448        let metrics_logger_clone = self.metrics_logger.clone();
449        while let Some(input_event) = self.device_event_receiver.next().await {
450            if let Err(e) = self.pipeline_sender.unbounded_send(input_event) {
451                metrics_logger_clone.log_error(
452                    InputPipelineErrorMetricDimensionEvent::InputPipelineCouldNotForwardEventFromDriver,
453                    std::format!("could not forward event from driver: {:?}", e));
454            }
455        }
456
457        metrics_logger_clone.log_error(
458            InputPipelineErrorMetricDimensionEvent::InputPipelineStopHandlingEvents,
459            "Input pipeline stopped handling input events.".to_string(),
460        );
461    }
462
463    /// Watches the input report service directory for new input devices. Creates InputDeviceBindings
464    /// if new devices match a type in `device_types`.
465    ///
466    /// # Parameters
467    /// - `device_watcher`: Watches the input report service directory for new devices.
468    /// - `dir_proxy`: The directory containing InputDevice connections.
469    /// - `device_types`: The types of devices to watch for.
470    /// - `input_event_sender`: The channel new InputDeviceBindings will send InputEvents to.
471    /// - `bindings`: Holds all the InputDeviceBindings
472    /// - `input_devices_node`: The parent node for all device bindings' inspect nodes.
473    /// - `break_on_idle`: If true, stops watching for devices once all existing devices are handled.
474    /// - `metrics_logger`: The metrics logger.
475    ///
476    /// # Errors
477    /// If the input report service directory or a file within it cannot be read.
478    async fn watch_for_devices(
479        mut device_watcher: Watcher,
480        dir_proxy: fio::DirectoryProxy,
481        device_types: &[input_device::InputDeviceType],
482        input_event_sender: UnboundedSender<Vec<input_device::InputEvent>>,
483        bindings: InputDeviceBindingMap,
484        input_devices_node: &fuchsia_inspect::Node,
485        break_on_idle: bool,
486        feature_flags: input_device::InputPipelineFeatureFlags,
487        metrics_logger: metrics::MetricsLogger,
488    ) -> Result<(), Error> {
489        // Add non-static properties to inspect node.
490        let devices_discovered = input_devices_node.create_uint("devices_discovered", 0);
491        let devices_connected = input_devices_node.create_uint("devices_connected", 0);
492        while let Some(msg) = device_watcher.try_next().await? {
493            if let Ok(instance_name) = msg.filename.into_os_string().into_string() {
494                if instance_name == "." {
495                    continue;
496                }
497
498                match msg.event {
499                    WatchEvent::EXISTING | WatchEvent::ADD_FILE => {
500                        log::info!("found input device {}", instance_name);
501                        devices_discovered.add(1);
502                        let device_path = Path::new(&instance_name).join("input_device");
503                        let device_proxy = match input_device::get_device_from_dir_entry_path(
504                            &dir_proxy,
505                            &device_path,
506                        ) {
507                            Ok(proxy) => proxy,
508                            Err(e) => {
509                                log::error!(
510                                    "Failed to connect to input device {}: {:?}",
511                                    instance_name,
512                                    e
513                                );
514                                continue;
515                            }
516                        };
517                        add_device_bindings(
518                            device_types,
519                            &instance_name,
520                            device_proxy,
521                            &input_event_sender,
522                            &bindings,
523                            get_next_device_id(),
524                            input_devices_node,
525                            Some(&devices_connected),
526                            feature_flags.clone(),
527                            metrics_logger.clone(),
528                            false,
529                        )
530                        .await;
531                    }
532                    WatchEvent::IDLE => {
533                        if break_on_idle {
534                            break;
535                        }
536                    }
537                    _ => (),
538                }
539            }
540        }
541        // Ensure inspect properties persist for debugging if device watch loop ends.
542        input_devices_node.record(devices_discovered);
543        input_devices_node.record(devices_connected);
544        Err(format_err!("Input pipeline stopped watching for new input devices."))
545    }
546
547    /// Handles the incoming InputDeviceRegistryRequestStream.
548    ///
549    /// This method will end when the request stream is closed. If the stream closes with an
550    /// error the error will be returned in the Result.
551    ///
552    /// **NOTE**: Only one stream is handled at a time. https://fxbug.dev/42061078
553    ///
554    /// # Parameters
555    /// - `stream`: The stream of InputDeviceRegistryRequests.
556    /// - `device_types`: The types of devices to watch for.
557    /// - `input_event_sender`: The channel new InputDeviceBindings will send InputEvents to.
558    /// - `bindings`: Holds all the InputDeviceBindings associated with the InputPipeline.
559    /// - `input_devices_node`: The parent node for all injected devices' inspect nodes.
560    /// - `metrics_logger`: The metrics logger.
561    pub async fn handle_input_device_registry_request_stream(
562        mut stream: fidl_fuchsia_input_injection::InputDeviceRegistryRequestStream,
563        device_types: &[input_device::InputDeviceType],
564        input_event_sender: &UnboundedSender<Vec<input_device::InputEvent>>,
565        bindings: &InputDeviceBindingMap,
566        input_devices_node: &fuchsia_inspect::Node,
567        feature_flags: input_device::InputPipelineFeatureFlags,
568        metrics_logger: metrics::MetricsLogger,
569    ) -> Result<(), Error> {
570        while let Some(request) = stream
571            .try_next()
572            .await
573            .context("Error handling input device registry request stream")?
574        {
575            match request {
576                fidl_fuchsia_input_injection::InputDeviceRegistryRequest::Register {
577                    device,
578                    ..
579                } => {
580                    // Add a binding if the device is a type being tracked
581                    let device = fidl_next::ClientEnd::<
582                        fidl_next_fuchsia_input_report::InputDevice,
583                        zx::Channel,
584                    >::from_untyped(device.into_channel());
585                    let device = Dispatcher::client_from_zx_channel(device);
586                    let device = device.spawn();
587                    let device_id = get_next_device_id();
588
589                    add_device_bindings(
590                        device_types,
591                        &format!("input-device-registry-{}", device_id),
592                        device,
593                        input_event_sender,
594                        bindings,
595                        device_id,
596                        input_devices_node,
597                        None,
598                        feature_flags.clone(),
599                        metrics_logger.clone(),
600                        true,
601                    )
602                    .await;
603                }
604                fidl_fuchsia_input_injection::InputDeviceRegistryRequest::RegisterAndGetDeviceInfo {
605                    device,
606                    responder,
607                    .. } => {
608                    // Add a binding if the device is a type being tracked
609                    let device = fidl_next::ClientEnd::<
610                        fidl_next_fuchsia_input_report::InputDevice,
611                        zx::Channel,
612                    >::from_untyped(device.into_channel());
613                    let device = Dispatcher::client_from_zx_channel(device);
614                    let device = device.spawn();
615                    let device_id = get_next_device_id();
616
617                    add_device_bindings(
618                        device_types,
619                        &format!("input-device-registry-{}", device_id),
620                        device,
621                        input_event_sender,
622                        bindings,
623                        device_id,
624                        input_devices_node,
625                        None,
626                        feature_flags.clone(),
627                        metrics_logger.clone(),
628                        true,
629                    )
630                    .await;
631
632                    responder.send(fidl_fuchsia_input_injection::InputDeviceRegistryRegisterAndGetDeviceInfoResponse{
633                        device_id: Some(device_id),
634                        ..Default::default()
635                    }).expect("Failed to respond to RegisterAndGetDeviceInfo request");
636                }
637            }
638        }
639
640        Ok(())
641    }
642
643    /// Initializes all handlers and starts the input pipeline loop in an asynchronous executor.
644    fn run(
645        mut receiver: UnboundedReceiver<Vec<input_device::InputEvent>>,
646        handlers: Vec<Rc<dyn input_handler::BatchInputHandler>>,
647        metrics_logger: metrics::MetricsLogger,
648    ) -> dispatcher::TaskHandle<()> {
649        Dispatcher::spawn_local(async move {
650            for handler in &handlers {
651                handler.clone().set_handler_healthy();
652            }
653
654            let mut handlers_by_type: [Vec<Rc<dyn input_handler::BatchInputHandler>>;
655                InputEventType::COUNT] = Default::default();
656
657            // TODO: b/478262850 - We can use supported_input_devices to populate this list.
658            let event_types = vec![
659                InputEventType::Keyboard,
660                InputEventType::LightSensor,
661                InputEventType::ConsumerControls,
662                InputEventType::Mouse,
663                InputEventType::TouchScreen,
664                InputEventType::Touchpad,
665                #[cfg(test)]
666                InputEventType::Fake,
667            ];
668
669            for event_type in event_types {
670                let handlers_for_type: Vec<Rc<dyn input_handler::BatchInputHandler>> = handlers
671                    .iter()
672                    .filter(|h| h.interest().contains(&event_type))
673                    .cloned()
674                    .collect();
675                handlers_by_type[event_type as usize] = handlers_for_type;
676            }
677
678            while let Some(events) = receiver.next().await {
679                if events.is_empty() {
680                    continue;
681                }
682
683                let mut groups_seen = 0;
684                let events = events.into_iter().chunk_by(|e| InputEventType::from(&e.device_event));
685                let events = events.into_iter().map(|(k, v)| (k, v.collect::<Vec<_>>()));
686                for (event_type, event_group) in events {
687                    groups_seen += 1;
688                    if groups_seen == 2 {
689                        metrics_logger.log_error(
690                                InputPipelineErrorMetricDimensionEvent::InputFrameContainsMultipleTypesOfEvents,
691                                "it is not recommended to contain multiple types of events in 1 send".to_string(),
692                            );
693                    }
694                    let mut events_in_group = event_group;
695
696                    // Get pre-computed handlers for this event type.
697                    let handlers = &handlers_by_type[event_type as usize];
698
699                    for handler in handlers {
700                        events_in_group =
701                            handler.clone().handle_input_events(events_in_group).await;
702                    }
703
704                    for event in events_in_group {
705                        if event.handled == input_device::Handled::No {
706                            log::warn!("unhandled input event: {:?}", event);
707                        }
708                        if let Some(trace_id) = event.trace_id {
709                            fuchsia_trace::flow_end!(
710                                "input",
711                                "event_in_input_pipeline",
712                                trace_id.into()
713                            );
714                        }
715                    }
716                }
717            }
718            for handler in &handlers {
719                handler.clone().set_handler_unhealthy("Pipeline loop terminated");
720            }
721            panic!("Runner task is not supposed to terminate.")
722        })
723    }
724}
725
726/// Adds `InputDeviceBinding`s to `bindings` for all `device_types` exposed by `device_proxy`.
727///
728/// # Parameters
729/// - `device_types`: The types of devices to watch for.
730/// - `device_proxy`: A proxy to the input device.
731/// - `input_event_sender`: The channel new InputDeviceBindings will send InputEvents to.
732/// - `bindings`: Holds all the InputDeviceBindings associated with the InputPipeline.
733/// - `device_id`: The device id of the associated bindings.
734/// - `input_devices_node`: The parent node for all device bindings' inspect nodes.
735///
736/// # Note
737/// This will create multiple bindings, in the case where
738/// * `device_proxy().get_descriptor()` returns a `fidl_fuchsia_input_report::DeviceDescriptor`
739///   with multiple table fields populated, and
740/// * multiple populated table fields correspond to device types present in `device_types`
741///
742/// This is used, for example, to support the Atlas touchpad. In that case, a single
743/// instance of `fuchsia.input.report.Service` provides both a `fuchsia.input.report.MouseDescriptor` and
744/// a `fuchsia.input.report.TouchDescriptor`.
745async fn add_device_bindings(
746    device_types: &[input_device::InputDeviceType],
747    instance_name: &str,
748    device_proxy: fidl_next::Client<fidl_next_fuchsia_input_report::InputDevice, Transport>,
749    input_event_sender: &UnboundedSender<Vec<input_device::InputEvent>>,
750    bindings: &InputDeviceBindingMap,
751    device_id: u32,
752    input_devices_node: &fuchsia_inspect::Node,
753    devices_connected: Option<&fuchsia_inspect::UintProperty>,
754    feature_flags: InputPipelineFeatureFlags,
755    metrics_logger: metrics::MetricsLogger,
756    is_injected: bool,
757) {
758    let mut matched_device_types = vec![];
759    if let Ok(res) = device_proxy.get_descriptor().await {
760        for device_type in device_types {
761            if input_device::is_device_type(&res.descriptor, *device_type).await {
762                matched_device_types.push(device_type);
763                match devices_connected {
764                    Some(dev_connected) => {
765                        let _ = dev_connected.add(1);
766                    }
767                    None => (),
768                };
769            }
770        }
771        if matched_device_types.is_empty() {
772            log::info!(
773                "device {} did not match any supported device types: {:?}",
774                instance_name,
775                device_types
776            );
777            let device_node =
778                input_devices_node.create_child(format!("{}_Unsupported", instance_name));
779            let mut health = fuchsia_inspect::health::Node::new(&device_node);
780            health.set_unhealthy("Unsupported device type.");
781            device_node.record(health);
782            input_devices_node.record(device_node);
783            return;
784        }
785    } else {
786        metrics_logger.clone().log_error(
787            InputPipelineErrorMetricDimensionEvent::InputPipelineNoDeviceDescriptor,
788            std::format!("cannot bind device {} without a device descriptor", instance_name),
789        );
790        return;
791    }
792
793    log::info!(
794        "binding {} to device types: {}",
795        instance_name,
796        matched_device_types
797            .iter()
798            .fold(String::new(), |device_types_string, device_type| device_types_string
799                + &format!("{:?}, ", device_type))
800    );
801
802    let mut new_bindings: Vec<BoxedInputDeviceBinding> = vec![];
803    for device_type in matched_device_types {
804        // Clone `device_proxy`, so that multiple bindings (e.g. a `MouseBinding` and a
805        // `TouchBinding`) can read data from the same `fuchsia.input.report.Service` instance.
806        //
807        // There's no conflict in having multiple bindings read from the same node,
808        // since:
809        // * each binding will create its own `fuchsia.input.report.InputReportsReader`, and
810        // * the device driver will copy each incoming report to each connected reader.
811        //
812        // This does mean that reports from the Atlas touchpad device get read twice
813        // (by a `MouseBinding` and a `TouchBinding`), regardless of whether the device
814        // is operating in mouse mode or touchpad mode.
815        //
816        // This hasn't been an issue because:
817        // * Semantically: things are fine, because each binding discards irrelevant reports.
818        //   (E.g. `MouseBinding` discards anything that isn't a `MouseInputReport`), and
819        // * Performance wise: things are fine, because the data rate of the touchpad is low
820        //   (125 HZ).
821        //
822        // If we add additional cases where bindings share an underlying service instance,
823        // we might consider adding a multiplexing binding, to avoid reading duplicate reports.
824        let proxy = device_proxy.clone();
825        let device_node =
826            input_devices_node.create_child(format!("{}_{}", instance_name, device_type));
827        match input_device::get_device_binding(
828            *device_type,
829            proxy,
830            device_id,
831            input_event_sender.clone(),
832            device_node,
833            feature_flags.clone(),
834            metrics_logger.clone(),
835            is_injected,
836        )
837        .await
838        {
839            Ok(binding) => new_bindings.push(binding),
840            Err(e) => {
841                metrics_logger.log_error(
842                    InputPipelineErrorMetricDimensionEvent::InputPipelineFailedToBind,
843                    std::format!("failed to bind {} as {:?}: {}", instance_name, device_type, e),
844                );
845            }
846        }
847    }
848
849    if !new_bindings.is_empty() {
850        let mut bindings = bindings.lock();
851        if let Some(v) = bindings.get_mut(&device_id) {
852            v.extend(new_bindings);
853        } else {
854            bindings.insert(device_id, new_bindings);
855        }
856    }
857}
858
859#[cfg(test)]
860mod tests {
861    use super::*;
862    use crate::input_device::{InputDeviceBinding, InputEventType};
863    use crate::utils::Position;
864    use crate::{fake_input_device_binding, mouse_binding, observe_fake_events_input_handler};
865    use async_trait::async_trait;
866    use diagnostics_assertions::AnyProperty;
867    use fidl::endpoints::{create_proxy_and_stream, create_request_stream};
868    use fuchsia_async as fasync;
869    use futures::FutureExt;
870    use pretty_assertions::assert_eq;
871    use rand::Rng;
872    use sorted_vec_map::SortedVecSet;
873    use vfs::{pseudo_directory, service as pseudo_fs_service};
874
875    /// Returns the InputEvent sent over `sender`.
876    ///
877    /// # Parameters
878    /// - `sender`: The channel to send the InputEvent over.
879    fn send_input_event(
880        sender: UnboundedSender<Vec<input_device::InputEvent>>,
881    ) -> Vec<input_device::InputEvent> {
882        let mut rng = rand::rng();
883        let offset =
884            Position { x: rng.random_range(0..10) as f32, y: rng.random_range(0..10) as f32 };
885        let input_event = input_device::InputEvent {
886            device_event: input_device::InputDeviceEvent::Mouse(mouse_binding::MouseEvent::new(
887                mouse_binding::MouseLocation::Relative(mouse_binding::RelativeLocation {
888                    counts: Position { x: offset.x, y: offset.y },
889                }),
890                None, /* wheel_delta_v */
891                None, /* wheel_delta_h */
892                mouse_binding::MousePhase::Move,
893                SortedVecSet::new(),
894                SortedVecSet::new(),
895                None, /* is_precision_scroll */
896                None, /* wake_lease */
897            )),
898            device_descriptor: input_device::InputDeviceDescriptor::Mouse(
899                mouse_binding::MouseDeviceDescriptor {
900                    device_id: 1,
901                    absolute_x_range: None,
902                    absolute_y_range: None,
903                    wheel_v_range: None,
904                    wheel_h_range: None,
905                    buttons: None,
906                },
907            ),
908            event_time: zx::MonotonicInstant::get(),
909            handled: input_device::Handled::No,
910            trace_id: None,
911        };
912        match sender.unbounded_send(vec![input_event.clone()]) {
913            Err(_) => assert!(false),
914            _ => {}
915        }
916
917        vec![input_event]
918    }
919
920    /// Returns a MouseDescriptor on an InputDeviceRequest.
921    ///
922    /// # Parameters
923    /// - `input_device_request`: The request to handle.
924    fn handle_input_device_request(
925        input_device_request: fidl_fuchsia_input_report::InputDeviceRequest,
926    ) {
927        match input_device_request {
928            fidl_fuchsia_input_report::InputDeviceRequest::GetDescriptor { responder } => {
929                let _ = responder.send(&fidl_fuchsia_input_report::DeviceDescriptor {
930                    device_information: None,
931                    mouse: Some(fidl_fuchsia_input_report::MouseDescriptor {
932                        input: Some(fidl_fuchsia_input_report::MouseInputDescriptor {
933                            movement_x: None,
934                            movement_y: None,
935                            scroll_v: None,
936                            scroll_h: None,
937                            buttons: Some(vec![0]),
938                            position_x: None,
939                            position_y: None,
940                            ..Default::default()
941                        }),
942                        ..Default::default()
943                    }),
944                    sensor: None,
945                    touch: None,
946                    keyboard: None,
947                    consumer_control: None,
948                    ..Default::default()
949                });
950            }
951            _ => {}
952        }
953    }
954
955    /// Tests that an input pipeline handles events from multiple devices.
956    #[fasync::run_singlethreaded(test)]
957    async fn multiple_devices_single_handler() {
958        // Create two fake device bindings.
959        let (device_event_sender, device_event_receiver) = futures::channel::mpsc::unbounded();
960        let first_device_binding =
961            fake_input_device_binding::FakeInputDeviceBinding::new(device_event_sender.clone());
962        let second_device_binding =
963            fake_input_device_binding::FakeInputDeviceBinding::new(device_event_sender.clone());
964
965        // Create a fake input handler.
966        let (handler_event_sender, mut handler_event_receiver) =
967            futures::channel::mpsc::channel(100);
968        let input_handler = observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
969            handler_event_sender,
970        );
971
972        // Build the input pipeline.
973        let (sender, receiver, handlers, _, _, _) =
974            InputPipelineAssembly::new(metrics::MetricsLogger::default())
975                .add_handler(input_handler)
976                .into_components();
977        let inspector = fuchsia_inspect::Inspector::default();
978        let test_node = inspector.root().create_child("input_pipeline");
979        let input_pipeline = InputPipeline {
980            pipeline_sender: sender,
981            device_event_sender,
982            device_event_receiver,
983            input_device_types: vec![],
984            input_device_bindings: Arc::new(Mutex::new(SortedVecMap::new())),
985            inspect_node: test_node,
986            metrics_logger: metrics::MetricsLogger::default(),
987            feature_flags: input_device::InputPipelineFeatureFlags::default(),
988            _tasks: vec![],
989            _fasync_tasks: vec![],
990        };
991        let _runner_task =
992            InputPipeline::run(receiver, handlers, metrics::MetricsLogger::default());
993
994        // Send an input event from each device.
995        let first_device_events = send_input_event(first_device_binding.input_event_sender());
996        let second_device_events = send_input_event(second_device_binding.input_event_sender());
997
998        // Run the pipeline.
999        let _pipeline_task = fasync::Task::local(async {
1000            input_pipeline.handle_input_events().await;
1001        });
1002
1003        // Assert the handler receives the events.
1004        let first_handled_event = handler_event_receiver.next().await;
1005        assert_eq!(first_handled_event, first_device_events.into_iter().next());
1006
1007        let second_handled_event = handler_event_receiver.next().await;
1008        assert_eq!(second_handled_event, second_device_events.into_iter().next());
1009    }
1010
1011    /// Tests that an input pipeline handles events through multiple input handlers.
1012    #[fasync::run_singlethreaded(test)]
1013    async fn single_device_multiple_handlers() {
1014        // Create two fake device bindings.
1015        let (device_event_sender, device_event_receiver) = futures::channel::mpsc::unbounded();
1016        let input_device_binding =
1017            fake_input_device_binding::FakeInputDeviceBinding::new(device_event_sender.clone());
1018
1019        // Create two fake input handlers.
1020        let (first_handler_event_sender, mut first_handler_event_receiver) =
1021            futures::channel::mpsc::channel(100);
1022        let first_input_handler =
1023            observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
1024                first_handler_event_sender,
1025            );
1026        let (second_handler_event_sender, mut second_handler_event_receiver) =
1027            futures::channel::mpsc::channel(100);
1028        let second_input_handler =
1029            observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
1030                second_handler_event_sender,
1031            );
1032
1033        // Build the input pipeline.
1034        let (sender, receiver, handlers, _, _, _) =
1035            InputPipelineAssembly::new(metrics::MetricsLogger::default())
1036                .add_handler(first_input_handler)
1037                .add_handler(second_input_handler)
1038                .into_components();
1039        let inspector = fuchsia_inspect::Inspector::default();
1040        let test_node = inspector.root().create_child("input_pipeline");
1041        let input_pipeline = InputPipeline {
1042            pipeline_sender: sender,
1043            device_event_sender,
1044            device_event_receiver,
1045            input_device_types: vec![],
1046            input_device_bindings: Arc::new(Mutex::new(SortedVecMap::new())),
1047            inspect_node: test_node,
1048            metrics_logger: metrics::MetricsLogger::default(),
1049            feature_flags: input_device::InputPipelineFeatureFlags::default(),
1050            _tasks: vec![],
1051            _fasync_tasks: vec![],
1052        };
1053        let _runner_task =
1054            InputPipeline::run(receiver, handlers, metrics::MetricsLogger::default());
1055
1056        // Send an input event.
1057        let input_events = send_input_event(input_device_binding.input_event_sender());
1058
1059        // Run the pipeline.
1060        let _pipeline_task = fasync::Task::local(async {
1061            input_pipeline.handle_input_events().await;
1062        });
1063
1064        // Assert both handlers receive the event.
1065        let expected_event = input_events.into_iter().next();
1066        let first_handler_event = first_handler_event_receiver.next().await;
1067        assert_eq!(first_handler_event, expected_event);
1068        let second_handler_event = second_handler_event_receiver.next().await;
1069        assert_eq!(second_handler_event, expected_event);
1070    }
1071
1072    /// Tests that a single mouse device binding is created for the one input device in the
1073    /// input report service directory.
1074    #[fasync::run_singlethreaded(test)]
1075    async fn watch_devices_one_match_exists() {
1076        // Create a pseudo directory representing a service instance for an input device.
1077        let mut count: i8 = 0;
1078        let dir = pseudo_directory! {
1079            "instance_0" => pseudo_directory! {
1080                "input_device" => pseudo_fs_service::host(
1081                    move |mut request_stream: fidl_fuchsia_input_report::InputDeviceRequestStream| {
1082                        async move {
1083                            while count < 3 {
1084                                if let Some(input_device_request) =
1085                                    request_stream.try_next().await.unwrap()
1086                                {
1087                                    handle_input_device_request(input_device_request);
1088                                    count += 1;
1089                                }
1090                            }
1091
1092                        }.boxed()
1093                    },
1094                )
1095            }
1096        };
1097
1098        // Create a Watcher on the pseudo directory.
1099        let dir_proxy_for_watcher = vfs::directory::serve_read_only(
1100            dir.clone(),
1101            vfs::execution_scope::ExecutionScope::new(),
1102        );
1103        let device_watcher = Watcher::new(&dir_proxy_for_watcher).await.unwrap();
1104        // Get a proxy to the pseudo directory for the input pipeline. The input pipeline uses this
1105        // proxy to get connections to input devices.
1106        let dir_proxy_for_pipeline =
1107            vfs::directory::serve_read_only(dir, vfs::execution_scope::ExecutionScope::new());
1108
1109        let (input_event_sender, _input_event_receiver) = futures::channel::mpsc::unbounded();
1110        let bindings: InputDeviceBindingMap = Arc::new(Mutex::new(SortedVecMap::new()));
1111        let supported_device_types = vec![input_device::InputDeviceType::Mouse];
1112
1113        let inspector = fuchsia_inspect::Inspector::default();
1114        let test_node = inspector.root().create_child("input_pipeline");
1115        test_node.record_string(
1116            "supported_input_devices",
1117            supported_device_types.clone().iter().join(", "),
1118        );
1119        let input_devices = test_node.create_child("input_devices");
1120        // Assert that inspect tree is initialized with no devices.
1121        diagnostics_assertions::assert_data_tree!(inspector, root: {
1122            input_pipeline: {
1123                supported_input_devices: "Mouse",
1124                input_devices: {}
1125            }
1126        });
1127
1128        let _ = InputPipeline::watch_for_devices(
1129            device_watcher,
1130            dir_proxy_for_pipeline,
1131            &supported_device_types,
1132            input_event_sender,
1133            bindings.clone(),
1134            &input_devices,
1135            true, /* break_on_idle */
1136            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1137            metrics::MetricsLogger::default(),
1138        )
1139        .await;
1140
1141        // Assert that one mouse device with accurate device id was found.
1142        let bindings_map = bindings.lock();
1143        assert_eq!(bindings_map.len(), 1);
1144        let bindings_vector = bindings_map.get(&10);
1145        assert!(bindings_vector.is_some());
1146        assert_eq!(bindings_vector.unwrap().len(), 1);
1147        let boxed_mouse_binding = bindings_vector.unwrap().get(0);
1148        assert!(boxed_mouse_binding.is_some());
1149        assert_eq!(
1150            boxed_mouse_binding.unwrap().get_device_descriptor(),
1151            input_device::InputDeviceDescriptor::Mouse(mouse_binding::MouseDeviceDescriptor {
1152                device_id: 10,
1153                absolute_x_range: None,
1154                absolute_y_range: None,
1155                wheel_v_range: None,
1156                wheel_h_range: None,
1157                buttons: Some(vec![0]),
1158            })
1159        );
1160
1161        // Assert that inspect tree reflects new device discovered and connected.
1162        diagnostics_assertions::assert_data_tree!(inspector, root: {
1163            input_pipeline: {
1164                supported_input_devices: "Mouse",
1165                input_devices: {
1166                    devices_discovered: 1u64,
1167                    devices_connected: 1u64,
1168                    "instance_0_Mouse": contains {
1169                        reports_received_count: 0u64,
1170                        reports_filtered_count: 0u64,
1171                        events_generated: 0u64,
1172                        last_received_timestamp_ns: 0u64,
1173                        last_generated_timestamp_ns: 0u64,
1174                        "fuchsia.inspect.Health": {
1175                            status: "OK",
1176                            // Timestamp value is unpredictable and not relevant in this context,
1177                            // so we only assert that the property is present.
1178                            start_timestamp_nanos: AnyProperty
1179                        },
1180                    }
1181                }
1182            }
1183        });
1184    }
1185
1186    /// Tests that no device bindings are created because the input pipeline looks for keyboard devices
1187    /// but only a mouse exists.
1188    #[fasync::run_singlethreaded(test)]
1189    async fn watch_devices_no_matches_exist() {
1190        // Create a pseudo directory representing a service instance for an input device.
1191        let mut count: i8 = 0;
1192        let dir = pseudo_directory! {
1193            "instance_0" => pseudo_directory! {
1194                "input_device" => pseudo_fs_service::host(
1195                    move |mut request_stream: fidl_fuchsia_input_report::InputDeviceRequestStream| {
1196                        async move {
1197                            while count < 1 {
1198                                if let Some(input_device_request) =
1199                                    request_stream.try_next().await.unwrap()
1200                                {
1201                                    handle_input_device_request(input_device_request);
1202                                    count += 1;
1203                                }
1204                            }
1205
1206                        }.boxed()
1207                    },
1208                )
1209            }
1210        };
1211
1212        // Create a Watcher on the pseudo directory.
1213        let dir_proxy_for_watcher = vfs::directory::serve_read_only(
1214            dir.clone(),
1215            vfs::execution_scope::ExecutionScope::new(),
1216        );
1217        let device_watcher = Watcher::new(&dir_proxy_for_watcher).await.unwrap();
1218        // Get a proxy to the pseudo directory for the input pipeline. The input pipeline uses this
1219        // proxy to get connections to input devices.
1220        let dir_proxy_for_pipeline =
1221            vfs::directory::serve_read_only(dir, vfs::execution_scope::ExecutionScope::new());
1222
1223        let (input_event_sender, _input_event_receiver) = futures::channel::mpsc::unbounded();
1224        let bindings: InputDeviceBindingMap = Arc::new(Mutex::new(SortedVecMap::new()));
1225        let supported_device_types = vec![input_device::InputDeviceType::Keyboard];
1226
1227        let inspector = fuchsia_inspect::Inspector::default();
1228        let test_node = inspector.root().create_child("input_pipeline");
1229        test_node.record_string(
1230            "supported_input_devices",
1231            supported_device_types.clone().iter().join(", "),
1232        );
1233        let input_devices = test_node.create_child("input_devices");
1234        // Assert that inspect tree is initialized with no devices.
1235        diagnostics_assertions::assert_data_tree!(inspector, root: {
1236            input_pipeline: {
1237                supported_input_devices: "Keyboard",
1238                input_devices: {}
1239            }
1240        });
1241
1242        let _ = InputPipeline::watch_for_devices(
1243            device_watcher,
1244            dir_proxy_for_pipeline,
1245            &supported_device_types,
1246            input_event_sender,
1247            bindings.clone(),
1248            &input_devices,
1249            true, /* break_on_idle */
1250            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1251            metrics::MetricsLogger::default(),
1252        )
1253        .await;
1254
1255        // Assert that no devices were found.
1256        let bindings = bindings.lock();
1257        assert_eq!(bindings.len(), 0);
1258
1259        // Assert that inspect tree reflects new device discovered, but not connected.
1260        diagnostics_assertions::assert_data_tree!(inspector, root: {
1261            input_pipeline: {
1262                supported_input_devices: "Keyboard",
1263                input_devices: {
1264                    devices_discovered: 1u64,
1265                    devices_connected: 0u64,
1266                    "instance_0_Unsupported": {
1267                        "fuchsia.inspect.Health": {
1268                            status: "UNHEALTHY",
1269                            message: "Unsupported device type.",
1270                            // Timestamp value is unpredictable and not relevant in this context,
1271                            // so we only assert that the property is present.
1272                            start_timestamp_nanos: AnyProperty
1273                        },
1274                    }
1275                }
1276            }
1277        });
1278    }
1279
1280    /// Tests that a single keyboard device binding is created for the input device registered
1281    /// through InputDeviceRegistry.
1282    #[fasync::run_singlethreaded(test)]
1283    async fn handle_input_device_registry_request_stream() {
1284        let (input_device_registry_proxy, input_device_registry_request_stream) =
1285            create_proxy_and_stream::<fidl_fuchsia_input_injection::InputDeviceRegistryMarker>();
1286        let (input_device_client_end, mut input_device_request_stream) =
1287            create_request_stream::<fidl_fuchsia_input_report::InputDeviceMarker>();
1288
1289        let device_types = vec![input_device::InputDeviceType::Mouse];
1290        let (input_event_sender, _input_event_receiver) = futures::channel::mpsc::unbounded();
1291        let bindings: InputDeviceBindingMap = Arc::new(Mutex::new(SortedVecMap::new()));
1292
1293        // Handle input device requests.
1294        let mut count: i8 = 0;
1295        let _task = fasync::Task::local(async move {
1296            // Register a device.
1297            let _ = input_device_registry_proxy.register(input_device_client_end);
1298
1299            while count < 3 {
1300                if let Some(input_device_request) =
1301                    input_device_request_stream.try_next().await.unwrap()
1302                {
1303                    handle_input_device_request(input_device_request);
1304                    count += 1;
1305                }
1306            }
1307
1308            // End handle_input_device_registry_request_stream() by taking the event stream.
1309            input_device_registry_proxy.take_event_stream();
1310        });
1311
1312        let inspector = fuchsia_inspect::Inspector::default();
1313        let test_node = inspector.root().create_child("input_pipeline");
1314
1315        // Start listening for InputDeviceRegistryRequests.
1316        let bindings_clone = bindings.clone();
1317        let _ = InputPipeline::handle_input_device_registry_request_stream(
1318            input_device_registry_request_stream,
1319            &device_types,
1320            &input_event_sender,
1321            &bindings_clone,
1322            &test_node,
1323            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1324            metrics::MetricsLogger::default(),
1325        )
1326        .await;
1327
1328        // Assert that a device was registered.
1329        let bindings = bindings.lock();
1330        assert_eq!(bindings.len(), 1);
1331    }
1332
1333    // Tests that correct properties are added to inspect node when InputPipeline is created.
1334    #[fasync::run_singlethreaded(test)]
1335    async fn check_inspect_node_has_correct_properties() {
1336        let device_types = vec![
1337            input_device::InputDeviceType::Touch,
1338            input_device::InputDeviceType::ConsumerControls,
1339        ];
1340        let inspector = fuchsia_inspect::Inspector::default();
1341        let test_node = inspector.root().create_child("input_pipeline");
1342        // Create fake input handler for assembly
1343        let (fake_handler_event_sender, _fake_handler_event_receiver) =
1344            futures::channel::mpsc::channel(100);
1345        let fake_input_handler =
1346            observe_fake_events_input_handler::ObserveFakeEventsInputHandler::new(
1347                fake_handler_event_sender,
1348            );
1349        let assembly = InputPipelineAssembly::new(metrics::MetricsLogger::default())
1350            .add_handler(fake_input_handler);
1351        let _test_input_pipeline = InputPipeline::new(
1352            &Incoming::new(),
1353            device_types,
1354            assembly,
1355            test_node,
1356            InputPipelineFeatureFlags { enable_merge_touch_events: false },
1357            metrics::MetricsLogger::default(),
1358        );
1359        diagnostics_assertions::assert_data_tree!(inspector, root: {
1360            input_pipeline: {
1361                supported_input_devices: "Touch, ConsumerControls",
1362                handlers_registered: 1u64,
1363                handlers_healthy: 1u64,
1364                input_devices: {}
1365            }
1366        });
1367    }
1368
1369    struct SpecificInterestFakeHandler {
1370        interest_types: Vec<input_device::InputEventType>,
1371        event_sender: std::cell::RefCell<futures::channel::mpsc::Sender<input_device::InputEvent>>,
1372    }
1373
1374    impl SpecificInterestFakeHandler {
1375        pub fn new(
1376            interest_types: Vec<input_device::InputEventType>,
1377            event_sender: futures::channel::mpsc::Sender<input_device::InputEvent>,
1378        ) -> Rc<Self> {
1379            Rc::new(SpecificInterestFakeHandler {
1380                interest_types,
1381                event_sender: std::cell::RefCell::new(event_sender),
1382            })
1383        }
1384    }
1385
1386    impl Handler for SpecificInterestFakeHandler {
1387        fn set_handler_healthy(self: std::rc::Rc<Self>) {}
1388        fn set_handler_unhealthy(self: std::rc::Rc<Self>, _msg: &str) {}
1389        fn get_name(&self) -> &'static str {
1390            "SpecificInterestFakeHandler"
1391        }
1392
1393        fn interest(&self) -> Vec<input_device::InputEventType> {
1394            self.interest_types.clone()
1395        }
1396    }
1397
1398    #[async_trait(?Send)]
1399    impl input_handler::InputHandler for SpecificInterestFakeHandler {
1400        async fn handle_input_event(
1401            self: Rc<Self>,
1402            input_event: input_device::InputEvent,
1403        ) -> Vec<input_device::InputEvent> {
1404            match self.event_sender.borrow_mut().try_send(input_event.clone()) {
1405                Err(e) => panic!("SpecificInterestFakeHandler failed to send event: {:?}", e),
1406                Ok(_) => {}
1407            }
1408            vec![input_event]
1409        }
1410    }
1411
1412    #[fasync::run_singlethreaded(test)]
1413    async fn run_only_sends_events_to_interested_handlers() {
1414        // Mouse Handler (Specific Interest: Mouse)
1415        let (mouse_sender, mut mouse_receiver) = futures::channel::mpsc::channel(1);
1416        let mouse_handler =
1417            SpecificInterestFakeHandler::new(vec![InputEventType::Mouse], mouse_sender);
1418
1419        // Fake Handler (Specific Interest: Fake)
1420        let (fake_sender, mut fake_receiver) = futures::channel::mpsc::channel(1);
1421        let fake_handler =
1422            SpecificInterestFakeHandler::new(vec![InputEventType::Fake], fake_sender);
1423
1424        let (pipeline_sender, pipeline_receiver, handlers, _, _, _) =
1425            InputPipelineAssembly::new(metrics::MetricsLogger::default())
1426                .add_handler(mouse_handler)
1427                .add_handler(fake_handler)
1428                .into_components();
1429
1430        // Run the pipeline logic
1431        let _runner_task =
1432            InputPipeline::run(pipeline_receiver, handlers, metrics::MetricsLogger::default());
1433
1434        // Create a Fake event
1435        let fake_event = input_device::InputEvent {
1436            device_event: input_device::InputDeviceEvent::Fake,
1437            device_descriptor: input_device::InputDeviceDescriptor::Fake,
1438            event_time: zx::MonotonicInstant::get(),
1439            handled: input_device::Handled::No,
1440            trace_id: None,
1441        };
1442
1443        // Send the Fake event
1444        pipeline_sender.unbounded_send(vec![fake_event.clone()]).expect("failed to send event");
1445
1446        // Verify Fake Handler received it
1447        let received_by_fake = fake_receiver.next().await;
1448        assert_eq!(received_by_fake, Some(fake_event));
1449
1450        // Verify Mouse Handler did NOT receive it
1451        assert!(mouse_receiver.try_next().is_err());
1452    }
1453
1454    fn create_mouse_event(x: f32, y: f32) -> input_device::InputEvent {
1455        input_device::InputEvent {
1456            device_event: input_device::InputDeviceEvent::Mouse(mouse_binding::MouseEvent::new(
1457                mouse_binding::MouseLocation::Relative(mouse_binding::RelativeLocation {
1458                    counts: Position { x, y },
1459                }),
1460                None,
1461                None,
1462                mouse_binding::MousePhase::Move,
1463                SortedVecSet::new(),
1464                SortedVecSet::new(),
1465                None,
1466                None,
1467            )),
1468            device_descriptor: input_device::InputDeviceDescriptor::Mouse(
1469                mouse_binding::MouseDeviceDescriptor {
1470                    device_id: 1,
1471                    absolute_x_range: None,
1472                    absolute_y_range: None,
1473                    wheel_v_range: None,
1474                    wheel_h_range: None,
1475                    buttons: None,
1476                },
1477            ),
1478            event_time: zx::MonotonicInstant::get(),
1479            handled: input_device::Handled::No,
1480            trace_id: None,
1481        }
1482    }
1483
1484    #[fasync::run_singlethreaded(test)]
1485    async fn run_mixed_event_types_dispatched_correctly() {
1486        // Mouse Handler (Specific Interest: Mouse)
1487        let (mouse_sender, mut mouse_receiver) = futures::channel::mpsc::channel(10);
1488        let mouse_handler =
1489            SpecificInterestFakeHandler::new(vec![InputEventType::Mouse], mouse_sender);
1490
1491        // Fake Handler (Specific Interest: Fake)
1492        let (fake_sender, mut fake_receiver) = futures::channel::mpsc::channel(10);
1493        let fake_handler =
1494            SpecificInterestFakeHandler::new(vec![InputEventType::Fake], fake_sender);
1495
1496        let (pipeline_sender, pipeline_receiver, handlers, _, _, _) =
1497            InputPipelineAssembly::new(metrics::MetricsLogger::default())
1498                .add_handler(mouse_handler)
1499                .add_handler(fake_handler)
1500                .into_components();
1501
1502        // Run the pipeline logic
1503        let _runner_task =
1504            InputPipeline::run(pipeline_receiver, handlers, metrics::MetricsLogger::default());
1505
1506        // Create events
1507        let mouse_event_1 = create_mouse_event(1.0, 1.0);
1508        let mouse_event_2 = create_mouse_event(2.0, 2.0);
1509        let mouse_event_3 = create_mouse_event(3.0, 3.0);
1510
1511        let fake_event_1 = input_device::InputEvent {
1512            device_event: input_device::InputDeviceEvent::Fake,
1513            device_descriptor: input_device::InputDeviceDescriptor::Fake,
1514            event_time: zx::MonotonicInstant::get(),
1515            handled: input_device::Handled::No,
1516            trace_id: None,
1517        };
1518
1519        // Send mixed batch: [Mouse, Mouse, Fake, Mouse]
1520        // This should result in 3 chunks: [Mouse, Mouse], [Fake], [Mouse]
1521        let mixed_batch = vec![
1522            mouse_event_1.clone(),
1523            mouse_event_2.clone(),
1524            fake_event_1.clone(),
1525            mouse_event_3.clone(),
1526        ];
1527        pipeline_sender.unbounded_send(mixed_batch).expect("failed to send events");
1528
1529        // Verify Mouse Handler received M1, M2, and then M3
1530        assert_eq!(mouse_receiver.next().await, Some(mouse_event_1));
1531        assert_eq!(mouse_receiver.next().await, Some(mouse_event_2));
1532        assert_eq!(mouse_receiver.next().await, Some(mouse_event_3));
1533
1534        // Verify Fake Handler received F1
1535        assert_eq!(fake_receiver.next().await, Some(fake_event_1));
1536    }
1537}
input_pipeline/input_pipeline.rs

input_pipeline/
input_pipeline.rs
