olai_http/
service.rs

1use std::fmt::Debug;
2use std::future::Future;
3use std::pin::Pin;
4use std::sync::Arc;
5
6use tokio::runtime::Handle;
7
8/// An abstraction over HTTP request execution.
9///
10/// This trait allows decoupling request building (which uses `reqwest::RequestBuilder`)
11/// from request execution. The primary use case is [`SpawnService`], which spawns
12/// execution on a dedicated I/O runtime -- useful when the CPU runtime has I/O disabled.
13pub trait HttpService: Debug + Send + Sync + 'static {
14    /// Execute an HTTP request, returning the response.
15    fn call(
16        &self,
17        request: reqwest::Request,
18    ) -> Pin<Box<dyn Future<Output = Result<reqwest::Response, reqwest::Error>> + Send + '_>>;
19}
20
21/// Default [`HttpService`] that delegates directly to [`reqwest::Client::execute`].
22#[derive(Debug, Clone)]
23pub struct ReqwestService(reqwest::Client);
24
25impl ReqwestService {
26    pub fn new(client: reqwest::Client) -> Self {
27        Self(client)
28    }
29}
30
31impl HttpService for ReqwestService {
32    fn call(
33        &self,
34        request: reqwest::Request,
35    ) -> Pin<Box<dyn Future<Output = Result<reqwest::Response, reqwest::Error>> + Send + '_>> {
36        Box::pin(self.0.execute(request))
37    }
38}
39
40/// An [`HttpService`] that spawns each request on a separate tokio runtime.
41///
42/// This is useful when the calling runtime (e.g. a CPU-bound DataFusion runtime)
43/// may have I/O disabled. All HTTP I/O -- including credential refresh -- is
44/// routed through the provided runtime handle.
45///
46/// # Example
47///
48/// ```ignore
49/// use olai_http::CloudClient;
50///
51/// let io_runtime = tokio::runtime::Runtime::new().unwrap();
52/// let client = CloudClient::new_with_token("tok")
53///     .with_runtime(io_runtime.handle().clone());
54/// ```
55#[derive(Debug)]
56pub struct SpawnService {
57    inner: Arc<dyn HttpService>,
58    handle: Handle,
59}
60
61impl SpawnService {
62    pub fn new(inner: Arc<dyn HttpService>, handle: Handle) -> Self {
63        Self { inner, handle }
64    }
65}
66
67impl HttpService for SpawnService {
68    fn call(
69        &self,
70        request: reqwest::Request,
71    ) -> Pin<Box<dyn Future<Output = Result<reqwest::Response, reqwest::Error>> + Send + '_>> {
72        let inner = Arc::clone(&self.inner);
73        let handle = self.handle.clone();
74        Box::pin(async move {
75            match handle.spawn(async move { inner.call(request).await }).await {
76                Ok(result) => result,
77                Err(join_err) => {
78                    // The spawned I/O task panicked or was cancelled. This is a
79                    // programming error (e.g. tokio runtime was shut down), not a
80                    // transient network failure, so we re-panic with context rather
81                    // than silently swallowing the cause.
82                    panic!("I/O runtime task failed: {join_err}")
83                }
84            }
85        })
86    }
87}
88
89/// Create an [`HttpService`] for the given client, optionally wrapping in [`SpawnService`].
90pub(crate) fn make_service(
91    client: reqwest::Client,
92    runtime: Option<&Handle>,
93) -> Arc<dyn HttpService> {
94    let base: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client));
95    match runtime {
96        Some(handle) => Arc::new(SpawnService::new(base, handle.clone())),
97        None => base,
98    }
99}
olai_http/service.rs

olai_http/
service.rs
