webx_core/
dispatch_runtime.rs1use std::future::Future;
10use std::sync::Arc;
11
12use crate::route::scan::HandlerCache;
13
14tokio::task_local! {
15 static ACTIVE_RUNTIME: Option<Arc<DispatchRuntime>>;
16 static ACTIVE_SCOPE_PROVIDER: Option<Arc<rust_dix::ServiceProvider>>;
22}
23
24#[derive(Clone)]
26pub struct DispatchRuntime {
27 provider: Arc<rust_dix::ServiceProvider>,
28 handler_cache: Arc<HandlerCache>,
29}
30
31impl DispatchRuntime {
32 pub fn new(provider: Arc<rust_dix::ServiceProvider>, handler_cache: Arc<HandlerCache>) -> Self {
33 Self {
34 provider,
35 handler_cache,
36 }
37 }
38
39 pub fn provider(&self) -> &Arc<rust_dix::ServiceProvider> {
40 &self.provider
41 }
42
43 pub fn handler_cache(&self) -> &Arc<HandlerCache> {
44 &self.handler_cache
45 }
46
47 pub async fn run<F, R>(&self, f: F) -> R
49 where
50 F: Future<Output = R>,
51 {
52 ACTIVE_RUNTIME.scope(Some(Arc::new(self.clone())), f).await
53 }
54
55 pub async fn run_in_scope<F, R>(self: &Arc<Self>, f: F) -> R
66 where
67 F: Future<Output = R>,
68 {
69 let scope = self.provider.scope();
70 let scoped_provider = Arc::clone(scope.scope_provider());
71 let result = ACTIVE_RUNTIME
72 .scope(
73 Some(Arc::clone(self)),
74 ACTIVE_SCOPE_PROVIDER.scope(Some(scoped_provider), f),
75 )
76 .await;
77 scope.dispose();
78 result
79 }
80}
81
82pub fn dispatch_provider() -> Arc<rust_dix::ServiceProvider> {
90 if let Ok(Some(provider)) = ACTIVE_SCOPE_PROVIDER.try_with(|p| p.clone()) {
93 return provider;
94 }
95
96 if let Ok(Some(provider)) =
97 ACTIVE_RUNTIME.try_with(|rt| rt.as_ref().map(|r| Arc::clone(&r.provider)))
98 {
99 return provider;
100 }
101
102 panic!(
103 "No active DispatchRuntime. HTTP requests and hosted services run inside \
104 Host::dispatch_runtime().run(); for in-process calls (Mediator::send, tests, \
105 background work) enter one explicitly with `runtime.run(...)` / \
106 `host.dispatch_runtime().run(...)`, or use `host.provider()` when no \
107 task-local runtime is needed."
108 );
109}
110
111pub fn dispatch_handler_cache() -> Arc<HandlerCache> {
117 if let Ok(Some(cache)) =
118 ACTIVE_RUNTIME.try_with(|rt| rt.as_ref().map(|r| Arc::clone(&r.handler_cache)))
119 {
120 return cache;
121 }
122
123 panic!(
124 "No active DispatchRuntime. Handler dispatch requires a host runtime: run \
125 `Host::dispatch_runtime().run(...)` around the call (HTTP requests and hosted \
126 services already do), or `host.dispatch_runtime()` for the host instance."
127 );
128}
129
130#[cfg(test)]
131mod tests {
132 use super::*;
133 use rust_dix::ServiceCollection;
134 use std::sync::atomic::{AtomicUsize, Ordering};
135
136 #[derive(Debug)]
137 struct ScopedThing(usize);
138
139 fn runtime_with_scoped_counter() -> (Arc<DispatchRuntime>, Arc<AtomicUsize>) {
140 let created = Arc::new(AtomicUsize::new(0));
141 let counter = Arc::clone(&created);
142 let svc = ServiceCollection::new().scoped::<ScopedThing>(move |_| {
143 let n = counter.fetch_add(1, Ordering::SeqCst);
144 Arc::new(ScopedThing(n))
145 });
146 let provider = svc.build().expect("provider");
147 let runtime = Arc::new(DispatchRuntime::new(
148 provider,
149 Arc::new(HandlerCache::build()),
150 ));
151 (runtime, created)
152 }
153
154 #[tokio::test]
157 async fn run_in_scope_gives_each_run_its_own_scoped_cache() {
158 let (runtime, created) = runtime_with_scoped_counter();
159
160 let same_within_one_scope = runtime
161 .run_in_scope(async {
162 let first = dispatch_provider().get::<ScopedThing>().expect("scoped");
163 let second = dispatch_provider().get::<ScopedThing>().expect("scoped");
164 Arc::ptr_eq(&first, &second)
165 })
166 .await;
167 assert!(
168 same_within_one_scope,
169 "scoped service must be cached per scope"
170 );
171 assert_eq!(created.load(Ordering::SeqCst), 1);
172
173 runtime
174 .run_in_scope(async {
175 let again = dispatch_provider().get::<ScopedThing>().expect("scoped");
176 assert_eq!(again.0, 1, "a new scope must not reuse the root cache");
177 })
178 .await;
179 assert_eq!(created.load(Ordering::SeqCst), 2);
180 }
181
182 #[tokio::test]
185 async fn scope_provider_does_not_leak_outside_run_in_scope() {
186 let (runtime, _created) = runtime_with_scoped_counter();
187 runtime
188 .run(async {
189 assert!(
190 dispatch_provider().get::<ScopedThing>().is_ok(),
191 "runtime provider still resolves the service"
192 );
193 })
194 .await;
195 }
196}