ruststream/runtime/
main_runtime.rs

1//! [`MainRuntime`]: the app's runtime as a handler reaches it, and the explicit way to send work
2//! there from a dedicated thread.
3
4use std::future::Future;
5
6use tokio::runtime::Handle;
7use tokio::task::{JoinError, JoinHandle};
8
9/// The runtime the app runs on, as a handler reaches it: `Ctx(main): Ctx<MainRuntime>` as a
10/// parameter, or [`Context::main_runtime`](super::Context::main_runtime).
11///
12/// A handler on dedicated threads (`threads(n)`) runs on its own thread, and so does everything
13/// it leaves behind: a plain `tokio::spawn`, a timer, an `and_after` continuation, an `after(..)`
14/// hook and the timer of a `retry_after` copy. `MainRuntime` is the one explicit way to send work
15/// to the app's runtime instead: [`spawn`](Self::spawn) starts it there and moves on,
16/// [`run`](Self::run) runs it there and hands its output back. What is sent must be `Send`.
17/// A plain `tokio::spawn` on a dedicated thread lives as long as that thread: a task still
18/// pending when the subscription ends is dropped with the thread's runtime.
19/// On any other placement the handler already runs on this runtime.
20///
21/// # Examples
22///
23/// ```
24/// # #[cfg(all(feature = "macros", feature = "json"))]
25/// # mod demo {
26/// use ruststream::prelude::*;
27///
28/// #[derive(serde::Deserialize)]
29/// struct Job {
30///     id: u64,
31/// }
32///
33/// async fn notify(_id: u64) {}
34///
35/// async fn lookup(id: u64) -> u64 {
36///     id * 2
37/// }
38///
39/// #[subscriber("jobs", threads(4))]
40/// async fn index(job: &Job, Ctx(main): Ctx<MainRuntime>) -> HandlerOutcome {
41///     // The notification leaves the thread; the handler does not wait for it.
42///     main.spawn(notify(job.id));
43///     // The lookup runs on the app's runtime, and its answer comes back here.
44///     match main.run(lookup(job.id)).await {
45///         Ok(_answer) => HandlerOutcome::ack(),
46///         Err(_) => HandlerOutcome::retry(),
47///     }
48/// }
49/// # }
50/// ```
51#[derive(Debug, Clone)]
52pub struct MainRuntime(Handle);
53
54impl MainRuntime {
55    /// The runtime the calling task runs on: a subscription captures it when it opens, inside
56    /// the app's startup.
57    pub(crate) fn current() -> Self {
58        Self(Handle::current())
59    }
60
61    /// Wraps `handle`.
62    #[cfg(test)]
63    pub(crate) const fn new(handle: Handle) -> Self {
64        Self(handle)
65    }
66
67    /// Starts `future` as a task of the app's runtime and returns its handle; the handler does
68    /// not wait for it unless it awaits the handle.
69    ///
70    /// # Examples
71    ///
72    /// ```
73    /// use ruststream::IncomingMessage;
74    /// use ruststream::runtime::{Context, HandlerOutcome};
75    ///
76    /// async fn notify(_id: u64) {}
77    ///
78    /// async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
79    ///     ctx.main_runtime().spawn(notify(7));
80    ///     HandlerOutcome::ack()
81    /// }
82    /// ```
83    pub fn spawn<Work>(&self, future: Work) -> JoinHandle<Work::Output>
84    where
85        Work: Future + Send + 'static,
86        Work::Output: Send + 'static,
87    {
88        self.0.spawn(future)
89    }
90
91    /// Runs `future` as a task of the app's runtime and resolves to its output.
92    ///
93    /// # Errors
94    ///
95    /// Returns the [`JoinError`] of a task that panicked, or that the runtime cancelled because
96    /// it is shutting down.
97    ///
98    /// # Cancel safety
99    ///
100    /// Dropping the returned future stops waiting, not the work: the task runs to its end on the
101    /// app's runtime.
102    ///
103    /// # Examples
104    ///
105    /// ```
106    /// use ruststream::IncomingMessage;
107    /// use ruststream::runtime::{Context, HandlerOutcome};
108    ///
109    /// async fn lookup(id: u64) -> u64 {
110    ///     id * 2
111    /// }
112    ///
113    /// async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
114    ///     match ctx.main_runtime().run(lookup(7)).await {
115    ///         Ok(14) => HandlerOutcome::ack(),
116    ///         _ => HandlerOutcome::retry(),
117    ///     }
118    /// }
119    /// ```
120    pub fn run<Work>(
121        &self,
122        future: Work,
123    ) -> impl Future<Output = Result<Work::Output, JoinError>> + Send + 'static
124    where
125        Work: Future + Send + 'static,
126        Work::Output: Send + 'static,
127    {
128        self.0.spawn(future)
129    }
130
131    /// The Tokio handle of the app's runtime, for what the two methods above do not cover
132    /// (`block_on` from a thread outside any runtime, the runtime's metrics).
133    ///
134    /// # Examples
135    ///
136    /// ```
137    /// use ruststream::IncomingMessage;
138    /// use ruststream::runtime::{Context, HandlerOutcome};
139    ///
140    /// async fn handle<M: IncomingMessage>(_msg: &M, ctx: &mut Context<'_>) -> HandlerOutcome {
141    ///     // The app's runtime metrics, read from a handler on a thread of its own.
142    ///     let metrics = ctx.main_runtime().as_handle().metrics();
143    ///     tracing::info!(
144    ///         workers = metrics.num_workers(),
145    ///         alive = metrics.num_alive_tasks(),
146    ///         "app runtime"
147    ///     );
148    ///     HandlerOutcome::ack()
149    /// }
150    /// ```
151    #[must_use]
152    pub const fn as_handle(&self) -> &Handle {
153        &self.0
154    }
155}
ruststream/runtime/main_runtime.rs

ruststream/runtime/
main_runtime.rs
