1use crate::{
2 BoxFuture, Container, Controller, EventHandler, EventHandlerDef, Injectable, MessageHandlerDef,
3 MessageHandlerKind, Microservice, MicroserviceDef, RegisterableEventHandler, Result,
4 WebSocketGateway,
5};
6use std::{
7 any::{Any, TypeId},
8 collections::{HashMap, HashSet},
9 fmt::Debug,
10 future::Future,
11 sync::Arc,
12 time::Instant,
13};
14
15type ProviderValue = Arc<dyn Any + Send + Sync>;
16type BuildProviderFn =
17 Box<dyn for<'a> Fn(&'a Container) -> BoxFuture<'a, crate::Result<ProviderValue>> + Send + Sync>;
18type LifecycleFn =
19 Box<dyn for<'a> Fn(&'a ProviderValue) -> BoxFuture<'a, crate::Result<()>> + Send + Sync>;
20
21#[derive(Clone, Copy)]
23pub struct ProviderDependency {
25 type_id: TypeId,
26 type_name: &'static str,
27}
28
29impl ProviderDependency {
30 pub fn of<T: Send + Sync + 'static>() -> Self {
32 Self {
33 type_id: TypeId::of::<T>(),
34 type_name: std::any::type_name::<T>(),
35 }
36 }
37
38 pub(crate) fn type_id(&self) -> TypeId {
39 self.type_id
40 }
41}
42
43#[macro_export]
51macro_rules! provider_dependencies {
52 ($($dependency:ty),* $(,)?) => {
53 vec![$($crate::ProviderDependency::of::<$dependency>()),*]
54 };
55}
56
57pub struct ProviderDef {
59 type_id: TypeId,
60 type_name: &'static str,
61 dependencies: Vec<ProviderDependency>,
62 build: BuildProviderFn,
63 init_fn: LifecycleFn,
64 bootstrap_fn: LifecycleFn,
65 shutdown_fn: LifecycleFn,
66}
67
68impl ProviderDef {
69 pub fn of<T: Injectable>() -> Self {
71 Self {
72 type_id: TypeId::of::<T>(),
73 type_name: std::any::type_name::<T>(),
74 dependencies: T::dependencies(),
75 build: Box::new(|container| {
76 Box::pin(async move { Ok(Arc::new(T::create(container).await?) as ProviderValue) })
77 }),
78 init_fn: Box::new(|value| {
79 let value = downcast_provider::<T>(value);
80 Box::pin(async move { value?.on_module_init().await })
81 }),
82 bootstrap_fn: Box::new(|value| {
83 let value = downcast_provider::<T>(value);
84 Box::pin(async move { value?.on_bootstrap().await })
85 }),
86 shutdown_fn: Box::new(|value| {
87 let value = downcast_provider::<T>(value);
88 Box::pin(async move { value?.on_shutdown().await })
89 }),
90 }
91 }
92
93 pub fn instance<T: Send + Sync + 'static>(value: T) -> Self {
96 let value = Arc::new(value) as ProviderValue;
97 Self {
98 type_id: TypeId::of::<T>(),
99 type_name: std::any::type_name::<T>(),
100 dependencies: vec![],
101 build: Box::new(move |_| {
102 let value = value.clone();
103 Box::pin(async move { Ok(value) })
104 }),
105 init_fn: noop_lifecycle(),
106 bootstrap_fn: noop_lifecycle(),
107 shutdown_fn: noop_lifecycle(),
108 }
109 }
110
111 pub fn async_factory<T, Fut, E>(
113 dependencies: Vec<ProviderDependency>,
114 factory: impl Fn(Arc<Container>) -> Fut + Send + Sync + 'static,
115 ) -> Self
116 where
117 T: Send + Sync + 'static,
118 Fut: Future<Output = std::result::Result<T, E>> + Send + 'static,
119 E: Debug + Send + 'static,
120 {
121 Self {
122 type_id: TypeId::of::<T>(),
123 type_name: std::any::type_name::<T>(),
124 dependencies,
125 build: Box::new(move |container| {
126 let future = factory(Arc::new(container.clone()));
127 Box::pin(async move {
128 let value = future.await.map_err(|err| {
129 crate::exception::startup_error(format!(
130 "async factory failed for {}: {:?}",
131 std::any::type_name::<T>(),
132 err
133 ))
134 })?;
135 Ok(Arc::new(value) as ProviderValue)
136 })
137 }),
138 init_fn: noop_lifecycle(),
139 bootstrap_fn: noop_lifecycle(),
140 shutdown_fn: noop_lifecycle(),
141 }
142 }
143
144 pub fn type_id(&self) -> TypeId {
146 self.type_id
147 }
148 pub fn type_name(&self) -> &'static str {
150 self.type_name
151 }
152
153 fn assert_registered(&self, container: &Container) -> crate::Result<()> {
154 if container.contains_type_id(self.type_id) {
155 return Ok(());
156 }
157 Err(crate::exception::startup_error(format!(
158 "missing provider at startup: {} was declared by module metadata but was not registered",
159 self.type_name
160 )))
161 }
162
163 async fn run_lifecycle(
164 &self,
165 value: &ProviderValue,
166 hook: &'static str,
167 callback: &LifecycleFn,
168 ) -> crate::Result<()> {
169 callback(value).await.map_err(|err| {
170 crate::exception::startup_error(format!(
171 "{hook} failed for {}: {}: {}",
172 self.type_name, err.error, err.message
173 ))
174 })
175 }
176
177 async fn run_lifecycle_from_container(
178 &self,
179 container: &Container,
180 hook: &'static str,
181 callback: &LifecycleFn,
182 ) -> crate::Result<()> {
183 let value = container.resolve_erased(self.type_id).ok_or_else(|| crate::exception::startup_error(format!(
184 "missing provider during {hook}: {} was declared by module metadata but was not registered", self.type_name
185 )))?;
186 self.run_lifecycle(&value, hook, callback).await
187 }
188}
189
190pub struct ProviderOverrides {
192 defs: HashMap<TypeId, ProviderDef>,
193}
194impl ProviderOverrides {
195 pub fn new() -> Self {
197 Self {
198 defs: HashMap::new(),
199 }
200 }
201 pub fn insert_instance<T: Send + Sync + 'static>(mut self, value: T) -> Self {
203 self.defs
204 .insert(TypeId::of::<T>(), ProviderDef::instance(value));
205 self
206 }
207 pub fn insert_factory<T, Fut, E>(
209 mut self,
210 dependencies: Vec<ProviderDependency>,
211 factory: impl Fn(Arc<Container>) -> Fut + Send + Sync + 'static,
212 ) -> Self
213 where
214 T: Send + Sync + 'static,
215 Fut: Future<Output = std::result::Result<T, E>> + Send + 'static,
216 E: Debug + Send + 'static,
217 {
218 self.defs.insert(
219 TypeId::of::<T>(),
220 ProviderDef::async_factory::<T, Fut, E>(dependencies, factory),
221 );
222 self
223 }
224 pub fn insert(mut self, def: ProviderDef) -> Self {
226 self.defs.insert(def.type_id, def);
227 self
228 }
229 pub(crate) fn into_inner(self) -> HashMap<TypeId, ProviderDef> {
230 self.defs
231 }
232}
233impl Default for ProviderOverrides {
234 fn default() -> Self {
235 Self::new()
236 }
237}
238
239fn downcast_provider<T: Send + Sync + 'static>(value: &ProviderValue) -> crate::Result<Arc<T>> {
240 value.clone().downcast::<T>().map_err(|_| {
241 crate::exception::startup_error(format!(
242 "type mismatch running lifecycle hook for {}",
243 std::any::type_name::<T>()
244 ))
245 })
246}
247fn noop_lifecycle() -> LifecycleFn {
248 Box::new(|_| Box::pin(async { Ok(()) }))
249}
250
251pub struct ControllerDef {
253 pub register_fn: fn(&mut dyn Any),
255 #[doc(hidden)]
257 pub register_with_container_fn: fn(&mut dyn Any, Arc<Container>),
258 pub route_log_fn: fn(),
260 pub(crate) route_count_fn: fn() -> usize,
261 validate_routes_fn: fn() -> Result<()>,
262 #[cfg(feature = "openapi")]
263 pub(crate) openapi_routes_fn: fn() -> &'static [crate::openapi::OpenApiRouteDef],
264 provider: ProviderDef,
265}
266impl ControllerDef {
267 pub fn of<C: Controller + Injectable + 'static>() -> Self {
269 let mut provider = ProviderDef::of::<C>();
270 for dependency in C::route_dependencies() {
271 if !provider
272 .dependencies
273 .iter()
274 .any(|existing| existing.type_id == dependency.type_id)
275 {
276 provider.dependencies.push(dependency);
277 }
278 }
279 Self {
280 register_fn: |any| C::register_routes(any),
281 register_with_container_fn: |any, container| {
282 C::register_routes_with_container(any, container)
283 },
284 route_log_fn: || crate::log_controller_routes::<C>(),
285 route_count_fn: || C::routes().len(),
286 validate_routes_fn: C::validate_routes,
287 #[cfg(feature = "openapi")]
288 openapi_routes_fn: || C::openapi_routes(),
289 provider,
290 }
291 }
292}
293
294pub struct GatewayDef {
296 pub path: &'static str,
298 pub type_id: TypeId,
300 provider: ProviderDef,
301 kind: GatewayKind,
302}
303enum GatewayKind {
304 WebSocket {
305 resolve_fn: fn(&Container) -> crate::Result<Arc<dyn WebSocketGateway>>,
306 },
307 SocketIo {
308 register_fn: fn(&Container, &dyn Any) -> Result<()>,
309 },
310}
311impl GatewayDef {
312 pub fn websocket<G: WebSocketGateway>(path: &'static str) -> Self {
314 Self {
315 path,
316 type_id: TypeId::of::<G>(),
317 provider: ProviderDef::of::<G>(),
318 kind: GatewayKind::WebSocket {
319 resolve_fn: |c| Ok(c.resolve::<G>()? as Arc<dyn WebSocketGateway>),
320 },
321 }
322 }
323 #[doc(hidden)]
324 pub fn socket_io<G: Injectable>(
325 path: &'static str,
326 register_fn: fn(&Container, &dyn Any) -> crate::Result<()>,
327 ) -> Self {
328 Self {
329 path,
330 type_id: TypeId::of::<G>(),
331 provider: ProviderDef::of::<G>(),
332 kind: GatewayKind::SocketIo { register_fn },
333 }
334 }
335 pub fn resolve(&self, container: &Container) -> Result<Arc<dyn WebSocketGateway>> {
337 match self.kind {
338 GatewayKind::WebSocket { resolve_fn } => resolve_fn(container),
339 GatewayKind::SocketIo { .. } => Err(crate::exception::startup_error(format!(
340 "Socket.IO gateway {} cannot be mounted by an RFC 6455 application",
341 self.path
342 ))),
343 }
344 }
345 #[doc(hidden)]
346 pub fn is_websocket(&self) -> bool {
347 matches!(self.kind, GatewayKind::WebSocket { .. })
348 }
349 #[doc(hidden)]
350 pub fn register_socket_io(&self, container: &Container, handle: &dyn Any) -> Result<()> {
351 match self.kind {
352 GatewayKind::WebSocket { .. } => Ok(()),
353 GatewayKind::SocketIo { register_fn } => register_fn(container, handle),
354 }
355 }
356}
357pub trait Gateway: Injectable {
359 #[doc(hidden)]
360 fn definition() -> GatewayDef;
361}
362
363pub struct ModuleDef {
365 pub(crate) type_id: TypeId,
366 type_name: &'static str,
367 pub(crate) metadata_fn: fn() -> ModuleMetadata,
368}
369impl ModuleDef {
370 pub fn of<M: Module + 'static>() -> Self {
372 Self {
373 type_id: TypeId::of::<M>(),
374 type_name: std::any::type_name::<M>(),
375 metadata_fn: M::register,
376 }
377 }
378}
379pub trait Module {
381 fn register() -> ModuleMetadata;
383}
384
385pub struct ModuleMetadata {
387 pub imports: Vec<ModuleDef>,
389 pub providers: Vec<ProviderDef>,
391 pub controllers: Vec<ControllerDef>,
393 pub event_handlers: Vec<EventHandlerDef>,
395 pub gateways: Vec<GatewayDef>,
397 pub microservices: Vec<MicroserviceDef>,
399 exports: Vec<ProviderDependency>,
400 global: bool,
401}
402impl ModuleMetadata {
403 pub fn new() -> Self {
405 Self {
406 imports: vec![],
407 providers: vec![],
408 controllers: vec![],
409 event_handlers: vec![],
410 gateways: vec![],
411 microservices: vec![],
412 exports: vec![],
413 global: false,
414 }
415 }
416 pub fn global() -> Self {
418 let mut metadata = Self::new();
419 metadata.global = true;
420 metadata
421 }
422 pub fn import<M: Module + 'static>(mut self) -> Self {
424 self.imports.push(ModuleDef::of::<M>());
425 self
426 }
427 pub fn provider<T: Injectable>(mut self) -> Self {
429 self.providers.push(ProviderDef::of::<T>());
430 self
431 }
432 pub fn provider_async_factory<T, Fut, E>(
434 mut self,
435 dependencies: Vec<ProviderDependency>,
436 factory: impl Fn(Arc<Container>) -> Fut + Send + Sync + 'static,
437 ) -> Self
438 where
439 T: Send + Sync + 'static,
440 Fut: Future<Output = std::result::Result<T, E>> + Send + 'static,
441 E: Debug + Send + 'static,
442 {
443 self.providers.push(ProviderDef::async_factory::<T, Fut, E>(
444 dependencies,
445 factory,
446 ));
447 self
448 }
449 pub fn controller<C: Controller + Injectable + 'static>(mut self) -> Self {
451 self.controllers.push(ControllerDef::of::<C>());
452 self
453 }
454 pub fn gateway<G: Gateway>(mut self) -> Self {
456 self.gateways.push(G::definition());
457 self
458 }
459 pub fn microservice<T: Microservice>(mut self) -> Self {
464 self.microservices.push(T::definition());
465 self
466 }
467 pub fn event_handler<H>(mut self) -> Self
469 where
470 H: RegisterableEventHandler + EventHandler<H::Event>,
471 {
472 self.event_handlers.push(EventHandlerDef::of::<H>());
473 self
474 }
475 pub fn event_handler_for<E, H>(mut self) -> Self
477 where
478 E: Clone + Send + Sync + 'static,
479 H: Injectable + EventHandler<E>,
480 {
481 self.event_handlers
482 .push(EventHandlerDef::for_event::<E, H>());
483 self
484 }
485 pub fn export<T: Send + Sync + 'static>(mut self) -> Self {
487 self.exports.push(ProviderDependency::of::<T>());
488 self
489 }
490}
491impl Default for ModuleMetadata {
492 fn default() -> Self {
493 Self::new()
494 }
495}
496
497struct ModuleNode {
498 type_id: TypeId,
499 type_name: &'static str,
500 metadata: ModuleMetadata,
501 imports: Vec<usize>,
502}
503struct ModuleGraph {
504 nodes: Vec<ModuleNode>,
505}
506#[derive(Clone, Copy)]
507enum ProviderSlot {
508 Provider(usize),
509 Controller(usize),
510 Gateway(usize),
511 Microservice(usize),
512}
513#[derive(Clone, Copy)]
514struct ProviderRegistration {
515 module: usize,
516 slot: ProviderSlot,
517}
518
519impl ModuleGraph {
520 fn discover<M: Module + 'static>() -> crate::Result<Self> {
521 Self::discover_from(ModuleDef::of::<M>())
522 }
523 fn discover_from(root: ModuleDef) -> crate::Result<Self> {
524 fn visit(
525 def: ModuleDef,
526 graph: &mut Vec<ModuleNode>,
527 states: &mut HashMap<TypeId, u8>,
528 path: &mut Vec<&'static str>,
529 ) -> crate::Result<usize> {
530 match states.get(&def.type_id).copied() {
531 Some(2) => {
532 return Ok(graph
533 .iter()
534 .position(|node| node.type_id == def.type_id)
535 .expect("discovered module missing"));
536 }
537 Some(1) => {
538 let mut cycle = path.clone();
539 cycle.push(def.type_name);
540 return Err(crate::exception::startup_error(format!(
541 "circular module import: {}",
542 cycle.join(" -> ")
543 )));
544 }
545 _ => {}
546 }
547 states.insert(def.type_id, 1);
548 path.push(def.type_name);
549 let metadata = (def.metadata_fn)();
550 for controller in &metadata.controllers {
551 (controller.validate_routes_fn)()?;
552 }
553 let index = graph.len();
554 graph.push(ModuleNode {
555 type_id: def.type_id,
556 type_name: def.type_name,
557 metadata,
558 imports: vec![],
559 });
560 let imports = std::mem::take(&mut graph[index].metadata.imports);
561 for import in imports {
562 let child = visit(import, graph, states, path)?;
563 graph[index].imports.push(child);
564 }
565 path.pop();
566 states.insert(def.type_id, 2);
567 Ok(index)
568 }
569 let mut nodes = vec![];
570 let mut states = HashMap::new();
571 let root_index = visit(root, &mut nodes, &mut states, &mut vec![])?;
572 let mut ordered = Vec::new();
574 let mut seen = HashSet::new();
575 fn order(
576 index: usize,
577 nodes: &Vec<ModuleNode>,
578 seen: &mut HashSet<usize>,
579 ordered: &mut Vec<usize>,
580 ) {
581 if !seen.insert(index) {
582 return;
583 }
584 for &child in &nodes[index].imports {
585 order(child, nodes, seen, ordered);
586 }
587 ordered.push(index);
588 }
589 order(root_index, &nodes, &mut seen, &mut ordered);
590 let mut remap = HashMap::new();
591 for (new, old) in ordered.iter().enumerate() {
592 remap.insert(*old, new);
593 }
594 let mut slots: Vec<Option<ModuleNode>> = nodes.into_iter().map(Some).collect();
595 let mut result = Vec::with_capacity(slots.len());
596 for old in ordered {
597 let mut node = slots[old]
598 .take()
599 .expect("module graph ordering repeated a node");
600 node.imports = node.imports.into_iter().map(|i| remap[&i]).collect();
601 result.push(node);
602 }
603 let _ = remap[&root_index];
604 Ok(Self { nodes: result })
605 }
606
607 fn definitions(&self) -> Vec<ProviderRegistration> {
608 let mut values = vec![];
609 for (module, node) in self.nodes.iter().enumerate() {
610 values.extend(
611 (0..node.metadata.providers.len()).map(|slot| ProviderRegistration {
612 module,
613 slot: ProviderSlot::Provider(slot),
614 }),
615 );
616 values.extend(
617 (0..node.metadata.controllers.len()).map(|slot| ProviderRegistration {
618 module,
619 slot: ProviderSlot::Controller(slot),
620 }),
621 );
622 for slot in 0..node.metadata.gateways.len() {
623 values.push(ProviderRegistration {
624 module,
625 slot: ProviderSlot::Gateway(slot),
626 });
627 }
628 values.extend((0..node.metadata.microservices.len()).map(|slot| {
629 ProviderRegistration {
630 module,
631 slot: ProviderSlot::Microservice(slot),
632 }
633 }));
634 }
635 values
636 }
637 fn def(&self, registration: ProviderRegistration) -> &ProviderDef {
638 match registration.slot {
639 ProviderSlot::Provider(index) => {
640 &self.nodes[registration.module].metadata.providers[index]
641 }
642 ProviderSlot::Controller(index) => {
643 &self.nodes[registration.module].metadata.controllers[index].provider
644 }
645 ProviderSlot::Gateway(index) => {
646 &self.nodes[registration.module].metadata.gateways[index].provider
647 }
648 ProviderSlot::Microservice(index) => {
649 &self.nodes[registration.module].metadata.microservices[index].provider
650 }
651 }
652 }
653 fn local_types(&self, module: usize) -> HashSet<TypeId> {
654 self.definitions()
655 .into_iter()
656 .filter(|r| r.module == module)
657 .map(|r| self.def(r).type_id)
658 .collect()
659 }
660 fn preflight(&self, container: Option<&Container>) -> crate::Result<Vec<ProviderRegistration>> {
666 let definitions = self.definitions();
667 let mut by_type = HashMap::new();
668 for registration in &definitions {
669 let def = self.def(*registration);
670 if let Some(existing) = by_type.insert(def.type_id, *registration) {
671 if !container.is_some_and(|container| {
672 container.has_pending_override(def.type_id)
673 || container.was_overridden(def.type_id)
674 }) {
675 return Err(crate::exception::startup_error(format!(
676 "duplicate provider registration for {} in {} and {}",
677 def.type_name,
678 self.nodes[existing.module].type_name,
679 self.nodes[registration.module].type_name
680 )));
681 }
682 }
683 }
684 let locals: Vec<HashSet<TypeId>> =
685 (0..self.nodes.len()).map(|i| self.local_types(i)).collect();
686 let mut exports = vec![HashSet::new(); self.nodes.len()];
687 for index in 0..self.nodes.len() {
688 let node = &self.nodes[index];
689 for export in &node.metadata.exports {
690 let imported = node
691 .imports
692 .iter()
693 .any(|&child| exports[child].contains(&export.type_id));
694 if !locals[index].contains(&export.type_id) && !imported {
695 return Err(crate::exception::startup_error(format!(
696 "module {} cannot export {}: it is neither declared locally nor exported by a direct import",
697 node.type_name, export.type_name
698 )));
699 }
700 exports[index].insert(export.type_id);
701 }
702 }
703 let global_exports: HashSet<TypeId> = self
704 .nodes
705 .iter()
706 .enumerate()
707 .filter(|(_, node)| node.metadata.global)
708 .flat_map(|(index, _)| exports[index].iter().copied())
709 .collect();
710 let visible = |module: usize, type_id: TypeId| {
711 locals[module].contains(&type_id)
712 || global_exports.contains(&type_id)
713 || self.nodes[module]
714 .imports
715 .iter()
716 .any(|&child| exports[child].contains(&type_id))
717 };
718 for registration in &definitions {
719 let production = self.def(*registration);
720 let effective = container
721 .and_then(|container| container.pending_override(production.type_id))
722 .unwrap_or(production);
723 for dependency in &effective.dependencies {
724 if dependency.type_id == TypeId::of::<crate::Logger>() {
725 continue;
726 }
727 if by_type.contains_key(&dependency.type_id) {
728 if !visible(registration.module, dependency.type_id) {
729 return Err(crate::exception::startup_error(format!(
730 "{} depends on {} but it is not visible in module {}; import and export its module",
731 effective.type_name,
732 dependency.type_name,
733 self.nodes[registration.module].type_name
734 )));
735 }
736 } else if !container
737 .is_some_and(|container| container.contains_type_id(dependency.type_id))
738 {
739 return Err(crate::exception::startup_error(format!(
740 "missing provider at startup: {} depends on {} but no provider is registered",
741 effective.type_name, dependency.type_name
742 )));
743 }
744 }
745 }
746 for (module, node) in self.nodes.iter().enumerate() {
747 for handler in &node.metadata.event_handlers {
748 handler.assert_registered_or_declared(&locals[module])?;
749 if !visible(module, TypeId::of::<crate::EventBus>()) {
750 return Err(crate::exception::startup_error(format!(
751 "no provider registered for EventBus in module {}; import EventModule",
752 node.type_name
753 )));
754 }
755 }
756 }
757 let mut command_patterns: HashMap<&'static str, usize> = HashMap::new();
758 let mut event_patterns: HashMap<&'static str, usize> = HashMap::new();
759 for (module, node) in self.nodes.iter().enumerate() {
760 for microservice in &node.metadata.microservices {
761 for handler in microservice.handlers() {
762 if !valid_microservice_pattern(
763 handler.pattern,
764 handler.kind == MessageHandlerKind::Command,
765 ) {
766 return Err(crate::exception::startup_error(format!(
767 "invalid microservice subject `{}` in module {}",
768 handler.pattern, node.type_name
769 )));
770 }
771 let (other, existing) = if handler.kind == MessageHandlerKind::Command {
772 (&event_patterns, command_patterns.get(handler.pattern))
773 } else {
774 (&command_patterns, event_patterns.get(handler.pattern))
775 };
776 if let Some(other_module) = other.get(handler.pattern) {
777 return Err(crate::exception::startup_error(format!(
778 "command and event handlers cannot share subject `{}` in {} and {}",
779 handler.pattern, self.nodes[*other_module].type_name, node.type_name
780 )));
781 }
782 if let Some(existing) = existing {
783 return Err(crate::exception::startup_error(format!(
784 "duplicate {:?} handler subject `{}` in {} and {}",
785 handler.kind,
786 handler.pattern,
787 self.nodes[*existing].type_name,
788 node.type_name
789 )));
790 }
791 if handler.kind == MessageHandlerKind::Command {
792 command_patterns.insert(handler.pattern, module);
793 } else {
794 event_patterns.insert(handler.pattern, module);
795 }
796 }
797 }
798 }
799 for command in command_patterns.keys() {
800 for event in event_patterns.keys() {
801 if microservice_patterns_intersect(command, event) {
802 return Err(crate::exception::startup_error(format!(
803 "command subject `{command}` overlaps event subject `{event}`; command and event transports must be disjoint"
804 )));
805 }
806 }
807 }
808 let event_subjects: Vec<_> = event_patterns.keys().copied().collect();
809 for (index, first) in event_subjects.iter().enumerate() {
810 for second in event_subjects.iter().skip(index + 1) {
811 if microservice_patterns_intersect(first, second) {
812 return Err(crate::exception::startup_error(format!(
813 "event subjects `{first}` and `{second}` overlap; each event delivery must have one owner"
814 )));
815 }
816 }
817 }
818 let mut sorted = vec![];
819 let mut states = HashMap::new();
820 let mut stack = vec![];
821 fn schedule(
822 reg: ProviderRegistration,
823 graph: &ModuleGraph,
824 by_type: &HashMap<TypeId, ProviderRegistration>,
825 container: Option<&Container>,
826 states: &mut HashMap<TypeId, u8>,
827 stack: &mut Vec<&'static str>,
828 sorted: &mut Vec<ProviderRegistration>,
829 ) -> crate::Result<()> {
830 let def = graph.def(reg);
831 match states.get(&def.type_id).copied() {
832 Some(2) => return Ok(()),
833 Some(1) => {
834 let mut cycle = stack.clone();
835 cycle.push(def.type_name);
836 return Err(crate::exception::startup_error(format!(
837 "provider dependency cycle: {}",
838 cycle.join(" -> ")
839 )));
840 }
841 _ => {}
842 }
843 states.insert(def.type_id, 1);
844 stack.push(def.type_name);
845 let effective = container
846 .and_then(|container| container.pending_override(def.type_id))
847 .unwrap_or(def);
848 for dep in &effective.dependencies {
849 if let Some(next) = by_type.get(&dep.type_id) {
850 schedule(*next, graph, by_type, container, states, stack, sorted)?;
851 }
852 }
853 stack.pop();
854 states.insert(def.type_id, 2);
855 sorted.push(reg);
856 Ok(())
857 }
858 for registration in definitions {
859 schedule(
860 registration,
861 self,
862 &by_type,
863 container,
864 &mut states,
865 &mut stack,
866 &mut sorted,
867 )?;
868 }
869 Ok(sorted)
870 }
871 fn def_by_type(&self, type_id: TypeId) -> Option<&ProviderDef> {
872 self.definitions().into_iter().find_map(|r| {
873 let def = self.def(r);
874 (def.type_id == type_id).then_some(def)
875 })
876 }
877}
878
879async fn initialize_graph(graph: &ModuleGraph, container: &mut Container) -> crate::Result<()> {
880 let order = graph.preflight(Some(container))?;
881 for registration in order {
882 let declared = graph.def(registration);
883 container.mark_provider_declared(declared.type_id);
884 if container.contains_type_id(declared.type_id)
885 || container.was_overridden(declared.type_id)
886 {
887 continue;
888 }
889 let start = Instant::now();
890 let override_def = container.take_pending_override(declared.type_id);
891 if override_def.is_some() {
892 container.mark_override_applied(declared.type_id);
893 }
894 let effective = override_def.as_ref().unwrap_or(declared);
895 let scoped_container =
896 container.scoped_for_provider(effective.type_name, &effective.dependencies);
897 let value = match (effective.build)(&scoped_container).await {
898 Ok(value) => value,
899 Err(error) => {
900 rollback_graph(graph, container).await;
901 return Err(error);
902 }
903 };
904 container.register_erased(effective.type_id, value.clone());
905 if let Err(error) = effective
906 .run_lifecycle(&value, "on_module_init", &effective.init_fn)
907 .await
908 {
909 rollback_graph(graph, container).await;
910 return Err(error);
911 }
912 if override_def.is_none() {
913 container.record_initialized_provider(declared.type_id);
914 }
915 crate::log_provider_initialized(effective.type_name, start.elapsed());
916 }
917 for node in &graph.nodes {
918 for handler in &node.metadata.event_handlers {
919 if let Err(error) = handler.register(container) {
920 rollback_graph(graph, container).await;
921 return Err(error);
922 }
923 }
924 crate::log_module_initialized(node.type_name);
925 }
926 Ok(())
927}
928
929async fn bootstrap_graph(graph: &ModuleGraph, container: &Container) -> crate::Result<()> {
930 let order = graph.preflight(Some(container))?;
931 let initialized: HashSet<TypeId> = container.initialized_provider_types().into_iter().collect();
932 for registration in order {
933 let def = graph.def(registration);
934 if initialized.contains(&def.type_id)
935 && !container.was_overridden(def.type_id)
936 && container.begin_provider_bootstrap(def.type_id)
937 {
938 if let Err(error) = def
939 .run_lifecycle_from_container(container, "on_bootstrap", &def.bootstrap_fn)
940 .await
941 {
942 rollback_graph(graph, container).await;
943 return Err(error);
944 }
945 }
946 }
947 Ok(())
948}
949
950async fn rollback_graph(graph: &ModuleGraph, container: &Container) {
951 for type_id in container.take_initialized_providers().into_iter().rev() {
952 if let Some(def) = graph.def_by_type(type_id) {
953 let _ = def
954 .run_lifecycle_from_container(container, "on_shutdown", &def.shutdown_fn)
955 .await;
956 }
957 }
958}
959
960pub async fn register_module<M: Module + 'static>(container: &mut Container) -> Result<()> {
962 let graph = ModuleGraph::discover::<M>()?;
963 initialize_graph(&graph, container).await
964}
965pub async fn bootstrap_module<M: Module + 'static>(container: &Container) -> Result<()> {
967 let graph = ModuleGraph::discover::<M>()?;
968 bootstrap_graph(&graph, container).await
969}
970pub async fn shutdown_module<M: Module + 'static>(container: &Container) -> Result<()> {
972 let graph = ModuleGraph::discover::<M>()?;
973 let mut first = None;
974 for type_id in container.take_initialized_providers().into_iter().rev() {
975 if let Some(def) = graph.def_by_type(type_id) {
976 if let Err(error) = def
977 .run_lifecycle_from_container(container, "on_shutdown", &def.shutdown_fn)
978 .await
979 {
980 if first.is_none() {
981 first = Some(error);
982 }
983 }
984 }
985 }
986 first.map_or(Ok(()), Err)
987}
988
989pub async fn build_container<M: Module + 'static>() -> Result<Container> {
991 build_container_with_overrides::<M>(ProviderOverrides::new()).await
992}
993#[doc(hidden)]
994pub async fn build_container_with_setup<M: Module + 'static>(
995 setup: impl FnOnce(&mut Container),
996) -> Result<Container> {
997 let mut container = Container::new();
998 setup(&mut container);
999 if let Err(error) = register_module::<M>(&mut container).await {
1000 return Err(error);
1001 }
1002 if let Err(error) =
1003 validate_module_providers::<M>(&container).and_then(|_| validate_gateway_paths::<M>())
1004 {
1005 let graph = ModuleGraph::discover::<M>()?;
1006 rollback_graph(&graph, &container).await;
1007 return Err(error);
1008 }
1009 if let Err(error) = bootstrap_module::<M>(&container).await {
1010 return Err(error);
1011 }
1012 Ok(container)
1013}
1014pub async fn build_container_with_overrides<M: Module + 'static>(
1016 overrides: ProviderOverrides,
1017) -> crate::Result<Container> {
1018 let mut container = Container::new();
1019 container.seed_overrides(overrides);
1020 if let Err(error) = register_module::<M>(&mut container).await {
1021 return Err(error);
1022 }
1023 if let Err(error) = container
1024 .assert_no_unused_overrides()
1025 .and_then(|_| validate_module_providers::<M>(&container))
1026 .and_then(|_| validate_gateway_paths::<M>())
1027 {
1028 let graph = ModuleGraph::discover::<M>()?;
1029 rollback_graph(&graph, &container).await;
1030 return Err(error);
1031 }
1032 if let Err(error) = bootstrap_module::<M>(&container).await {
1033 return Err(error);
1034 }
1035 Ok(container)
1036}
1037
1038pub fn collect_module_message_handlers<M: Module + 'static>() -> Result<Vec<MessageHandlerDef>> {
1044 collect_module_message_handlers_with_container::<M>(None)
1045}
1046
1047pub fn collect_module_message_handlers_with_container<M: Module + 'static>(
1051 container: Option<&Container>,
1052) -> Result<Vec<MessageHandlerDef>> {
1053 let graph = ModuleGraph::discover::<M>()?;
1054 graph.preflight(container)?;
1055 let mut handlers = Vec::new();
1056 for node in &graph.nodes {
1057 for microservice in &node.metadata.microservices {
1058 handlers.extend(microservice.handlers());
1059 }
1060 }
1061 Ok(handlers)
1062}
1063
1064fn validate_gateway_paths_in_graph(graph: &ModuleGraph) -> crate::Result<()> {
1065 let mut paths = HashSet::new();
1066 for node in &graph.nodes {
1067 for gateway in &node.metadata.gateways {
1068 if !gateway.path.starts_with('/') {
1069 return Err(crate::exception::startup_error(format!(
1070 "websocket gateway path must start with '/': {}",
1071 gateway.path
1072 )));
1073 }
1074 if !paths.insert((gateway.path, gateway.is_websocket())) {
1075 return Err(crate::exception::startup_error(format!(
1076 "duplicate gateway path: {}",
1077 gateway.path
1078 )));
1079 }
1080 }
1081 }
1082 Ok(())
1083}
1084
1085fn valid_microservice_pattern(pattern: &str, command: bool) -> bool {
1086 if pattern.is_empty() || pattern.chars().any(char::is_whitespace) {
1087 return false;
1088 }
1089 let tokens: Vec<_> = pattern.split('.').collect();
1090 tokens.iter().enumerate().all(|(index, token)| {
1091 !token.is_empty()
1092 && (!token.contains('*') || (!command && *token == "*"))
1093 && (!token.contains('>') || (!command && *token == ">" && index + 1 == tokens.len()))
1094 })
1095}
1096
1097fn microservice_patterns_intersect(first: &str, second: &str) -> bool {
1098 let first: Vec<_> = first.split('.').collect();
1099 let second: Vec<_> = second.split('.').collect();
1100 let mut index = 0;
1101 loop {
1102 let left = first.get(index).copied();
1103 let right = second.get(index).copied();
1104 match (left, right) {
1105 (None, None) => return true,
1106 (Some(">"), Some(_)) | (Some(_), Some(">")) => return true,
1107 (Some(left), Some(right)) if left == "*" || right == "*" || left == right => {
1108 index += 1;
1109 }
1110 _ => return false,
1111 }
1112 }
1113}
1114
1115#[cfg(test)]
1116mod microservice_pattern_tests {
1117 use super::microservice_patterns_intersect;
1118
1119 #[test]
1120 fn terminal_wildcards_do_not_overlap_their_empty_prefix() {
1121 assert!(!microservice_patterns_intersect("orders", "orders.>"));
1122 assert!(microservice_patterns_intersect(
1123 "orders.created",
1124 "orders.>"
1125 ));
1126 }
1127}
1128
1129fn validate_gateway_paths<M: Module + 'static>() -> crate::Result<()> {
1130 validate_gateway_paths_in_graph(&ModuleGraph::discover::<M>()?)
1131}
1132
1133pub fn validate_module<M: Module + 'static>() -> Result<()> {
1140 let graph = ModuleGraph::discover::<M>()?;
1141 graph.preflight(None)?;
1142 validate_gateway_paths_in_graph(&graph)
1143}
1144
1145pub fn validate_module_providers<M: Module + 'static>(container: &Container) -> Result<()> {
1147 let graph = ModuleGraph::discover::<M>()?;
1148 graph.preflight(Some(container))?;
1149 for registration in graph.definitions() {
1150 graph.def(registration).assert_registered(container)?;
1151 }
1152 Ok(())
1153}
1154
1155pub fn register_module_controllers<M: Module + 'static>(any: &mut dyn Any) {
1157 if let Ok(graph) = ModuleGraph::discover::<M>() {
1158 for node in graph.nodes {
1159 for controller in &node.metadata.controllers {
1160 (controller.register_fn)(any);
1161 }
1162 }
1163 }
1164}
1165
1166#[doc(hidden)]
1168pub fn register_module_controllers_with_container<M: Module + 'static>(
1169 any: &mut dyn Any,
1170 container: Arc<Container>,
1171) {
1172 if let Ok(graph) = ModuleGraph::discover::<M>() {
1173 for node in graph.nodes {
1174 for controller in &node.metadata.controllers {
1175 (controller.register_with_container_fn)(any, container.clone());
1176 }
1177 }
1178 }
1179}
1180#[cfg(feature = "openapi")]
1181pub(crate) fn module_registers_provider<M: Module + 'static, T: 'static>() -> bool {
1182 ModuleGraph::discover::<M>().is_ok_and(|graph| {
1183 graph.nodes.iter().any(|node| {
1184 node.metadata
1185 .providers
1186 .iter()
1187 .any(|provider| provider.type_id == TypeId::of::<T>())
1188 })
1189 })
1190}
1191#[cfg(feature = "openapi")]
1192#[doc(hidden)]
1193pub fn visit_module_openapi_routes<M: Module + 'static>(
1194 visitor: &mut impl FnMut(&crate::openapi::OpenApiRouteDef),
1195) {
1196 if let Ok(graph) = ModuleGraph::discover::<M>() {
1197 for node in graph.nodes {
1198 for controller in &node.metadata.controllers {
1199 for route in (controller.openapi_routes_fn)() {
1200 visitor(route);
1201 }
1202 }
1203 }
1204 }
1205}
1206pub fn visit_module_gateways<M: Module + 'static>(visitor: &mut impl FnMut(&GatewayDef)) {
1208 if let Ok(graph) = ModuleGraph::discover::<M>() {
1209 let mut seen = HashSet::new();
1210 for node in graph.nodes {
1211 for gateway in &node.metadata.gateways {
1212 if seen.insert(gateway.type_id) {
1213 visitor(gateway);
1214 }
1215 }
1216 }
1217 }
1218}