irtt_client/managed/
blocking.rs

1use std::{
2    thread::{self, JoinHandle},
3    time::Duration,
4};
5
6use thiserror::Error;
7use tokio::runtime::{Builder, Runtime};
8
9use super::{
10    ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, ManagedConfigError,
11    ManagedEventSubscription, ManagedOutcome, ManagedSubscribeError, ManagedTargetConfig,
12};
13
14const BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(100);
15
16/// Owns the runtime used exclusively by the blocking managed worker.
17pub(super) struct WorkerRuntime {
18    runtime: Option<Runtime>,
19}
20
21impl WorkerRuntime {
22    pub(super) fn new(runtime: Runtime) -> Self {
23        Self {
24            runtime: Some(runtime),
25        }
26    }
27
28    fn block_on<F: std::future::Future>(&self, task: F) -> F::Output {
29        self.runtime
30            .as_ref()
31            .expect("worker runtime is available until shutdown")
32            .block_on(task)
33    }
34
35    /// Shut down residual runtime work without delaying managed completion indefinitely.
36    pub(super) fn shutdown(mut self) {
37        self.runtime
38            .take()
39            .expect("worker runtime is available until shutdown")
40            .shutdown_timeout(BLOCKING_RUNTIME_SHUTDOWN_TIMEOUT);
41    }
42}
43
44impl Drop for WorkerRuntime {
45    fn drop(&mut self) {
46        if let Some(runtime) = self.runtime.take() {
47            runtime.shutdown_background();
48        }
49    }
50}
51
52/// Failure to start a blocking managed client owner.
53#[derive(Debug, Error)]
54pub enum BlockingManagedStartError {
55    /// The managed task configuration was invalid.
56    #[error(transparent)]
57    Config(#[from] ManagedConfigError),
58
59    /// The managed event subscription could not be created before startup.
60    #[error(transparent)]
61    Subscribe(#[from] ManagedSubscribeError),
62
63    /// The dedicated Tokio runtime could not be created.
64    #[error("failed to create blocking managed Tokio runtime: {source}")]
65    Runtime {
66        #[source]
67        source: std::io::Error,
68    },
69
70    /// The dedicated worker thread could not be spawned.
71    #[error("failed to spawn blocking managed worker thread: {source}")]
72    ThreadSpawn {
73        #[source]
74        source: std::io::Error,
75    },
76}
77
78/// Failure to obtain completion from a blocking managed client owner.
79#[derive(Debug, Error, Eq, PartialEq)]
80pub enum BlockingManagedJoinError {
81    /// The dedicated managed worker thread panicked.
82    #[error("blocking managed worker thread panicked")]
83    WorkerPanicked,
84}
85
86/// Blocking completion owner for one unified managed Tokio task.
87///
88/// This owner runs exactly one [`ManagedClient`] task on one dedicated OS thread
89/// with its own current-thread Tokio runtime. Dropping a live owner requests a
90/// graceful stop and synchronously joins that worker, so drop may block while
91/// bounded managed cleanup completes.
92#[must_use = "dropping the blocking managed owner requests stop and joins its worker"]
93pub struct BlockingManagedClient {
94    handle: ManagedClientHandle,
95    worker: Option<JoinHandle<ManagedOutcome>>,
96}
97
98impl BlockingManagedClient {
99    /// Validate configuration, create a dedicated runtime, and start its worker.
100    pub fn start(
101        config: ManagedClientConfig,
102        targets: Vec<ManagedTargetConfig>,
103    ) -> Result<Self, BlockingManagedStartError> {
104        let (task, handle) = ManagedClient::task(config, targets)?;
105        Self::start_task(task, handle)
106    }
107
108    /// Validate configuration, subscribe to presentation events, and start a
109    /// dedicated managed worker.
110    ///
111    /// The returned subscription exists before the managed task receives its
112    /// first poll, so it cannot miss startup events because of subscription
113    /// ordering.
114    pub fn start_with_subscription(
115        config: ManagedClientConfig,
116        targets: Vec<ManagedTargetConfig>,
117    ) -> Result<(Self, ManagedEventSubscription), BlockingManagedStartError> {
118        let (task, handle) = ManagedClient::task(config, targets)?;
119        let subscription = handle.subscribe()?;
120        let owner = Self::start_task(task, handle)?;
121        Ok((owner, subscription))
122    }
123
124    fn start_task(
125        task: ManagedClientTask,
126        handle: ManagedClientHandle,
127    ) -> Result<Self, BlockingManagedStartError> {
128        let runtime = Builder::new_current_thread()
129            .enable_io()
130            .enable_time()
131            .build()
132            .map_err(|source| BlockingManagedStartError::Runtime { source })?;
133        let runtime = WorkerRuntime::new(runtime);
134        let worker = thread::Builder::new()
135            .name("irtt-managed".to_owned())
136            .spawn(move || {
137                let outcome = runtime.block_on(task);
138                runtime.shutdown();
139                outcome
140            })
141            .map_err(|source| BlockingManagedStartError::ThreadSpawn { source })?;
142
143        Ok(Self {
144            handle,
145            worker: Some(worker),
146        })
147    }
148
149    /// Return a cloned managed control and observation capability.
150    pub fn handle(&self) -> ManagedClientHandle {
151        self.handle.clone()
152    }
153
154    /// Wait synchronously for the managed task's authoritative outcome.
155    pub fn join(mut self) -> Result<ManagedOutcome, BlockingManagedJoinError> {
156        self.join_worker()
157    }
158
159    fn join_worker(&mut self) -> Result<ManagedOutcome, BlockingManagedJoinError> {
160        self.worker
161            .take()
162            .expect("blocking managed worker is joined at most once")
163            .join()
164            .map_err(|_| BlockingManagedJoinError::WorkerPanicked)
165    }
166}
167
168impl Drop for BlockingManagedClient {
169    fn drop(&mut self) {
170        if self.worker.is_some() {
171            let _receipt = self.handle.stop();
172            let _ = self.join_worker();
173        }
174    }
175}
irtt_client/managed/blocking.rs

irtt_client/managed/
blocking.rs
