use std::future::Future;
use std::sync::Arc;
use crate::route::scan::HandlerCache;
tokio::task_local! {
static ACTIVE_RUNTIME: Option<Arc<DispatchRuntime>>;
static ACTIVE_SCOPE_PROVIDER: Option<Arc<rust_dix::ServiceProvider>>;
}
#[derive(Clone)]
pub struct DispatchRuntime {
provider: Arc<rust_dix::ServiceProvider>,
handler_cache: Arc<HandlerCache>,
}
impl DispatchRuntime {
pub fn new(provider: Arc<rust_dix::ServiceProvider>, handler_cache: Arc<HandlerCache>) -> Self {
Self {
provider,
handler_cache,
}
}
pub fn provider(&self) -> &Arc<rust_dix::ServiceProvider> {
&self.provider
}
pub fn handler_cache(&self) -> &Arc<HandlerCache> {
&self.handler_cache
}
pub async fn run<F, R>(&self, f: F) -> R
where
F: Future<Output = R>,
{
ACTIVE_RUNTIME.scope(Some(Arc::new(self.clone())), f).await
}
pub async fn run_in_scope<F, R>(self: &Arc<Self>, f: F) -> R
where
F: Future<Output = R>,
{
let scope = self.provider.scope();
let scoped_provider = Arc::clone(scope.scope_provider());
let result = ACTIVE_RUNTIME
.scope(
Some(Arc::clone(self)),
ACTIVE_SCOPE_PROVIDER.scope(Some(scoped_provider), f),
)
.await;
scope.dispose();
result
}
}
pub fn dispatch_provider() -> Arc<rust_dix::ServiceProvider> {
if let Ok(Some(provider)) = ACTIVE_SCOPE_PROVIDER.try_with(|p| p.clone()) {
return provider;
}
if let Ok(Some(provider)) =
ACTIVE_RUNTIME.try_with(|rt| rt.as_ref().map(|r| Arc::clone(&r.provider)))
{
return provider;
}
panic!(
"No active DispatchRuntime. HTTP requests and hosted services run inside \
Host::dispatch_runtime().run(); for in-process calls (Mediator::send, tests, \
background work) enter one explicitly with `runtime.run(...)` / \
`host.dispatch_runtime().run(...)`, or use `host.provider()` when no \
task-local runtime is needed."
);
}
pub fn dispatch_handler_cache() -> Arc<HandlerCache> {
if let Ok(Some(cache)) =
ACTIVE_RUNTIME.try_with(|rt| rt.as_ref().map(|r| Arc::clone(&r.handler_cache)))
{
return cache;
}
panic!(
"No active DispatchRuntime. Handler dispatch requires a host runtime: run \
`Host::dispatch_runtime().run(...)` around the call (HTTP requests and hosted \
services already do), or `host.dispatch_runtime()` for the host instance."
);
}
#[cfg(test)]
mod tests {
use super::*;
use rust_dix::ServiceCollection;
use std::sync::atomic::{AtomicUsize, Ordering};
#[derive(Debug)]
struct ScopedThing(usize);
fn runtime_with_scoped_counter() -> (Arc<DispatchRuntime>, Arc<AtomicUsize>) {
let created = Arc::new(AtomicUsize::new(0));
let counter = Arc::clone(&created);
let svc = ServiceCollection::new().scoped::<ScopedThing>(move |_| {
let n = counter.fetch_add(1, Ordering::SeqCst);
Arc::new(ScopedThing(n))
});
let provider = svc.build().expect("provider");
let runtime = Arc::new(DispatchRuntime::new(
provider,
Arc::new(HandlerCache::build()),
));
(runtime, created)
}
#[tokio::test]
async fn run_in_scope_gives_each_run_its_own_scoped_cache() {
let (runtime, created) = runtime_with_scoped_counter();
let same_within_one_scope = runtime
.run_in_scope(async {
let first = dispatch_provider().get::<ScopedThing>().expect("scoped");
let second = dispatch_provider().get::<ScopedThing>().expect("scoped");
Arc::ptr_eq(&first, &second)
})
.await;
assert!(
same_within_one_scope,
"scoped service must be cached per scope"
);
assert_eq!(created.load(Ordering::SeqCst), 1);
runtime
.run_in_scope(async {
let again = dispatch_provider().get::<ScopedThing>().expect("scoped");
assert_eq!(again.0, 1, "a new scope must not reuse the root cache");
})
.await;
assert_eq!(created.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn scope_provider_does_not_leak_outside_run_in_scope() {
let (runtime, _created) = runtime_with_scoped_counter();
runtime
.run(async {
assert!(
dispatch_provider().get::<ScopedThing>().is_ok(),
"runtime provider still resolves the service"
);
})
.await;
}
}