webx_core/route/
ext.rs

1//! IServiceCollectionExt — Extension trait for rust_dix ServiceCollection.
2
3use crate::handler::IHostedService;
4use crate::mediator::Mediator;
5use crate::middleware::IMiddleware;
6use crate::pipeline::IPipelineBehavior;
7use rust_dix::{IServiceResolver, ServiceCollection, ServiceLifetime};
8use std::sync::Arc;
9
10/// Extension methods for `rust_dix::ServiceCollection` to enable
11/// framework-level service registration patterns.
12pub trait IServiceCollectionExt: Sized {
13    /// Register the mediator in the DI container.
14    ///
15    /// The actual `IRequestHandler` and `IEventHandler` implementations
16    /// are registered via `#[rust_dix::module]` blocks (re-exported as `webx::module`).
17    fn add_mediator(self) -> Self;
18
19    /// Signal that endpoints should be auto-discovered from `#[get]` / `#[endpoint]`
20    /// annotations at compile time.  The actual route table is built during
21    /// `Host::build()` by scanning the inventory for `RouteEntry` items.
22    fn add_request_endpoints(self) -> Self;
23
24    /// Register controller types scoped per request.
25    fn add_controllers(self) -> Self;
26
27    /// Register a middleware type in the DI container as a singleton.
28    fn add_middleware<T>(self) -> Self
29    where
30        T: IMiddleware + Default + Send + Sync + 'static;
31
32    /// Register a **typed** pipeline behavior in the DI container as a singleton.
33    ///
34    /// `B` declares the request/response types by implementing
35    /// [`IPipelineBehavior<Req, Rsp>`]; register those same types here:
36    ///
37    /// ```ignore
38    /// #[derive(Default)]
39    /// struct AuditBehavior;
40    ///
41    /// #[async_trait::async_trait]
42    /// impl IPipelineBehavior<CreateOrderRequest, OrderView> for AuditBehavior {
43    ///     async fn handle(&self, req: CreateOrderRequest, next: Next<CreateOrderRequest, OrderView>)
44    ///         -> ActionResult<OrderView> { next.run(req).await }
45    /// }
46    ///
47    /// svc.add_typed_pipeline::<AuditBehavior, CreateOrderRequest, OrderView>();
48    /// ```
49    ///
50    /// The behavior only runs for that request/response pair. Short-circuiting is
51    /// type-checked: return `Rsp` without calling
52    /// [`Next::run`](crate::pipeline::Next::run).
53    fn add_typed_pipeline<B, Req, Rsp>(self) -> Self
54    where
55        B: IPipelineBehavior<Req, Rsp> + Default + 'static,
56        Req: Send + 'static,
57        Rsp: Send + 'static;
58
59    /// Register a typed pipeline behavior built from a factory (singleton).
60    ///
61    /// The factory receives the DI resolver at construction time, so the
62    /// behavior can take constructor dependencies. It runs once per provider;
63    /// the resolver is **not** available inside `handle`.
64    fn add_typed_pipeline_with<F, Req, Rsp>(self, factory: F) -> Self
65    where
66        F: Fn(&dyn IServiceResolver) -> Arc<dyn IPipelineBehavior<Req, Rsp>>
67            + Send
68            + Sync
69            + 'static,
70        Req: Send + 'static,
71        Rsp: Send + 'static;
72
73    /// Register a hosted service in the DI container as a singleton.
74    ///
75    /// Hosted services are started when the host starts (before accepting
76    /// requests) and stopped during graceful shutdown.
77    ///
78    /// Analogous to ASP.NET Core's `AddHostedService<T>()`.
79    ///
80    /// # Example
81    ///
82    /// ```ignore
83    /// Host::builder()
84    ///     .register(|svc| {
85    ///         svc.add_hosted_service::<DbInitService>()
86    ///     })
87    ///     .build()
88    ///     .run()
89    ///     .await
90    /// ```
91    fn add_hosted_service<T>(self) -> Self
92    where
93        T: IHostedService + Default + Send + Sync + 'static;
94
95    /// Register a hosted service built by `factory` (singleton).
96    ///
97    /// Unlike [`add_hosted_service`](IServiceCollectionExt::add_hosted_service)
98    /// this does **not** require `Default`: the factory receives the DI resolver
99    /// when the service is constructed, so configuration and constructor
100    /// dependencies are injected normally.
101    ///
102    /// Analogous to ASP.NET Core's
103    /// `AddHostedService(sp => new T(sp.GetRequiredService<...>()))`.
104    ///
105    /// ```ignore
106    /// svc
107    ///     .add_options::<OutboxOptions>("Outbox")
108    ///     .add_hosted_service_with(|resolver| {
109    ///         let options = resolve::<OutboxOptions>(resolver);
110    ///         Arc::new(OutboxConsumer::new(options))
111    ///     })
112    /// ```
113    fn add_hosted_service_with<T, F>(self, factory: F) -> Self
114    where
115        T: IHostedService + 'static,
116        F: Fn(&dyn IServiceResolver) -> Arc<T> + Send + Sync + 'static;
117
118    /// Register a hosted service with an explicit start/stop order.
119    ///
120    /// Lower `order` starts first and stops last; equal orders keep the
121    /// registration order. See [`IHostedService::order`].
122    fn add_hosted_service_ordered<T>(self, order: i32) -> Self
123    where
124        T: IHostedService + Default + Send + Sync + 'static;
125
126    /// Register a factory-built hosted service with an explicit start/stop order.
127    fn add_hosted_service_with_ordered<T, F>(self, order: i32, factory: F) -> Self
128    where
129        T: IHostedService + 'static,
130        F: Fn(&dyn IServiceResolver) -> Arc<T> + Send + Sync + 'static;
131
132    /// Register a hosted service only when `enabled` is true.
133    ///
134    /// Conditions belong to configuration (`if options.Outbox.Enabled`), so a
135    /// deployment can switch a migration/consumer off without a code change:
136    ///
137    /// ```ignore
138    /// svc.add_hosted_service_if::<SeedService>(options.seed.enabled)
139    /// ```
140    fn add_hosted_service_if<T>(self, enabled: bool) -> Self
141    where
142        T: IHostedService + Default + Send + Sync + 'static;
143
144    /// Register a factory-built hosted service only when `enabled` is true.
145    fn add_hosted_service_with_if<T, F>(self, enabled: bool, factory: F) -> Self
146    where
147        T: IHostedService + 'static,
148        F: Fn(&dyn IServiceResolver) -> Arc<T> + Send + Sync + 'static;
149
150    /// Register a [`BackgroundService`](crate::hosted::BackgroundService)
151    /// worker (singleton, `Default` constructor).
152    ///
153    /// The host starts the worker without blocking startup and cancels its
154    /// [`CancellationToken`](crate::hosted::CancellationToken) on shutdown;
155    /// business code only implements `execute`.
156    fn add_background_service<B>(self) -> Self
157    where
158        B: crate::hosted::BackgroundService + Default;
159
160    /// Register a factory-built [`BackgroundService`](crate::hosted::BackgroundService)
161    /// worker (singleton).
162    ///
163    /// The factory receives the DI resolver at construction time, so the worker
164    /// can take configuration and `Arc<dyn ScopeFactory>` for its per-tick
165    /// scopes — the pattern that used to require a hand-rolled
166    /// `start`/`stop`/`JoinHandle` hosted service.
167    fn add_background_service_with<B, F>(self, factory: F) -> Self
168    where
169        B: crate::hosted::BackgroundService,
170        F: Fn(&dyn IServiceResolver) -> Arc<B> + Send + Sync + 'static;
171
172    /// Register a factory-built background worker only when `enabled` is true.
173    fn add_background_service_with_if<B, F>(self, enabled: bool, factory: F) -> Self
174    where
175        B: crate::hosted::BackgroundService,
176        F: Fn(&dyn IServiceResolver) -> Arc<B> + Send + Sync + 'static;
177}
178
179impl IServiceCollectionExt for ServiceCollection {
180    fn add_mediator(self) -> Self {
181        // Transient (not singleton) so build-time validation does not require a
182        // fully wired ServiceProvider. Each resolution shares the root provider
183        // Arc; the handler registry is registered by `Host::build` as an
184        // instance (or rebuilt when the mediator is used without a host).
185        MEDIATOR_ACTIVE.store(true, std::sync::atomic::Ordering::SeqCst);
186        self.transient::<Mediator>(|resolver| {
187            let provider = resolver
188                .provider_arc()
189                .expect("ServiceProvider not available from DI resolver");
190            let handler_cache = resolver
191                .get_any(std::any::type_name::<Arc<crate::route::scan::HandlerCache>>())
192                .and_then(|service| {
193                    service
194                        .downcast::<Arc<crate::route::scan::HandlerCache>>()
195                        .ok()
196                })
197                .map(|service| Arc::clone(&*service))
198                .unwrap_or_else(|| Arc::new(crate::route::scan::HandlerCache::build()));
199            Arc::new(Mediator::new(provider, handler_cache))
200        })
201    }
202
203    fn add_request_endpoints(self) -> Self {
204        // Routes are collected from inventory at Host::build() time.
205        // This call just signals that auto-scanning is desired.
206        ENDPOINT_SCAN.store(true, std::sync::atomic::Ordering::SeqCst);
207        self
208    }
209
210    fn add_controllers(self) -> Self {
211        self
212    }
213
214    fn add_middleware<T>(self) -> Self
215    where
216        T: IMiddleware + Default + Send + Sync + 'static,
217    {
218        self.singleton::<dyn IMiddleware>(|_| Arc::new(T::default()))
219    }
220
221    fn add_typed_pipeline<B, Req, Rsp>(self) -> Self
222    where
223        B: IPipelineBehavior<Req, Rsp> + Default + 'static,
224        Req: Send + 'static,
225        Rsp: Send + 'static,
226    {
227        self.add_typed_pipeline_with::<_, Req, Rsp>(|_| {
228            Arc::new(B::default()) as Arc<dyn IPipelineBehavior<Req, Rsp>>
229        })
230    }
231
232    fn add_typed_pipeline_with<F, Req, Rsp>(self, factory: F) -> Self
233    where
234        F: Fn(&dyn IServiceResolver) -> Arc<dyn IPipelineBehavior<Req, Rsp>>
235            + Send
236            + Sync
237            + 'static,
238        Req: Send + 'static,
239        Rsp: Send + 'static,
240    {
241        // Singleton per (Req, Rsp) pair: the factory (and therefore constructor
242        // injection) runs once per provider; behaviors are stateless wrappers.
243        self.add(ServiceLifetime::Singleton, factory)
244    }
245
246    fn add_hosted_service<T>(self) -> Self
247    where
248        T: IHostedService + Default + Send + Sync + 'static,
249    {
250        self.singleton::<dyn IHostedService>(|_| Arc::new(T::default()))
251    }
252
253    fn add_hosted_service_with<T, F>(self, factory: F) -> Self
254    where
255        T: IHostedService + 'static,
256        F: Fn(&dyn IServiceResolver) -> Arc<T> + Send + Sync + 'static,
257    {
258        self.add(
259            ServiceLifetime::Singleton,
260            move |resolver: &dyn IServiceResolver| -> Arc<dyn IHostedService> { factory(resolver) },
261        )
262    }
263
264    fn add_hosted_service_ordered<T>(self, order: i32) -> Self
265    where
266        T: IHostedService + Default + Send + Sync + 'static,
267    {
268        self.add_hosted_service_with_ordered(order, |_| Arc::new(T::default()))
269    }
270
271    fn add_hosted_service_with_ordered<T, F>(self, order: i32, factory: F) -> Self
272    where
273        T: IHostedService + 'static,
274        F: Fn(&dyn IServiceResolver) -> Arc<T> + Send + Sync + 'static,
275    {
276        self.add(
277            ServiceLifetime::Singleton,
278            move |resolver: &dyn IServiceResolver| -> Arc<dyn IHostedService> {
279                Arc::new(crate::hosted::OrderedHostedService::new(
280                    order,
281                    factory(resolver),
282                ))
283            },
284        )
285    }
286
287    fn add_hosted_service_if<T>(self, enabled: bool) -> Self
288    where
289        T: IHostedService + Default + Send + Sync + 'static,
290    {
291        if enabled {
292            self.add_hosted_service::<T>()
293        } else {
294            self
295        }
296    }
297
298    fn add_hosted_service_with_if<T, F>(self, enabled: bool, factory: F) -> Self
299    where
300        T: IHostedService + 'static,
301        F: Fn(&dyn IServiceResolver) -> Arc<T> + Send + Sync + 'static,
302    {
303        if enabled {
304            self.add_hosted_service_with(factory)
305        } else {
306            self
307        }
308    }
309
310    fn add_background_service<B>(self) -> Self
311    where
312        B: crate::hosted::BackgroundService + Default,
313    {
314        self.add_background_service_with(|_| Arc::new(B::default()))
315    }
316
317    fn add_background_service_with<B, F>(self, factory: F) -> Self
318    where
319        B: crate::hosted::BackgroundService,
320        F: Fn(&dyn IServiceResolver) -> Arc<B> + Send + Sync + 'static,
321    {
322        self.add(
323            ServiceLifetime::Singleton,
324            move |resolver: &dyn IServiceResolver| -> Arc<dyn IHostedService> {
325                Arc::new(crate::hosted::BackgroundServiceHost::new(factory(resolver)))
326            },
327        )
328    }
329
330    fn add_background_service_with_if<B, F>(self, enabled: bool, factory: F) -> Self
331    where
332        B: crate::hosted::BackgroundService,
333        F: Fn(&dyn IServiceResolver) -> Arc<B> + Send + Sync + 'static,
334    {
335        if enabled {
336            self.add_background_service_with(factory)
337        } else {
338            self
339        }
340    }
341}
342
343// Global flags so HostBuilder knows what to scan at build time.
344use std::sync::atomic::AtomicBool;
345static ENDPOINT_SCAN: AtomicBool = AtomicBool::new(false);
346static MEDIATOR_ACTIVE: AtomicBool = AtomicBool::new(false);
347
348pub fn should_scan_endpoints() -> bool {
349    ENDPOINT_SCAN.load(std::sync::atomic::Ordering::SeqCst)
350}
351
352pub fn is_mediator_active() -> bool {
353    MEDIATOR_ACTIVE.load(std::sync::atomic::Ordering::SeqCst)
354}
355
356#[cfg(test)]
357mod tests {
358    use super::*;
359    use crate::error::ActionResult;
360    use crate::pipeline::{IPipelineBehavior, Next};
361    use std::sync::atomic::{AtomicU32, Ordering};
362
363    /// Resolve a registered singleton from the type-erased resolver the
364    /// registration factories receive.
365    fn resolve<T: Send + Sync + 'static>(resolver: &dyn IServiceResolver) -> Arc<T> {
366        resolver
367            .get_any(std::any::type_name::<T>())
368            .and_then(|service| service.downcast::<Arc<T>>().ok())
369            .map(|service| Arc::clone(&*service))
370            .expect("the dependency must be registered before the pipeline")
371    }
372
373    // ----- typed behaviors (0.10) -----
374
375    #[derive(Default)]
376    struct Uppercase;
377
378    #[async_trait::async_trait]
379    impl IPipelineBehavior<u32, String> for Uppercase {
380        async fn handle(&self, req: u32, next: Next<u32, String>) -> ActionResult<String> {
381            let inner = next.run(req + 1).await?;
382            Ok(inner.to_uppercase())
383        }
384    }
385
386    fn string_terminal() -> Next<u32, String> {
387        Next::new(|n: u32| async move { Ok::<String, crate::error::Error>(format!("n={n}")) })
388    }
389
390    #[tokio::test]
391    async fn typed_behavior_registers_only_for_its_request_response_pair() {
392        let provider = ServiceCollection::new()
393            .add_typed_pipeline::<Uppercase, u32, String>()
394            .build()
395            .expect("provider must build");
396
397        let behaviors = provider.get_all::<dyn IPipelineBehavior<u32, String>>();
398        assert_eq!(behaviors.len(), 1);
399        let out = behaviors[0]
400            .handle(7, string_terminal())
401            .await
402            .expect("typed behavior must run");
403        assert_eq!(out, "N=8");
404
405        assert!(
406            provider
407                .get_all::<dyn IPipelineBehavior<u64, String>>()
408                .is_empty(),
409            "a typed behavior must only attach to its own request/response pair"
410        );
411    }
412
413    struct TickBehavior {
414        ticks: Arc<AtomicU32>,
415    }
416
417    #[async_trait::async_trait]
418    impl IPipelineBehavior<u32, u32> for TickBehavior {
419        async fn handle(&self, req: u32, next: Next<u32, u32>) -> ActionResult<u32> {
420            self.ticks.fetch_add(1, Ordering::SeqCst);
421            next.run(req).await
422        }
423    }
424
425    #[tokio::test]
426    async fn typed_add_pipeline_with_constructor_injects_from_the_resolver() {
427        let ticks = Arc::new(AtomicU32::new(0));
428
429        let provider = ServiceCollection::new()
430            .instance(Arc::clone(&ticks))
431            .add_typed_pipeline_with(|resolver| {
432                let ticks: Arc<AtomicU32> = resolve::<AtomicU32>(resolver);
433                Arc::new(TickBehavior { ticks }) as Arc<dyn IPipelineBehavior<u32, u32>>
434            })
435            .build()
436            .expect("provider must build");
437
438        let behavior = &provider.get_all::<dyn IPipelineBehavior<u32, u32>>()[0];
439        let out = behavior
440            .handle(5, Next::new(|n: u32| async move { Ok(n * 2) }))
441            .await
442            .expect("typed behavior must run");
443        assert_eq!(out, 10);
444        assert_eq!(ticks.load(Ordering::SeqCst), 1);
445    }
446
447    struct ShortCircuit;
448
449    #[async_trait::async_trait]
450    impl IPipelineBehavior<u32, String> for ShortCircuit {
451        async fn handle(&self, _req: u32, _next: Next<u32, String>) -> ActionResult<String> {
452            Ok("short".to_string())
453        }
454    }
455
456    #[tokio::test]
457    async fn try_add_keeps_the_existing_registration() {
458        let provider = ServiceCollection::new()
459            .instance(Arc::new(1u32))
460            .try_add(|_| Arc::new(2u32))
461            .build()
462            .expect("provider must build");
463
464        let value = provider.get::<u32>().expect("registered");
465        assert_eq!(
466            *value, 1,
467            "try_add must not replace an earlier registration (MEDI TryAdd); rust-dix 0.8+ resolves the last registration otherwise"
468        );
469    }
470
471    #[tokio::test]
472    async fn typed_behavior_may_short_circuit_with_its_response_type() {
473        let out = ShortCircuit
474            .handle(1, string_terminal())
475            .await
476            .expect("short-circuit must return the typed response");
477        assert_eq!(out, "short");
478    }
479}
webx_core/route/ext.rs

webx_core/route/
ext.rs
