camel_core/lifecycle/application/
runtime_bus.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use tokio::sync::OnceCell;
5
6use camel_api::{
7    CamelError, RuntimeCommand, RuntimeCommandBus, RuntimeCommandResult, RuntimeQuery,
8    RuntimeQueryBus, RuntimeQueryResult,
9};
10
11use crate::health_registry::HealthCheckRegistry;
12use crate::lifecycle::application::commands::{
13    CommandDeps, execute_command, handle_register_internal,
14};
15use crate::lifecycle::application::queries::{QueryDeps, execute_query};
16use crate::lifecycle::application::route_definition::RouteDefinition;
17use crate::lifecycle::domain::DomainError;
18use crate::lifecycle::ports::RouteRegistrationPort;
19use crate::lifecycle::ports::{
20    CommandDedupPort, EventPublisherPort, InFlightCountResult, ProjectionStorePort,
21    RouteRepositoryPort, RuntimeExecutionPort, RuntimeUnitOfWorkPort,
22};
23
24impl From<InFlightCountResult> for RuntimeQueryResult {
25    fn from(r: InFlightCountResult) -> Self {
26        match r {
27            InFlightCountResult::InFlightCount { route_id, count } => {
28                RuntimeQueryResult::InFlightCount { route_id, count }
29            }
30            InFlightCountResult::RouteNotFound { route_id } => {
31                RuntimeQueryResult::RouteNotFound { route_id }
32            }
33        }
34    }
35}
36
37pub struct RuntimeBus {
38    repo: Arc<dyn RouteRepositoryPort>,
39    projections: Arc<dyn ProjectionStorePort>,
40    events: Arc<dyn EventPublisherPort>,
41    dedup: Arc<dyn CommandDedupPort>,
42    uow: Option<Arc<dyn RuntimeUnitOfWorkPort>>,
43    execution: Option<Arc<dyn RuntimeExecutionPort>>,
44    health_registry: Option<Arc<HealthCheckRegistry>>,
45    journal_recovered_once: OnceCell<()>,
46}
47
48impl RuntimeBus {
49    pub fn new(
50        repo: Arc<dyn RouteRepositoryPort>,
51        projections: Arc<dyn ProjectionStorePort>,
52        events: Arc<dyn EventPublisherPort>,
53        dedup: Arc<dyn CommandDedupPort>,
54    ) -> Self {
55        Self {
56            repo,
57            projections,
58            events,
59            dedup,
60            uow: None,
61            execution: None,
62            health_registry: None,
63            journal_recovered_once: OnceCell::new(),
64        }
65    }
66
67    pub fn with_uow(mut self, uow: Arc<dyn RuntimeUnitOfWorkPort>) -> Self {
68        self.uow = Some(uow);
69        self
70    }
71
72    pub fn with_execution(mut self, execution: Arc<dyn RuntimeExecutionPort>) -> Self {
73        self.execution = Some(execution);
74        self
75    }
76
77    pub fn with_health_registry(mut self, health_registry: Arc<HealthCheckRegistry>) -> Self {
78        self.health_registry = Some(health_registry);
79        self
80    }
81
82    pub fn repo(&self) -> &Arc<dyn RouteRepositoryPort> {
83        &self.repo
84    }
85
86    /// H8 boot reconciliation: fail any route still in a transient state
87    /// (`Starting` / `Stopping`) from a previous run. Called from
88    /// `CamelContext::start()` before `auto_startup_route_ids()`.
89    pub async fn reconcile_transient_states(&self) -> Result<(), CamelError> {
90        self.ensure_journal_recovered().await?;
91        let deps = self.deps();
92        crate::lifecycle::application::commands::reconcile_transient_states(&deps).await
93    }
94
95    pub(crate) async fn register_aggregate_only(&self, route_id: String) -> Result<(), CamelError> {
96        self.ensure_journal_recovered().await?;
97        let deps = self.deps();
98        if deps.repo.load(&route_id).await?.is_some() {
99            return Err(CamelError::RouteError(format!(
100                "route '{route_id}' already registered"
101            )));
102        }
103        let (aggregate, events) =
104            crate::lifecycle::domain::RouteRuntimeAggregate::register(route_id.clone());
105        if let Some(uow) = &deps.uow {
106            uow.persist_upsert(
107                aggregate.clone(),
108                None,
109                crate::lifecycle::application::commands::project_from_aggregate(&aggregate),
110                &events,
111            )
112            .await?;
113        } else {
114            deps.repo.save(aggregate.clone()).await?;
115            if let Some(primary_error) =
116                crate::lifecycle::application::commands::upsert_projection_with_reconciliation(
117                    &*deps.projections,
118                    crate::lifecycle::application::commands::project_from_aggregate(&aggregate),
119                )
120                .await?
121            {
122                deps.events.publish(&events).await?;
123                return Err(CamelError::RouteError(format!(
124                    "post-effect reconciliation recovered after runtime persistence error: {primary_error}"
125                )));
126            }
127            deps.events.publish(&events).await?;
128        }
129        Ok(())
130    }
131
132    fn deps(&self) -> CommandDeps {
133        CommandDeps {
134            repo: Arc::clone(&self.repo),
135            projections: Arc::clone(&self.projections),
136            events: Arc::clone(&self.events),
137            uow: self.uow.clone(),
138            execution: self.execution.clone(),
139            health_registry: self.health_registry.clone(),
140        }
141    }
142
143    fn query_deps(&self) -> QueryDeps {
144        QueryDeps {
145            projections: Arc::clone(&self.projections),
146        }
147    }
148
149    async fn ensure_journal_recovered(&self) -> Result<(), CamelError> {
150        let Some(uow) = &self.uow else {
151            return Ok(());
152        };
153
154        self.journal_recovered_once
155            .get_or_try_init(|| async {
156                uow.recover_from_journal().await?;
157                Ok::<(), CamelError>(())
158            })
159            .await?;
160        Ok(())
161    }
162}
163
164#[async_trait]
165impl RuntimeCommandBus for RuntimeBus {
166    async fn execute(&self, cmd: RuntimeCommand) -> Result<RuntimeCommandResult, CamelError> {
167        self.ensure_journal_recovered().await?;
168        let command_id = cmd.command_id().to_string();
169        if !self.dedup.first_seen(&command_id).await? {
170            return Ok(RuntimeCommandResult::Duplicate { command_id });
171        }
172        let deps = self.deps();
173        match execute_command(&deps, cmd).await {
174            Ok(result) => Ok(result),
175            Err(err) => {
176                let _ = self.dedup.forget_seen(&command_id).await;
177                Err(err)
178            }
179        }
180    }
181}
182
183#[async_trait]
184impl RuntimeQueryBus for RuntimeBus {
185    async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
186        self.ensure_journal_recovered().await?;
187
188        match query {
189            RuntimeQuery::InFlightCount { route_id } => {
190                if let Some(execution) = &self.execution {
191                    execution
192                        .in_flight_count(&route_id)
193                        .await
194                        .map(|r| r.into())
195                        .map_err(Into::into)
196                } else {
197                    Ok(RuntimeQueryResult::RouteNotFound { route_id })
198                }
199            }
200            other => {
201                let deps = self.query_deps();
202                execute_query(&deps, other).await
203            }
204        }
205    }
206}
207
208#[async_trait]
209impl RouteRegistrationPort for RuntimeBus {
210    async fn register_route(&self, def: RouteDefinition) -> Result<(), DomainError> {
211        self.ensure_journal_recovered()
212            .await
213            .map_err(|e| DomainError::InvalidState(e.to_string()))?;
214        let deps = self.deps();
215        handle_register_internal(&deps, def)
216            .await
217            .map(|_| ())
218            .map_err(|e| match e {
219                CamelError::RouteError(msg) => DomainError::InvalidState(msg),
220                other => DomainError::InvalidState(other.to_string()),
221            })
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use crate::lifecycle::domain::DomainError;
228
229    use super::*;
230    use std::collections::{HashMap, HashSet};
231    use std::sync::Mutex;
232
233    use crate::lifecycle::application::route_definition::RouteDefinition;
234    use crate::lifecycle::domain::{RouteRuntimeAggregate, RuntimeEvent};
235    use crate::lifecycle::ports::RouteRegistrationPort as InternalRuntimeCommandBus;
236    use crate::lifecycle::ports::RouteStatusProjection;
237
238    #[derive(Clone, Default)]
239    struct InMemoryTestRepo {
240        routes: Arc<Mutex<HashMap<String, RouteRuntimeAggregate>>>,
241    }
242
243    #[async_trait]
244    impl RouteRepositoryPort for InMemoryTestRepo {
245        async fn load(&self, route_id: &str) -> Result<Option<RouteRuntimeAggregate>, DomainError> {
246            Ok(self
247                .routes
248                .lock()
249                .expect("lock test routes")
250                .get(route_id)
251                .cloned())
252        }
253
254        async fn save(&self, aggregate: RouteRuntimeAggregate) -> Result<(), DomainError> {
255            self.routes
256                .lock()
257                .expect("lock test routes")
258                .insert(aggregate.route_id().to_string(), aggregate);
259            Ok(())
260        }
261
262        async fn save_if_version(
263            &self,
264            aggregate: RouteRuntimeAggregate,
265            expected_version: u64,
266        ) -> Result<(), DomainError> {
267            let route_id = aggregate.route_id().to_string();
268            let mut routes = self.routes.lock().expect("lock test routes");
269            let current = routes.get(&route_id).ok_or_else(|| {
270                DomainError::InvalidState(format!(
271                    "optimistic lock conflict for route '{route_id}': route not found"
272                ))
273            })?;
274
275            if current.version() != expected_version {
276                return Err(DomainError::InvalidState(format!(
277                    "optimistic lock conflict for route '{route_id}': expected version {expected_version}, actual {}",
278                    current.version()
279                )));
280            }
281
282            routes.insert(route_id, aggregate);
283            Ok(())
284        }
285
286        async fn delete(&self, route_id: &str) -> Result<(), DomainError> {
287            self.routes
288                .lock()
289                .expect("lock test routes")
290                .remove(route_id);
291            Ok(())
292        }
293    }
294
295    #[derive(Clone, Default)]
296    struct InMemoryTestProjectionStore {
297        statuses: Arc<Mutex<HashMap<String, RouteStatusProjection>>>,
298    }
299
300    #[async_trait]
301    impl ProjectionStorePort for InMemoryTestProjectionStore {
302        async fn upsert_status(&self, status: RouteStatusProjection) -> Result<(), DomainError> {
303            self.statuses
304                .lock()
305                .expect("lock test statuses")
306                .insert(status.route_id.clone(), status);
307            Ok(())
308        }
309
310        async fn get_status(
311            &self,
312            route_id: &str,
313        ) -> Result<Option<RouteStatusProjection>, DomainError> {
314            Ok(self
315                .statuses
316                .lock()
317                .expect("lock test statuses")
318                .get(route_id)
319                .cloned())
320        }
321
322        async fn list_statuses(&self) -> Result<Vec<RouteStatusProjection>, DomainError> {
323            Ok(self
324                .statuses
325                .lock()
326                .expect("lock test statuses")
327                .values()
328                .cloned()
329                .collect())
330        }
331
332        async fn remove_status(&self, route_id: &str) -> Result<(), DomainError> {
333            self.statuses
334                .lock()
335                .expect("lock test statuses")
336                .remove(route_id);
337            Ok(())
338        }
339    }
340
341    #[derive(Clone, Default)]
342    struct InMemoryTestEventPublisher;
343
344    #[async_trait]
345    impl EventPublisherPort for InMemoryTestEventPublisher {
346        async fn publish(&self, _events: &[RuntimeEvent]) -> Result<(), DomainError> {
347            Ok(())
348        }
349    }
350
351    #[derive(Clone, Default)]
352    struct InMemoryTestDedup {
353        seen: Arc<Mutex<HashSet<String>>>,
354    }
355
356    #[derive(Clone, Default)]
357    struct InspectableDedup {
358        seen: Arc<Mutex<HashSet<String>>>,
359        forget_calls: Arc<Mutex<u32>>,
360    }
361
362    #[async_trait]
363    impl CommandDedupPort for InMemoryTestDedup {
364        async fn first_seen(&self, command_id: &str) -> Result<bool, DomainError> {
365            let mut seen = self.seen.lock().expect("lock dedup set");
366            Ok(seen.insert(command_id.to_string()))
367        }
368
369        async fn forget_seen(&self, command_id: &str) -> Result<(), DomainError> {
370            self.seen.lock().expect("lock dedup set").remove(command_id);
371            Ok(())
372        }
373    }
374
375    #[async_trait]
376    impl CommandDedupPort for InspectableDedup {
377        async fn first_seen(&self, command_id: &str) -> Result<bool, DomainError> {
378            let mut seen = self.seen.lock().expect("lock dedup set");
379            Ok(seen.insert(command_id.to_string()))
380        }
381
382        async fn forget_seen(&self, command_id: &str) -> Result<(), DomainError> {
383            self.seen.lock().expect("lock dedup set").remove(command_id);
384            let mut calls = self.forget_calls.lock().expect("forget calls");
385            *calls += 1;
386            Ok(())
387        }
388    }
389
390    fn build_test_runtime_bus() -> RuntimeBus {
391        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
392        let projections: Arc<dyn ProjectionStorePort> =
393            Arc::new(InMemoryTestProjectionStore::default());
394        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
395        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
396        RuntimeBus::new(repo, projections, events, dedup)
397    }
398
399    #[derive(Default)]
400    struct CountingUow {
401        recover_calls: Arc<Mutex<u32>>,
402    }
403
404    #[derive(Default)]
405    struct FailingRecoverUow;
406
407    #[async_trait]
408    impl RuntimeUnitOfWorkPort for CountingUow {
409        async fn persist_upsert(
410            &self,
411            _aggregate: RouteRuntimeAggregate,
412            _expected_version: Option<u64>,
413            _projection: RouteStatusProjection,
414            _events: &[RuntimeEvent],
415        ) -> Result<(), DomainError> {
416            Ok(())
417        }
418
419        async fn persist_delete(
420            &self,
421            _route_id: &str,
422            _events: &[RuntimeEvent],
423        ) -> Result<(), DomainError> {
424            Ok(())
425        }
426
427        async fn recover_from_journal(&self) -> Result<(), DomainError> {
428            let mut calls = self.recover_calls.lock().expect("recover_calls");
429            *calls += 1;
430            Ok(())
431        }
432    }
433
434    #[async_trait]
435    impl RuntimeUnitOfWorkPort for FailingRecoverUow {
436        async fn persist_upsert(
437            &self,
438            _aggregate: RouteRuntimeAggregate,
439            _expected_version: Option<u64>,
440            _projection: RouteStatusProjection,
441            _events: &[RuntimeEvent],
442        ) -> Result<(), DomainError> {
443            Ok(())
444        }
445
446        async fn persist_delete(
447            &self,
448            _route_id: &str,
449            _events: &[RuntimeEvent],
450        ) -> Result<(), DomainError> {
451            Ok(())
452        }
453
454        async fn recover_from_journal(&self) -> Result<(), DomainError> {
455            Err(DomainError::InvalidState("recover failed".into()))
456        }
457    }
458
459    #[derive(Default)]
460    struct InFlightExecutionPort;
461
462    #[async_trait]
463    impl RuntimeExecutionPort for InFlightExecutionPort {
464        async fn register_route(&self, _definition: RouteDefinition) -> Result<(), DomainError> {
465            Ok(())
466        }
467        async fn start_route(&self, _route_id: &str) -> Result<(), DomainError> {
468            Ok(())
469        }
470        async fn stop_route(&self, _route_id: &str) -> Result<(), DomainError> {
471            Ok(())
472        }
473        async fn suspend_route(&self, _route_id: &str) -> Result<(), DomainError> {
474            Ok(())
475        }
476        async fn resume_route(&self, _route_id: &str) -> Result<(), DomainError> {
477            Ok(())
478        }
479        async fn reload_route(&self, _route_id: &str) -> Result<(), DomainError> {
480            Ok(())
481        }
482        async fn remove_route(&self, _route_id: &str) -> Result<(), DomainError> {
483            Ok(())
484        }
485        async fn in_flight_count(
486            &self,
487            route_id: &str,
488        ) -> Result<InFlightCountResult, DomainError> {
489            if route_id == "known" {
490                Ok(InFlightCountResult::InFlightCount {
491                    route_id: route_id.to_string(),
492                    count: 3,
493                })
494            } else {
495                Ok(InFlightCountResult::RouteNotFound {
496                    route_id: route_id.to_string(),
497                })
498            }
499        }
500    }
501
502    #[tokio::test]
503    async fn runtime_bus_implements_internal_command_bus() {
504        let bus = build_test_runtime_bus();
505        let def = RouteDefinition::new("timer:test", vec![]).with_route_id("internal-route");
506        let result = InternalRuntimeCommandBus::register_route(&bus, def).await;
507        assert!(
508            result.is_ok(),
509            "internal bus registration failed: {:?}",
510            result
511        );
512
513        let status = bus
514            .ask(RuntimeQuery::GetRouteStatus {
515                route_id: "internal-route".to_string(),
516            })
517            .await
518            .unwrap();
519        match status {
520            RuntimeQueryResult::RouteStatus { status, .. } => {
521                assert_eq!(status, "Registered");
522            }
523            _ => panic!("unexpected query result"),
524        }
525    }
526
527    #[tokio::test]
528    async fn execute_returns_duplicate_for_replayed_command_id() {
529        use camel_api::runtime::{CanonicalRouteSpec, CanonicalStepSpec, RuntimeCommand};
530
531        let bus = build_test_runtime_bus();
532
533        let mut spec = CanonicalRouteSpec::new("dup-route", "timer:tick");
534        spec.steps = vec![CanonicalStepSpec::Stop];
535
536        let cmd = RuntimeCommand::RegisterRoute {
537            spec: spec.clone(),
538            command_id: "dup-cmd".into(),
539            causation_id: None,
540        };
541        let first = bus.execute(cmd).await.unwrap();
542        assert!(matches!(
543            first,
544            RuntimeCommandResult::RouteRegistered { route_id } if route_id == "dup-route"
545        ));
546
547        let second = bus
548            .execute(RuntimeCommand::RegisterRoute {
549                spec,
550                command_id: "dup-cmd".into(),
551                causation_id: None,
552            })
553            .await
554            .unwrap();
555        assert!(matches!(
556            second,
557            RuntimeCommandResult::Duplicate { command_id } if command_id == "dup-cmd"
558        ));
559    }
560
561    #[tokio::test]
562    async fn ask_in_flight_count_without_execution_returns_route_not_found() {
563        let bus = build_test_runtime_bus();
564        let res = bus
565            .ask(RuntimeQuery::InFlightCount {
566                route_id: "missing".into(),
567            })
568            .await
569            .unwrap();
570        assert!(matches!(
571            res,
572            RuntimeQueryResult::RouteNotFound { route_id } if route_id == "missing"
573        ));
574    }
575
576    #[tokio::test]
577    async fn ask_in_flight_count_with_execution_delegates_to_adapter() {
578        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
579        let projections: Arc<dyn ProjectionStorePort> =
580            Arc::new(InMemoryTestProjectionStore::default());
581        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
582        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
583        let execution: Arc<dyn RuntimeExecutionPort> = Arc::new(InFlightExecutionPort);
584        let bus = RuntimeBus::new(repo, projections, events, dedup).with_execution(execution);
585
586        let known = bus
587            .ask(RuntimeQuery::InFlightCount {
588                route_id: "known".into(),
589            })
590            .await
591            .unwrap();
592        assert!(matches!(
593            known,
594            RuntimeQueryResult::InFlightCount { route_id, count }
595            if route_id == "known" && count == 3
596        ));
597    }
598
599    #[tokio::test]
600    async fn journal_recovery_runs_once_even_with_multiple_commands() {
601        use camel_api::runtime::{CanonicalRouteSpec, CanonicalStepSpec, RuntimeCommand};
602
603        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
604        let projections: Arc<dyn ProjectionStorePort> =
605            Arc::new(InMemoryTestProjectionStore::default());
606        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
607        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
608        let uow = Arc::new(CountingUow::default());
609        let bus = RuntimeBus::new(repo, projections, events, dedup).with_uow(uow.clone());
610
611        let mut spec_a = CanonicalRouteSpec::new("a", "timer:a");
612        spec_a.steps = vec![CanonicalStepSpec::Stop];
613        let mut spec_b = CanonicalRouteSpec::new("b", "timer:b");
614        spec_b.steps = vec![CanonicalStepSpec::Stop];
615
616        bus.execute(RuntimeCommand::RegisterRoute {
617            spec: spec_a,
618            command_id: "c-a".into(),
619            causation_id: None,
620        })
621        .await
622        .unwrap();
623
624        bus.execute(RuntimeCommand::RegisterRoute {
625            spec: spec_b,
626            command_id: "c-b".into(),
627            causation_id: None,
628        })
629        .await
630        .unwrap();
631
632        let calls = *uow.recover_calls.lock().expect("recover calls");
633        assert_eq!(calls, 1, "journal recovery should run once");
634    }
635
636    #[tokio::test]
637    async fn execute_on_command_error_forgets_dedup_marker() {
638        use camel_api::runtime::{CanonicalRouteSpec, RuntimeCommand};
639
640        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
641        let projections: Arc<dyn ProjectionStorePort> =
642            Arc::new(InMemoryTestProjectionStore::default());
643        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
644        let dedup = Arc::new(InspectableDedup::default());
645        let dedup_port: Arc<dyn CommandDedupPort> = dedup.clone();
646
647        let bus = RuntimeBus::new(repo, projections, events, dedup_port);
648
649        // Invalid canonical contract: empty route_id -> execute_command should fail.
650        let cmd = RuntimeCommand::RegisterRoute {
651            spec: CanonicalRouteSpec::new("", "timer:tick"),
652            command_id: "err-cmd".into(),
653            causation_id: None,
654        };
655
656        let err = bus.execute(cmd).await.expect_err("must fail");
657        assert!(err.to_string().contains("route_id cannot be empty"));
658
659        assert_eq!(*dedup.forget_calls.lock().expect("forget calls"), 1);
660        assert!(!dedup.seen.lock().expect("seen").contains("err-cmd"));
661    }
662
663    #[tokio::test]
664    async fn execute_propagates_recover_error_from_uow() {
665        use camel_api::runtime::{CanonicalRouteSpec, CanonicalStepSpec, RuntimeCommand};
666
667        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
668        let projections: Arc<dyn ProjectionStorePort> =
669            Arc::new(InMemoryTestProjectionStore::default());
670        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
671        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
672        let uow: Arc<dyn RuntimeUnitOfWorkPort> = Arc::new(FailingRecoverUow);
673
674        let bus = RuntimeBus::new(repo, projections, events, dedup).with_uow(uow);
675
676        let mut spec = CanonicalRouteSpec::new("x", "timer:x");
677        spec.steps = vec![CanonicalStepSpec::Stop];
678        let err = bus
679            .execute(RuntimeCommand::RegisterRoute {
680                spec,
681                command_id: "recover-err".into(),
682                causation_id: None,
683            })
684            .await
685            .expect_err("recover should fail");
686
687        assert!(err.to_string().contains("recover failed"));
688    }
689
690    #[tokio::test]
691    async fn ask_in_flight_count_with_execution_handles_unknown_route() {
692        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
693        let projections: Arc<dyn ProjectionStorePort> =
694            Arc::new(InMemoryTestProjectionStore::default());
695        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
696        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
697        let execution: Arc<dyn RuntimeExecutionPort> = Arc::new(InFlightExecutionPort);
698        let bus = RuntimeBus::new(repo, projections, events, dedup).with_execution(execution);
699
700        let unknown = bus
701            .ask(RuntimeQuery::InFlightCount {
702                route_id: "unknown".into(),
703            })
704            .await
705            .unwrap();
706        assert!(matches!(
707            unknown,
708            RuntimeQueryResult::RouteNotFound { route_id } if route_id == "unknown"
709        ));
710    }
711}
camel_core/lifecycle/application/runtime_bus.rs

camel_core/lifecycle/application/
runtime_bus.rs
