rig_http/
runtime.rs

1//! The tokio runtime the bundled transports poll their I/O in.
2//!
3//! [`bind`] and [`bind_stream`] enter a runtime handle on every poll, so a
4//! caller on another executor (Bevy task pools, smol, `futures::executor`)
5//! owns and polls the future or stream itself: no spawned task, no channel.
6//! The handle is the current runtime's, else a process-wide fallback runtime
7//! started on first use (one worker thread, I/O and timers enabled).
8//!
9//! ```no_run
10//! let handle = rig_http::runtime::handle()?;
11//! assert_eq!(handle.block_on(async { 6 * 7 }), 42);
12//! # Ok::<(), rig_http::http_client::Error>(())
13//! ```
14
15use crate::http_client::Error;
16use futures::{Stream, stream};
17use std::future::{Future, poll_fn};
18use std::pin::pin;
19use std::sync::LazyLock;
20use tokio::runtime::{Handle, Runtime};
21
22static RUNTIME: LazyLock<Result<Runtime, RuntimeUnavailable>> = LazyLock::new(|| {
23    tokio::runtime::Builder::new_multi_thread()
24        .worker_threads(1)
25        .thread_name("rig-http")
26        .enable_all()
27        .build()
28        .map_err(|err| RuntimeUnavailable(err.to_string()))
29});
30
31#[derive(Debug, Clone, thiserror::Error)]
32#[error("rig-http: failed to start the fallback tokio runtime: {0}")]
33struct RuntimeUnavailable(String);
34
35/// The current runtime when called inside one, else the fallback runtime,
36/// which runs for the rest of the process. A host with no tokio runtime of
37/// its own hands this to tokio-based libraries instead of starting another.
38/// A supplied current runtime must have I/O and timers enabled and stay
39/// driven until its operations finish.
40///
41/// # Errors
42///
43/// When there is no current runtime and the fallback could not start.
44pub fn handle() -> Result<Handle, Error> {
45    match Handle::try_current() {
46        Ok(handle) => Ok(handle),
47        Err(_) => RUNTIME
48            .as_ref()
49            .map(|runtime| runtime.handle().clone())
50            .map_err(|error| Error::instance(error.clone())),
51    }
52}
53
54/// `future`, polled with [`handle`] entered.
55///
56/// # Errors
57///
58/// As [`handle`].
59pub fn bind<F: Future>(future: F) -> Result<impl Future<Output = F::Output>, Error> {
60    let handle = handle()?;
61    Ok(async move {
62        let mut future = pin!(future);
63        poll_fn(|cx| {
64            let _entered = handle.enter();
65            future.as_mut().poll(cx)
66        })
67        .await
68    })
69}
70
71/// `stream`, polled with [`handle`] entered.
72///
73/// # Errors
74///
75/// As [`handle`].
76pub fn bind_stream<S: Stream>(stream: S) -> Result<impl Stream<Item = S::Item>, Error> {
77    let handle = handle()?;
78    let mut stream = Box::pin(stream);
79    Ok(stream::poll_fn(move |cx| {
80        let _entered = handle.enter();
81        stream.as_mut().poll_next(cx)
82    }))
83}
84
85#[cfg(test)]
86mod tests;
rig_http/runtime.rs

rig_http/
runtime.rs
