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        // ── TLS cert reload intercept ──────────────────────────────────────
168        // Infrastructure command — bypasses journal recovery + dedup.
169        // Reloads are idempotent and NOT journaled.
170        if let RuntimeCommand::ReloadTlsCerts {
171            scheme, host, port, ..
172        } = &cmd
173        {
174            let registry = camel_component_api::tls_source::TlsReloadRegistry::global();
175            match registry.find(scheme, host, *port) {
176                Some(handler) => {
177                    handler.reload().await?;
178                    return Ok(RuntimeCommandResult::TlsCertsReloaded {
179                        scheme: scheme.clone(),
180                        host: host.clone(),
181                        port: *port,
182                    });
183                }
184                None => {
185                    return Err(CamelError::Config(format!(
186                        "no TLS server found for {scheme}://{host}:{port}"
187                    )));
188                }
189            }
190        }
191        // ── End TLS reload intercept ────────────────────────────────────────
192
193        self.ensure_journal_recovered().await?;
194        let command_id = cmd.command_id().to_string();
195        if !self.dedup.first_seen(&command_id).await? {
196            return Ok(RuntimeCommandResult::Duplicate { command_id });
197        }
198        let deps = self.deps();
199        match execute_command(&deps, cmd).await {
200            Ok(result) => Ok(result),
201            Err(err) => {
202                let _ = self.dedup.forget_seen(&command_id).await;
203                Err(err)
204            }
205        }
206    }
207}
208
209#[async_trait]
210impl RuntimeQueryBus for RuntimeBus {
211    async fn ask(&self, query: RuntimeQuery) -> Result<RuntimeQueryResult, CamelError> {
212        self.ensure_journal_recovered().await?;
213
214        match query {
215            RuntimeQuery::InFlightCount { route_id } => {
216                if let Some(execution) = &self.execution {
217                    execution
218                        .in_flight_count(&route_id)
219                        .await
220                        .map(|r| r.into())
221                        .map_err(Into::into)
222                } else {
223                    Ok(RuntimeQueryResult::RouteNotFound { route_id })
224                }
225            }
226            other => {
227                let deps = self.query_deps();
228                execute_query(&deps, other).await
229            }
230        }
231    }
232}
233
234#[async_trait]
235impl RouteRegistrationPort for RuntimeBus {
236    async fn register_route(&self, def: RouteDefinition) -> Result<(), DomainError> {
237        self.ensure_journal_recovered()
238            .await
239            .map_err(|e| DomainError::InvalidState(e.to_string()))?;
240        let deps = self.deps();
241        handle_register_internal(&deps, def)
242            .await
243            .map(|_| ())
244            .map_err(|e| match e {
245                CamelError::RouteError(msg) => DomainError::InvalidState(msg),
246                other => DomainError::InvalidState(other.to_string()),
247            })
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use crate::lifecycle::domain::DomainError;
254
255    use super::*;
256    use std::collections::{HashMap, HashSet};
257    use std::sync::Mutex;
258
259    use crate::lifecycle::application::route_definition::RouteDefinition;
260    use crate::lifecycle::domain::{RouteRuntimeAggregate, RuntimeEvent};
261    use crate::lifecycle::ports::RouteRegistrationPort as InternalRuntimeCommandBus;
262    use crate::lifecycle::ports::RouteStatusProjection;
263
264    #[derive(Clone, Default)]
265    struct InMemoryTestRepo {
266        routes: Arc<Mutex<HashMap<String, RouteRuntimeAggregate>>>,
267    }
268
269    #[async_trait]
270    impl RouteRepositoryPort for InMemoryTestRepo {
271        async fn load(&self, route_id: &str) -> Result<Option<RouteRuntimeAggregate>, DomainError> {
272            Ok(self
273                .routes
274                .lock()
275                .expect("lock test routes")
276                .get(route_id)
277                .cloned())
278        }
279
280        async fn save(&self, aggregate: RouteRuntimeAggregate) -> Result<(), DomainError> {
281            self.routes
282                .lock()
283                .expect("lock test routes")
284                .insert(aggregate.route_id().to_string(), aggregate);
285            Ok(())
286        }
287
288        async fn save_if_version(
289            &self,
290            aggregate: RouteRuntimeAggregate,
291            expected_version: u64,
292        ) -> Result<(), DomainError> {
293            let route_id = aggregate.route_id().to_string();
294            let mut routes = self.routes.lock().expect("lock test routes");
295            let current = routes.get(&route_id).ok_or_else(|| {
296                DomainError::InvalidState(format!(
297                    "optimistic lock conflict for route '{route_id}': route not found"
298                ))
299            })?;
300
301            if current.version() != expected_version {
302                return Err(DomainError::InvalidState(format!(
303                    "optimistic lock conflict for route '{route_id}': expected version {expected_version}, actual {}",
304                    current.version()
305                )));
306            }
307
308            routes.insert(route_id, aggregate);
309            Ok(())
310        }
311
312        async fn delete(&self, route_id: &str) -> Result<(), DomainError> {
313            self.routes
314                .lock()
315                .expect("lock test routes")
316                .remove(route_id);
317            Ok(())
318        }
319    }
320
321    #[derive(Clone, Default)]
322    struct InMemoryTestProjectionStore {
323        statuses: Arc<Mutex<HashMap<String, RouteStatusProjection>>>,
324    }
325
326    #[async_trait]
327    impl ProjectionStorePort for InMemoryTestProjectionStore {
328        async fn upsert_status(&self, status: RouteStatusProjection) -> Result<(), DomainError> {
329            self.statuses
330                .lock()
331                .expect("lock test statuses")
332                .insert(status.route_id.clone(), status);
333            Ok(())
334        }
335
336        async fn get_status(
337            &self,
338            route_id: &str,
339        ) -> Result<Option<RouteStatusProjection>, DomainError> {
340            Ok(self
341                .statuses
342                .lock()
343                .expect("lock test statuses")
344                .get(route_id)
345                .cloned())
346        }
347
348        async fn list_statuses(&self) -> Result<Vec<RouteStatusProjection>, DomainError> {
349            Ok(self
350                .statuses
351                .lock()
352                .expect("lock test statuses")
353                .values()
354                .cloned()
355                .collect())
356        }
357
358        async fn remove_status(&self, route_id: &str) -> Result<(), DomainError> {
359            self.statuses
360                .lock()
361                .expect("lock test statuses")
362                .remove(route_id);
363            Ok(())
364        }
365    }
366
367    #[derive(Clone, Default)]
368    struct InMemoryTestEventPublisher;
369
370    #[async_trait]
371    impl EventPublisherPort for InMemoryTestEventPublisher {
372        async fn publish(&self, _events: &[RuntimeEvent]) -> Result<(), DomainError> {
373            Ok(())
374        }
375    }
376
377    #[derive(Clone, Default)]
378    struct InMemoryTestDedup {
379        seen: Arc<Mutex<HashSet<String>>>,
380    }
381
382    #[derive(Clone, Default)]
383    struct InspectableDedup {
384        seen: Arc<Mutex<HashSet<String>>>,
385        forget_calls: Arc<Mutex<u32>>,
386    }
387
388    #[async_trait]
389    impl CommandDedupPort for InMemoryTestDedup {
390        async fn first_seen(&self, command_id: &str) -> Result<bool, DomainError> {
391            let mut seen = self.seen.lock().expect("lock dedup set");
392            Ok(seen.insert(command_id.to_string()))
393        }
394
395        async fn forget_seen(&self, command_id: &str) -> Result<(), DomainError> {
396            self.seen.lock().expect("lock dedup set").remove(command_id);
397            Ok(())
398        }
399    }
400
401    #[async_trait]
402    impl CommandDedupPort for InspectableDedup {
403        async fn first_seen(&self, command_id: &str) -> Result<bool, DomainError> {
404            let mut seen = self.seen.lock().expect("lock dedup set");
405            Ok(seen.insert(command_id.to_string()))
406        }
407
408        async fn forget_seen(&self, command_id: &str) -> Result<(), DomainError> {
409            self.seen.lock().expect("lock dedup set").remove(command_id);
410            let mut calls = self.forget_calls.lock().expect("forget calls");
411            *calls += 1;
412            Ok(())
413        }
414    }
415
416    fn build_test_runtime_bus() -> RuntimeBus {
417        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
418        let projections: Arc<dyn ProjectionStorePort> =
419            Arc::new(InMemoryTestProjectionStore::default());
420        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
421        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
422        RuntimeBus::new(repo, projections, events, dedup)
423    }
424
425    #[derive(Default)]
426    struct CountingUow {
427        recover_calls: Arc<Mutex<u32>>,
428    }
429
430    #[derive(Default)]
431    struct FailingRecoverUow;
432
433    #[async_trait]
434    impl RuntimeUnitOfWorkPort for CountingUow {
435        async fn persist_upsert(
436            &self,
437            _aggregate: RouteRuntimeAggregate,
438            _expected_version: Option<u64>,
439            _projection: RouteStatusProjection,
440            _events: &[RuntimeEvent],
441        ) -> Result<(), DomainError> {
442            Ok(())
443        }
444
445        async fn persist_delete(
446            &self,
447            _route_id: &str,
448            _events: &[RuntimeEvent],
449        ) -> Result<(), DomainError> {
450            Ok(())
451        }
452
453        async fn recover_from_journal(&self) -> Result<(), DomainError> {
454            let mut calls = self.recover_calls.lock().expect("recover_calls");
455            *calls += 1;
456            Ok(())
457        }
458    }
459
460    #[async_trait]
461    impl RuntimeUnitOfWorkPort for FailingRecoverUow {
462        async fn persist_upsert(
463            &self,
464            _aggregate: RouteRuntimeAggregate,
465            _expected_version: Option<u64>,
466            _projection: RouteStatusProjection,
467            _events: &[RuntimeEvent],
468        ) -> Result<(), DomainError> {
469            Ok(())
470        }
471
472        async fn persist_delete(
473            &self,
474            _route_id: &str,
475            _events: &[RuntimeEvent],
476        ) -> Result<(), DomainError> {
477            Ok(())
478        }
479
480        async fn recover_from_journal(&self) -> Result<(), DomainError> {
481            Err(DomainError::InvalidState("recover failed".into()))
482        }
483    }
484
485    #[derive(Default)]
486    struct InFlightExecutionPort;
487
488    #[async_trait]
489    impl RuntimeExecutionPort for InFlightExecutionPort {
490        async fn register_route(&self, _definition: RouteDefinition) -> Result<(), DomainError> {
491            Ok(())
492        }
493        async fn start_route(&self, _route_id: &str) -> Result<(), DomainError> {
494            Ok(())
495        }
496        async fn stop_route(&self, _route_id: &str) -> Result<(), DomainError> {
497            Ok(())
498        }
499        async fn suspend_route(&self, _route_id: &str) -> Result<(), DomainError> {
500            Ok(())
501        }
502        async fn resume_route(&self, _route_id: &str) -> Result<(), DomainError> {
503            Ok(())
504        }
505        async fn reload_route(&self, _route_id: &str) -> Result<(), DomainError> {
506            Ok(())
507        }
508        async fn remove_route(&self, _route_id: &str) -> Result<(), DomainError> {
509            Ok(())
510        }
511        async fn in_flight_count(
512            &self,
513            route_id: &str,
514        ) -> Result<InFlightCountResult, DomainError> {
515            if route_id == "known" {
516                Ok(InFlightCountResult::InFlightCount {
517                    route_id: route_id.to_string(),
518                    count: 3,
519                })
520            } else {
521                Ok(InFlightCountResult::RouteNotFound {
522                    route_id: route_id.to_string(),
523                })
524            }
525        }
526    }
527
528    #[tokio::test]
529    async fn runtime_bus_implements_internal_command_bus() {
530        let bus = build_test_runtime_bus();
531        let def = RouteDefinition::new("timer:test", vec![]).with_route_id("internal-route");
532        let result = InternalRuntimeCommandBus::register_route(&bus, def).await;
533        assert!(
534            result.is_ok(),
535            "internal bus registration failed: {:?}",
536            result
537        );
538
539        let status = bus
540            .ask(RuntimeQuery::GetRouteStatus {
541                route_id: "internal-route".to_string(),
542            })
543            .await
544            .unwrap();
545        match status {
546            RuntimeQueryResult::RouteStatus { status, .. } => {
547                assert_eq!(status, "Registered");
548            }
549            _ => panic!("unexpected query result"),
550        }
551    }
552
553    #[tokio::test]
554    async fn execute_returns_duplicate_for_replayed_command_id() {
555        use camel_api::runtime::{CanonicalRouteSpec, CanonicalStepSpec, RuntimeCommand};
556
557        let bus = build_test_runtime_bus();
558
559        let mut spec = CanonicalRouteSpec::new("dup-route", "timer:tick");
560        spec.steps = vec![CanonicalStepSpec::Stop];
561
562        let cmd = RuntimeCommand::RegisterRoute {
563            spec: spec.clone(),
564            command_id: "dup-cmd".into(),
565            causation_id: None,
566        };
567        let first = bus.execute(cmd).await.unwrap();
568        assert!(matches!(
569            first,
570            RuntimeCommandResult::RouteRegistered { route_id } if route_id == "dup-route"
571        ));
572
573        let second = bus
574            .execute(RuntimeCommand::RegisterRoute {
575                spec,
576                command_id: "dup-cmd".into(),
577                causation_id: None,
578            })
579            .await
580            .unwrap();
581        assert!(matches!(
582            second,
583            RuntimeCommandResult::Duplicate { command_id } if command_id == "dup-cmd"
584        ));
585    }
586
587    #[tokio::test]
588    async fn ask_in_flight_count_without_execution_returns_route_not_found() {
589        let bus = build_test_runtime_bus();
590        let res = bus
591            .ask(RuntimeQuery::InFlightCount {
592                route_id: "missing".into(),
593            })
594            .await
595            .unwrap();
596        assert!(matches!(
597            res,
598            RuntimeQueryResult::RouteNotFound { route_id } if route_id == "missing"
599        ));
600    }
601
602    #[tokio::test]
603    async fn ask_in_flight_count_with_execution_delegates_to_adapter() {
604        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
605        let projections: Arc<dyn ProjectionStorePort> =
606            Arc::new(InMemoryTestProjectionStore::default());
607        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
608        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
609        let execution: Arc<dyn RuntimeExecutionPort> = Arc::new(InFlightExecutionPort);
610        let bus = RuntimeBus::new(repo, projections, events, dedup).with_execution(execution);
611
612        let known = bus
613            .ask(RuntimeQuery::InFlightCount {
614                route_id: "known".into(),
615            })
616            .await
617            .unwrap();
618        assert!(matches!(
619            known,
620            RuntimeQueryResult::InFlightCount { route_id, count }
621            if route_id == "known" && count == 3
622        ));
623    }
624
625    #[tokio::test]
626    async fn journal_recovery_runs_once_even_with_multiple_commands() {
627        use camel_api::runtime::{CanonicalRouteSpec, CanonicalStepSpec, RuntimeCommand};
628
629        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
630        let projections: Arc<dyn ProjectionStorePort> =
631            Arc::new(InMemoryTestProjectionStore::default());
632        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
633        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
634        let uow = Arc::new(CountingUow::default());
635        let bus = RuntimeBus::new(repo, projections, events, dedup).with_uow(uow.clone());
636
637        let mut spec_a = CanonicalRouteSpec::new("a", "timer:a");
638        spec_a.steps = vec![CanonicalStepSpec::Stop];
639        let mut spec_b = CanonicalRouteSpec::new("b", "timer:b");
640        spec_b.steps = vec![CanonicalStepSpec::Stop];
641
642        bus.execute(RuntimeCommand::RegisterRoute {
643            spec: spec_a,
644            command_id: "c-a".into(),
645            causation_id: None,
646        })
647        .await
648        .unwrap();
649
650        bus.execute(RuntimeCommand::RegisterRoute {
651            spec: spec_b,
652            command_id: "c-b".into(),
653            causation_id: None,
654        })
655        .await
656        .unwrap();
657
658        let calls = *uow.recover_calls.lock().expect("recover calls");
659        assert_eq!(calls, 1, "journal recovery should run once");
660    }
661
662    #[tokio::test]
663    async fn execute_on_command_error_forgets_dedup_marker() {
664        use camel_api::runtime::{CanonicalRouteSpec, RuntimeCommand};
665
666        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
667        let projections: Arc<dyn ProjectionStorePort> =
668            Arc::new(InMemoryTestProjectionStore::default());
669        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
670        let dedup = Arc::new(InspectableDedup::default());
671        let dedup_port: Arc<dyn CommandDedupPort> = dedup.clone();
672
673        let bus = RuntimeBus::new(repo, projections, events, dedup_port);
674
675        // Invalid canonical contract: empty route_id -> execute_command should fail.
676        let cmd = RuntimeCommand::RegisterRoute {
677            spec: CanonicalRouteSpec::new("", "timer:tick"),
678            command_id: "err-cmd".into(),
679            causation_id: None,
680        };
681
682        let err = bus.execute(cmd).await.expect_err("must fail");
683        assert!(err.to_string().contains("route_id cannot be empty"));
684
685        assert_eq!(*dedup.forget_calls.lock().expect("forget calls"), 1);
686        assert!(!dedup.seen.lock().expect("seen").contains("err-cmd"));
687    }
688
689    #[tokio::test]
690    async fn execute_propagates_recover_error_from_uow() {
691        use camel_api::runtime::{CanonicalRouteSpec, CanonicalStepSpec, RuntimeCommand};
692
693        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
694        let projections: Arc<dyn ProjectionStorePort> =
695            Arc::new(InMemoryTestProjectionStore::default());
696        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
697        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
698        let uow: Arc<dyn RuntimeUnitOfWorkPort> = Arc::new(FailingRecoverUow);
699
700        let bus = RuntimeBus::new(repo, projections, events, dedup).with_uow(uow);
701
702        let mut spec = CanonicalRouteSpec::new("x", "timer:x");
703        spec.steps = vec![CanonicalStepSpec::Stop];
704        let err = bus
705            .execute(RuntimeCommand::RegisterRoute {
706                spec,
707                command_id: "recover-err".into(),
708                causation_id: None,
709            })
710            .await
711            .expect_err("recover should fail");
712
713        assert!(err.to_string().contains("recover failed"));
714    }
715
716    #[tokio::test]
717    async fn ask_in_flight_count_with_execution_handles_unknown_route() {
718        let repo: Arc<dyn RouteRepositoryPort> = Arc::new(InMemoryTestRepo::default());
719        let projections: Arc<dyn ProjectionStorePort> =
720            Arc::new(InMemoryTestProjectionStore::default());
721        let events: Arc<dyn EventPublisherPort> = Arc::new(InMemoryTestEventPublisher);
722        let dedup: Arc<dyn CommandDedupPort> = Arc::new(InMemoryTestDedup::default());
723        let execution: Arc<dyn RuntimeExecutionPort> = Arc::new(InFlightExecutionPort);
724        let bus = RuntimeBus::new(repo, projections, events, dedup).with_execution(execution);
725
726        let unknown = bus
727            .ask(RuntimeQuery::InFlightCount {
728                route_id: "unknown".into(),
729            })
730            .await
731            .unwrap();
732        assert!(matches!(
733            unknown,
734            RuntimeQueryResult::RouteNotFound { route_id } if route_id == "unknown"
735        ));
736    }
737}
camel_core/lifecycle/application/runtime_bus.rs

camel_core/lifecycle/application/
runtime_bus.rs
