input_pipeline_dso/
task.rs

1// Copyright 2026 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 futures::future::LocalBoxFuture;
6use futures::prelude::*;
7
8/// A helper structure that manages the execution of all asynchronous tasks
9/// in the input pipeline concurrently.
10pub struct InputPipelineTasks {
11    pub watcher: LocalBoxFuture<'static, ()>,
12    pub runner: LocalBoxFuture<'static, ()>,
13    pub display_ownership: Option<LocalBoxFuture<'static, ()>>,
14    pub focus_listener: Option<LocalBoxFuture<'static, ()>>,
15    pub forwarder: LocalBoxFuture<'static, ()>,
16}
17
18impl InputPipelineTasks {
19    /// Runs all tasks concurrently. Returns when any of the critical tasks completes.
20    pub async fn run(self) {
21        let InputPipelineTasks { watcher, runner, display_ownership, focus_listener, forwarder } =
22            self;
23
24        let mut watcher = watcher.fuse();
25        let mut runner = runner.fuse();
26        let mut display_ownership =
27            display_ownership.unwrap_or_else(|| Box::pin(futures::future::pending())).fuse();
28        let mut focus_listener =
29            focus_listener.unwrap_or_else(|| Box::pin(futures::future::pending())).fuse();
30        let mut forwarder = forwarder.fuse();
31
32        loop {
33            futures::select! {
34                _ = watcher => {
35                    // Watcher finished (e.g. break_on_idle in tests).
36                    // This is non-fatal; continue running remaining tasks.
37                },
38                _ = runner => break,
39                _ = display_ownership => break,
40                _ = focus_listener => break,
41                _ = forwarder => break,
42                complete => break,
43            }
44        }
45    }
46}
47
48#[cfg(test)]
49mod tests {
50    use super::*;
51    use futures::channel::oneshot;
52
53    #[fuchsia::test]
54    async fn run_does_not_exit_on_watcher_completion() {
55        let (forwarder_tx, forwarder_rx) = oneshot::channel::<()>();
56        let tasks = InputPipelineTasks {
57            watcher: Box::pin(async {}),
58            runner: Box::pin(futures::future::pending()),
59            display_ownership: None,
60            focus_listener: None,
61            forwarder: Box::pin(async move {
62                let _ = forwarder_rx.await;
63            }),
64        };
65
66        let mut run_fut = tasks.run().boxed_local();
67
68        // Watcher finishes immediately, but run_fut shouldn't exit until a critical task finishes.
69        assert!(futures::poll!(&mut run_fut).is_pending());
70
71        // Send signal to forwarder: now run_fut should complete.
72        let _ = forwarder_tx.send(());
73        run_fut.await;
74    }
75
76    #[fuchsia::test]
77    async fn run_exits_on_runner_completion() {
78        let (runner_tx, runner_rx) = oneshot::channel::<()>();
79        let tasks = InputPipelineTasks {
80            watcher: Box::pin(async {}),
81            runner: Box::pin(async move {
82                let _ = runner_rx.await;
83            }),
84            display_ownership: None,
85            focus_listener: None,
86            forwarder: Box::pin(futures::future::pending()),
87        };
88
89        let mut run_fut = tasks.run().boxed_local();
90        assert!(futures::poll!(&mut run_fut).is_pending());
91
92        let _ = runner_tx.send(());
93        run_fut.await;
94    }
95}
input_pipeline_dso/task.rs

input_pipeline_dso/
task.rs
