Skip to main content

camel_core/
context.rs

1use std::any::{Any, TypeId};
2use std::collections::HashMap;
3use std::sync::Arc;
4use tokio_util::sync::CancellationToken;
5use tracing::{debug, trace};
6
7#[cfg(test)]
8use camel_api::StepLifecycle;
9use camel_api::component_metadata::ComponentMetadata;
10use camel_api::error_handler::ErrorHandlerConfig;
11use camel_api::{
12    CamelError, FunctionInvoker, HealthReport, Lifecycle, MetricsCollector, PlatformIdentity,
13    PlatformService, ReadinessGate, RouteTemplateSpec, RuntimeCommandBus, RuntimeQueryBus,
14    TemplateInstanceRecord,
15};
16use camel_component_api::{Component, ComponentContext, ComponentRegistrar};
17use camel_language_api::Language;
18
19use crate::health_registry::HealthCheckRegistry;
20use crate::language_registry::LanguageRegistryError;
21use crate::lifecycle::adapters::controller_actor::RouteControllerHandle;
22use crate::lifecycle::adapters::route_controller::SharedLanguageRegistry;
23use crate::lifecycle::application::route_definition::RouteDefinition;
24use crate::lifecycle::application::runtime_bus::RuntimeBus;
25use crate::registry::RegistryError;
26use crate::shared::components::domain::Registry;
27use crate::shared::observability::domain::TracerConfig;
28use crate::startup_validation::ConfigCheck;
29use crate::template::TemplateRegistry;
30
31pub use crate::context_builder::CamelContextBuilder;
32
33/// The CamelContext is the runtime engine that manages components, routes, and their lifecycle.
34///
35/// # Lifecycle
36///
37/// Call [`start()`](Self::start) to launch routes, then [`stop()`](Self::stop)
38/// or [`abort()`](Self::abort) to shut down. A stopped context can be restarted
39/// by calling `start()` again — the controller actor stays alive across stop/start
40/// cycles; only [`abort()`](Self::abort) is destructive.
41pub struct CamelContext {
42    registry: Arc<std::sync::Mutex<Registry>>,
43    route_controller: RouteControllerHandle,
44    actor_join: Option<tokio::task::JoinHandle<()>>,
45    supervision_join: Option<tokio::task::JoinHandle<()>>,
46    runtime: Arc<RuntimeBus>,
47    cancel_token: CancellationToken,
48    metrics: Arc<dyn MetricsCollector>,
49    // Platform ports
50    platform_service: Arc<dyn PlatformService>,
51    languages: SharedLanguageRegistry,
52    shutdown_timeout: std::time::Duration,
53    services: Vec<Box<dyn Lifecycle>>,
54    health_registry: Arc<HealthCheckRegistry>,
55    component_configs: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
56    function_invoker: Option<Arc<dyn FunctionInvoker>>,
57    template_registry: Arc<TemplateRegistry>,
58    idempotent_repositories: crate::registry::SharedIdempotentRegistry,
59    claim_check_repositories: crate::registry::SharedClaimCheckRegistry,
60    cache_repositories: crate::registry::SharedCacheRegistry,
61    /// Fail-closed startup validation registry (ADR-0033). Checks are drained
62    /// and executed synchronously at the head of [`start()`](Self::start) before
63    /// any route consumer is started.
64    startup_checks: Vec<Box<dyn ConfigCheck>>,
65}
66
67/// Parts bag used by [`CamelContextBuilder::build`] to construct a [`CamelContext`]
68/// without accessing private fields from a sibling module.
69pub(crate) struct FromParts {
70    pub(crate) registry: Arc<std::sync::Mutex<Registry>>,
71    pub(crate) route_controller: RouteControllerHandle,
72    pub(crate) _actor_join: tokio::task::JoinHandle<()>,
73    pub(crate) supervision_join: Option<tokio::task::JoinHandle<()>>,
74    pub(crate) runtime: Arc<RuntimeBus>,
75    pub(crate) cancel_token: CancellationToken,
76    pub(crate) metrics: Arc<dyn MetricsCollector>,
77    pub(crate) platform_service: Arc<dyn PlatformService>,
78    pub(crate) languages: SharedLanguageRegistry,
79    pub(crate) shutdown_timeout: std::time::Duration,
80    pub(crate) services: Vec<Box<dyn Lifecycle>>,
81    pub(crate) health_registry: Arc<HealthCheckRegistry>,
82    pub(crate) component_configs: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
83    pub(crate) function_invoker: Option<Arc<dyn FunctionInvoker>>,
84    pub(crate) template_registry: Arc<TemplateRegistry>,
85    pub(crate) idempotent_repositories: crate::registry::SharedIdempotentRegistry,
86    pub(crate) claim_check_repositories: crate::registry::SharedClaimCheckRegistry,
87    pub(crate) cache_repositories: crate::registry::SharedCacheRegistry,
88    pub(crate) startup_checks: Vec<Box<dyn ConfigCheck>>,
89}
90
91impl CamelContext {
92    pub(crate) fn from_parts(parts: FromParts) -> Self {
93        Self {
94            registry: parts.registry,
95            route_controller: parts.route_controller,
96            actor_join: Some(parts._actor_join),
97            supervision_join: parts.supervision_join,
98            runtime: parts.runtime,
99            cancel_token: parts.cancel_token,
100            metrics: parts.metrics,
101            platform_service: parts.platform_service,
102            languages: parts.languages,
103            shutdown_timeout: parts.shutdown_timeout,
104            services: parts.services,
105            health_registry: parts.health_registry,
106            component_configs: parts.component_configs,
107            function_invoker: parts.function_invoker,
108            template_registry: parts.template_registry,
109            idempotent_repositories: parts.idempotent_repositories,
110            claim_check_repositories: parts.claim_check_repositories,
111            cache_repositories: parts.cache_repositories,
112            startup_checks: parts.startup_checks,
113        }
114    }
115}
116
117/// Opaque handle for runtime side-effect execution operations.
118///
119/// This intentionally does not expose direct lifecycle mutation APIs to callers.
120#[derive(Clone)]
121pub struct RuntimeExecutionHandle {
122    pub(crate) controller: RouteControllerHandle,
123    pub(crate) runtime: Arc<RuntimeBus>,
124    pub(crate) function_invoker: Option<Arc<dyn FunctionInvoker>>,
125    /// Lifecycle handles to inject into the compiled pipeline during
126    /// `apply_swap`.  Used in tests to simulate lifecycle-bearing routes
127    /// (e.g. resequencer).  Always `None` in production.
128    #[cfg(test)]
129    #[allow(clippy::type_complexity)]
130    pub(crate) test_lifecycle_inject: Arc<std::sync::Mutex<Option<Vec<Arc<dyn StepLifecycle>>>>>,
131}
132
133impl RuntimeExecutionHandle {
134    pub(crate) async fn add_route_definition(
135        &self,
136        definition: RouteDefinition,
137    ) -> Result<(), CamelError> {
138        use crate::lifecycle::application::ports::RouteRegistrationPort;
139        self.runtime
140            .register_route(definition)
141            .await
142            .map_err(Into::into)
143    }
144
145    /// Compile a route definition into a bare BoxProcessor (no lifecycle).
146    /// Kept for tests; hot-reload uses the lifecycle-preserving variant instead.
147    #[allow(dead_code)]
148    pub(crate) async fn compile_route_definition(
149        &self,
150        definition: RouteDefinition,
151    ) -> Result<camel_api::BoxProcessor, CamelError> {
152        self.controller.compile_route_definition(definition).await
153    }
154
155    #[allow(dead_code)] // kept for potential future hot-reload paths
156    pub(crate) async fn compile_route_definition_with_generation(
157        &self,
158        definition: RouteDefinition,
159        generation: u64,
160    ) -> Result<camel_api::BoxProcessor, CamelError> {
161        self.controller
162            .compile_route_definition_with_generation(definition, generation)
163            .await
164    }
165
166    pub(crate) async fn compile_route_definition_pipeline(
167        &self,
168        definition: RouteDefinition,
169        generation: u64,
170    ) -> Result<crate::lifecycle::domain::CompiledPipeline, CamelError> {
171        self.controller
172            .compile_route_definition_pipeline(definition, generation)
173            .await
174    }
175
176    /// Compile without function generation, returning full CompiledPipeline.
177    /// Oracle Fix 1: stateless hot-reload path preserves lifecycle handles.
178    pub(crate) async fn compile_route_definition_dry_pipeline(
179        &self,
180        definition: RouteDefinition,
181    ) -> Result<crate::lifecycle::domain::CompiledPipeline, CamelError> {
182        self.controller
183            .compile_route_definition_dry_pipeline(definition)
184            .await
185    }
186
187    pub(crate) async fn prepare_route_definition_with_generation(
188        &self,
189        definition: RouteDefinition,
190        generation: u64,
191    ) -> Result<crate::lifecycle::domain::route_compilation::PreparedRoute, CamelError> {
192        self.controller
193            .prepare_route_definition_with_generation(definition, generation)
194            .await
195    }
196
197    pub(crate) async fn insert_prepared_route(
198        &self,
199        prepared: crate::lifecycle::domain::route_compilation::PreparedRoute,
200    ) -> Result<(), CamelError> {
201        self.controller.insert_prepared_route(prepared).await
202    }
203
204    pub(crate) async fn discard_prepared_staging(&self, route_id: &str) -> Result<(), CamelError> {
205        self.controller.discard_prepared_staging(route_id).await
206    }
207
208    pub(crate) async fn remove_route_preserving_functions(
209        &self,
210        route_id: String,
211    ) -> Result<(), CamelError> {
212        self.controller
213            .remove_route_preserving_functions(route_id)
214            .await
215    }
216
217    pub(crate) async fn register_route_aggregate(
218        &self,
219        route_id: String,
220    ) -> Result<(), CamelError> {
221        self.runtime.register_aggregate_only(route_id).await
222    }
223
224    pub(crate) async fn swap_route_pipeline(
225        &self,
226        route_id: &str,
227        pipeline: camel_api::BoxProcessor,
228    ) -> Result<(), CamelError> {
229        self.controller.swap_pipeline(route_id, pipeline).await
230    }
231
232    /// Stop the route via the reload path (graceful lifecycle drain).
233    pub(crate) async fn stop_route_reload(&self, route_id: &str) -> Result<(), CamelError> {
234        self.controller.stop_route_reload(route_id).await
235    }
236
237    /// Start the route via the reload path (re-create consumer).
238    pub(crate) async fn start_route_reload(&self, route_id: &str) -> Result<(), CamelError> {
239        self.controller.start_route_reload(route_id).await
240    }
241
242    /// Raw pipeline swap — bypasses the lifecycle/aggregate rejection check.
243    /// Only safe after the route has been stopped (Restart path).
244    pub(crate) async fn swap_route_pipeline_raw(
245        &self,
246        route_id: &str,
247        pipeline: camel_api::BoxProcessor,
248        lifecycle: Vec<Arc<dyn camel_api::StepLifecycle>>,
249    ) -> Result<(), CamelError> {
250        self.controller
251            .swap_pipeline_raw(route_id, pipeline, lifecycle)
252            .await
253    }
254
255    pub(crate) async fn execute_runtime_command(
256        &self,
257        cmd: camel_api::RuntimeCommand,
258    ) -> Result<camel_api::RuntimeCommandResult, CamelError> {
259        self.runtime.execute(cmd).await
260    }
261
262    pub(crate) async fn runtime_route_status(
263        &self,
264        route_id: &str,
265    ) -> Result<Option<String>, CamelError> {
266        match self
267            .runtime
268            .ask(camel_api::RuntimeQuery::GetRouteStatus {
269                route_id: route_id.to_string(),
270            })
271            .await
272        {
273            Ok(camel_api::RuntimeQueryResult::RouteStatus { status, .. }) => Ok(Some(status)),
274            Ok(_) => Err(CamelError::RouteError(
275                "unexpected runtime query response for route status".to_string(),
276            )),
277            Err(CamelError::RouteError(msg)) if msg.contains("not found") => Ok(None),
278            Err(err) => Err(err),
279        }
280    }
281
282    pub(crate) async fn runtime_route_ids(&self) -> Result<Vec<String>, CamelError> {
283        match self.runtime.ask(camel_api::RuntimeQuery::ListRoutes).await {
284            Ok(camel_api::RuntimeQueryResult::Routes { route_ids }) => Ok(route_ids),
285            Ok(_) => Err(CamelError::RouteError(
286                "unexpected runtime query response for route listing".to_string(),
287            )),
288            Err(err) => Err(err),
289        }
290    }
291
292    pub(crate) async fn route_source_hash(&self, route_id: &str) -> Option<u64> {
293        self.controller.route_source_hash(route_id).await
294    }
295
296    pub(crate) async fn in_flight_count(&self, route_id: &str) -> Result<u64, CamelError> {
297        if !self.controller.route_exists(route_id).await? {
298            return Err(CamelError::RouteError(format!(
299                "Route '{}' not found",
300                route_id
301            )));
302        }
303        Ok(self
304            .controller
305            .in_flight_count(route_id)
306            .await?
307            .unwrap_or(0))
308    }
309
310    /// Check whether the running route has lifecycle-bearing steps.
311    pub(crate) async fn route_has_lifecycle(&self, route_id: &str) -> bool {
312        self.controller
313            .route_has_lifecycle(route_id)
314            .await
315            .unwrap_or(false)
316    }
317
318    pub(crate) fn function_invoker(&self) -> Option<Arc<dyn FunctionInvoker>> {
319        self.function_invoker.clone()
320    }
321
322    #[cfg(test)]
323    pub(crate) async fn force_start_route_for_test(
324        &self,
325        route_id: &str,
326    ) -> Result<(), CamelError> {
327        self.controller.start_route(route_id).await
328    }
329
330    pub async fn controller_route_count_for_test(&self) -> usize {
331        self.controller.route_count().await.unwrap_or(0)
332    }
333}
334
335#[async_trait::async_trait]
336impl crate::hot_reload::ports::ReloadExecutorPort for RuntimeExecutionHandle {
337    async fn add_route_definition(&self, definition: RouteDefinition) -> Result<(), CamelError> {
338        RuntimeExecutionHandle::add_route_definition(self, definition).await
339    }
340
341    async fn compile_route_definition_pipeline(
342        &self,
343        definition: RouteDefinition,
344        generation: u64,
345    ) -> Result<crate::lifecycle::domain::CompiledPipeline, CamelError> {
346        RuntimeExecutionHandle::compile_route_definition_pipeline(self, definition, generation)
347            .await
348    }
349
350    async fn compile_route_definition_dry_pipeline(
351        &self,
352        definition: RouteDefinition,
353    ) -> Result<crate::lifecycle::domain::CompiledPipeline, CamelError> {
354        RuntimeExecutionHandle::compile_route_definition_dry_pipeline(self, definition).await
355    }
356
357    async fn prepare_route_definition_with_generation(
358        &self,
359        definition: RouteDefinition,
360        generation: u64,
361    ) -> Result<crate::lifecycle::domain::route_compilation::PreparedRoute, CamelError> {
362        RuntimeExecutionHandle::prepare_route_definition_with_generation(
363            self, definition, generation,
364        )
365        .await
366    }
367
368    async fn insert_prepared_route(
369        &self,
370        prepared: crate::lifecycle::domain::route_compilation::PreparedRoute,
371    ) -> Result<(), CamelError> {
372        RuntimeExecutionHandle::insert_prepared_route(self, prepared).await
373    }
374
375    async fn discard_prepared_staging(&self, route_id: &str) -> Result<(), CamelError> {
376        RuntimeExecutionHandle::discard_prepared_staging(self, route_id).await
377    }
378
379    async fn remove_route_preserving_functions(&self, route_id: String) -> Result<(), CamelError> {
380        RuntimeExecutionHandle::remove_route_preserving_functions(self, route_id).await
381    }
382
383    async fn register_route_aggregate(&self, route_id: String) -> Result<(), CamelError> {
384        RuntimeExecutionHandle::register_route_aggregate(self, route_id).await
385    }
386
387    async fn swap_route_pipeline(
388        &self,
389        route_id: &str,
390        pipeline: camel_api::BoxProcessor,
391    ) -> Result<(), CamelError> {
392        RuntimeExecutionHandle::swap_route_pipeline(self, route_id, pipeline).await
393    }
394
395    async fn stop_route_reload(&self, route_id: &str) -> Result<(), CamelError> {
396        RuntimeExecutionHandle::stop_route_reload(self, route_id).await
397    }
398
399    async fn start_route_reload(&self, route_id: &str) -> Result<(), CamelError> {
400        RuntimeExecutionHandle::start_route_reload(self, route_id).await
401    }
402
403    async fn swap_route_pipeline_raw(
404        &self,
405        route_id: &str,
406        pipeline: camel_api::BoxProcessor,
407        lifecycle: Vec<std::sync::Arc<dyn camel_api::StepLifecycle>>,
408    ) -> Result<(), CamelError> {
409        RuntimeExecutionHandle::swap_route_pipeline_raw(self, route_id, pipeline, lifecycle).await
410    }
411
412    async fn execute_runtime_command(
413        &self,
414        cmd: camel_api::RuntimeCommand,
415    ) -> Result<camel_api::RuntimeCommandResult, CamelError> {
416        RuntimeExecutionHandle::execute_runtime_command(self, cmd).await
417    }
418
419    async fn runtime_route_status(&self, route_id: &str) -> Result<Option<String>, CamelError> {
420        RuntimeExecutionHandle::runtime_route_status(self, route_id).await
421    }
422
423    async fn in_flight_count(&self, route_id: &str) -> Result<u64, CamelError> {
424        RuntimeExecutionHandle::in_flight_count(self, route_id).await
425    }
426
427    async fn route_has_lifecycle(&self, route_id: &str) -> bool {
428        RuntimeExecutionHandle::route_has_lifecycle(self, route_id).await
429    }
430
431    #[cfg(test)]
432    fn take_test_lifecycle_inject(
433        &self,
434    ) -> Option<Vec<std::sync::Arc<dyn camel_api::StepLifecycle>>> {
435        self.test_lifecycle_inject.lock().unwrap().take()
436    }
437}
438
439impl CamelContext {
440    pub fn builder() -> CamelContextBuilder {
441        CamelContextBuilder::new()
442    }
443
444    /// Set a global error handler applied to all routes without a per-route handler.
445    pub async fn set_error_handler(&mut self, config: ErrorHandlerConfig) {
446        let _ = self.route_controller.set_error_handler(config).await;
447    }
448
449    /// Install per-bind public-exposure acknowledgements (ADR-0061).
450    /// Built by the CLI from `CamelConfig.binds`; the per-bind gate fails
451    /// closed on non-loopback binds until acknowledged.
452    pub async fn set_bind_exposure_acks(
453        &mut self,
454        acks: crate::lifecycle::adapters::route_controller_trait::BindExposureAcks,
455    ) {
456        let _ = self.route_controller.set_bind_exposure_acks(acks).await;
457    }
458
459    /// Enable or disable tracing globally.
460    pub async fn set_tracing(&mut self, enabled: bool) {
461        let _ = self
462            .route_controller
463            .set_tracer_config(TracerConfig {
464                enabled,
465                ..Default::default()
466            })
467            .await;
468    }
469
470    /// Configure tracing with full config.
471    pub async fn set_tracer_config(&mut self, config: TracerConfig) {
472        // Inject metrics collector if not already set
473        let config = if config.metrics_collector.is_none() {
474            TracerConfig {
475                metrics_collector: Some(Arc::clone(&self.metrics)),
476                ..config
477            }
478        } else {
479            config
480        };
481
482        let _ = self.route_controller.set_tracer_config(config).await;
483    }
484
485    /// Builder-style: enable tracing with default config.
486    pub async fn with_tracing(mut self) -> Self {
487        self.set_tracing(true).await;
488        self
489    }
490
491    /// Builder-style: configure tracing with custom config.
492    /// Note: tracing subscriber initialization (stdout/file output) is handled
493    /// separately via init_tracing_subscriber (called in camel-config bridge).
494    pub async fn with_tracer_config(mut self, config: TracerConfig) -> Self {
495        self.set_tracer_config(config).await;
496        self
497    }
498
499    /// Register a lifecycle service (Apache Camel: addService pattern)
500    ///
501    /// For services exposing `as_function_invoker()`, the invoker is propagated
502    /// to the route controller so that subsequent route definitions with function
503    /// steps work correctly.
504    ///
505    /// Prefer [`CamelContextBuilder::with_lifecycle`] when possible, which wires
506    /// the invoker at build time before any routes are added.
507    pub fn with_lifecycle<L: Lifecycle + 'static>(mut self, service: L) -> Self {
508        if let Some(collector) = service.as_metrics_collector() {
509            self.metrics = collector;
510        }
511        if let Some(invoker) = service.as_function_invoker() {
512            self.function_invoker = Some(invoker.clone());
513            if let Err(e) = self.route_controller.try_set_function_invoker(invoker) {
514                tracing::debug!("Failed to propagate function invoker to route controller: {e}");
515            }
516        }
517
518        self.services.push(Box::new(service));
519        self
520    }
521
522    /// Register a component with this context.
523    ///
524    /// Delegates to [`register_component_dyn`](Self::register_component_dyn)
525    /// so metadata harvesting happens regardless of which entry point
526    /// is used.
527    pub fn register_component<C: Component + 'static>(&mut self, component: C) {
528        self.register_component_dyn(Arc::new(component));
529    }
530
531    /// Register a startup `ConfigCheck` to be evaluated at the head of
532    /// [`start()`](Self::start). Established by ADR-0033.
533    ///
534    /// Checks are drained and executed synchronously before any route consumer
535    /// is started. If any check returns `Err`, `start()` fails closed with
536    /// `CamelError::Config(_)` and no route is started. The check list is
537    /// consumed (moved) during `start()` so this method may be called multiple
538    /// times to register an arbitrary number of checks.
539    pub fn add_startup_check(&mut self, check: Box<dyn ConfigCheck>) {
540        self.startup_checks.push(check);
541    }
542
543    /// Register a language with this context, keyed by name.
544    ///
545    /// Returns `Err(LanguageRegistryError::AlreadyRegistered)` if a language
546    /// with the same name is already registered. Use
547    /// [`resolve_language`](Self::resolve_language) to check before
548    /// registering, or choose a distinct name.
549    pub fn register_language(
550        &mut self,
551        name: impl Into<String>,
552        lang: Box<dyn Language>,
553    ) -> Result<(), LanguageRegistryError> {
554        let name = name.into();
555        let mut languages = self
556            .languages
557            .lock()
558            .expect("mutex poisoned: another thread panicked while holding this lock"); // allow-unwrap
559        if languages.contains_key(&name) {
560            return Err(LanguageRegistryError::AlreadyRegistered { name });
561        }
562        languages.insert(name, Arc::from(lang));
563        Ok(())
564    }
565
566    /// Resolve a language by name. Returns `None` if not registered.
567    pub fn resolve_language(&self, name: &str) -> Option<Arc<dyn Language>> {
568        let languages = self
569            .languages
570            .lock()
571            .expect("mutex poisoned: another thread panicked while holding this lock"); // allow-unwrap
572        languages.get(name).cloned()
573    }
574
575    /// Add a route definition to this context.
576    ///
577    /// The route must have an ID. Steps are resolved immediately using registered components.
578    pub async fn add_route_definition(
579        &self,
580        definition: RouteDefinition,
581    ) -> Result<(), CamelError> {
582        use crate::lifecycle::application::ports::RouteRegistrationPort;
583        debug!(
584            from = definition.from_uri(),
585            route_id = %definition.route_id(),
586            "Adding route definition"
587        );
588        self.runtime
589            .register_route(definition)
590            .await
591            .map_err(Into::into)
592    }
593
594    /// Access the component registry.
595    pub fn registry(&self) -> std::sync::MutexGuard<'_, Registry> {
596        self.registry
597            .lock()
598            .expect("mutex poisoned: another thread panicked while holding this lock") // allow-unwrap
599    }
600
601    /// Access the shared component registry Arc.
602    pub fn registry_arc(&self) -> Arc<std::sync::Mutex<Registry>> {
603        Arc::clone(&self.registry)
604    }
605
606    /// Get runtime execution handle for file-watcher integrations.
607    pub fn runtime_execution_handle(&self) -> RuntimeExecutionHandle {
608        RuntimeExecutionHandle {
609            controller: self.route_controller.clone(),
610            runtime: Arc::clone(&self.runtime),
611            function_invoker: self.function_invoker.clone(),
612            #[cfg(test)]
613            test_lifecycle_inject: Arc::new(std::sync::Mutex::new(None)),
614        }
615    }
616
617    /// Get the metrics collector.
618    pub fn metrics(&self) -> Arc<dyn MetricsCollector> {
619        Arc::clone(&self.metrics)
620    }
621
622    /// Get the platform service.
623    pub fn platform_service(&self) -> Arc<dyn PlatformService> {
624        Arc::clone(&self.platform_service)
625    }
626
627    /// Get the readiness gate port.
628    pub fn readiness_gate(&self) -> Arc<dyn ReadinessGate> {
629        self.platform_service.readiness_gate()
630    }
631
632    /// Get the platform identity.
633    pub fn platform_identity(&self) -> PlatformIdentity {
634        self.platform_service.identity()
635    }
636
637    /// Get the leadership service port.
638    pub fn leadership(&self) -> Arc<dyn camel_api::LeadershipService> {
639        self.platform_service.leadership()
640    }
641
642    /// Get runtime command/query bus handle.
643    pub fn runtime(&self) -> Arc<dyn camel_api::RuntimeHandle> {
644        self.runtime.clone()
645    }
646
647    /// Build a producer context wired to this runtime.
648    pub fn producer_context(&self) -> camel_api::ProducerContext {
649        camel_api::ProducerContext::new().with_runtime(self.runtime())
650    }
651
652    /// Query route status via runtime read-model.
653    pub async fn runtime_route_status(&self, route_id: &str) -> Result<Option<String>, CamelError> {
654        match self
655            .runtime()
656            .ask(camel_api::RuntimeQuery::GetRouteStatus {
657                route_id: route_id.to_string(),
658            })
659            .await
660        {
661            Ok(camel_api::RuntimeQueryResult::RouteStatus { status, .. }) => Ok(Some(status)),
662            Ok(_) => Err(CamelError::RouteError(
663                "unexpected runtime query response for route status".to_string(),
664            )),
665            Err(CamelError::RouteError(msg)) if msg.contains("not found") => Ok(None),
666            Err(err) => Err(err),
667        }
668    }
669
670    /// Start all routes. Each route's consumer will begin producing exchanges.
671    ///
672    /// Only routes with `auto_startup == true` will be started, in order of their
673    /// `startup_order` (lower values start first).
674    ///
675    /// Algorithm lives in `lifecycle::application::context_lifecycle::start_context`
676    /// (Tier C C2). Public signature is unchanged.
677    pub async fn start(&mut self) -> Result<(), CamelError> {
678        crate::lifecycle::application::context_lifecycle::start_context(
679            &mut self.services,
680            &mut self.startup_checks,
681            &self.runtime,
682            &self.route_controller,
683            &mut self.cancel_token,
684        )
685        .await
686    }
687
688    /// Graceful shutdown with default 30-second timeout.
689    pub async fn stop(&mut self) -> Result<(), CamelError> {
690        self.stop_timeout(self.shutdown_timeout).await
691    }
692
693    /// Graceful shutdown with custom timeout.
694    ///
695    /// Note: The timeout parameter is currently not propagated to the
696    /// RouteController's per-route shutdown timeout. The RouteController
697    /// uses a hardcoded 5-second default (`DEFAULT_SHUTDOWN_TIMEOUT`).
698    /// Full propagation is planned for a future version.
699    ///
700    /// Algorithm lives in `lifecycle::application::context_lifecycle::stop_context`
701    /// (Tier C C2). Public signature is unchanged.
702    pub async fn stop_timeout(&mut self, _timeout: std::time::Duration) -> Result<(), CamelError> {
703        crate::lifecycle::application::context_lifecycle::stop_context(
704            &self.cancel_token,
705            &mut self.supervision_join,
706            &self.runtime,
707            &self.route_controller,
708            &mut self.services,
709        )
710        .await
711    }
712
713    /// Get the graceful shutdown timeout used by [`stop()`](Self::stop).
714    pub fn shutdown_timeout(&self) -> std::time::Duration {
715        self.shutdown_timeout
716    }
717
718    /// Set the graceful shutdown timeout used by [`stop()`](Self::stop).
719    pub fn set_shutdown_timeout(&mut self, timeout: std::time::Duration) {
720        self.shutdown_timeout = timeout;
721    }
722
723    /// Test-only: take the actor join handle out of the context.
724    /// Used to verify the actor exits gracefully after stop().
725    #[cfg(test)]
726    pub(crate) fn take_actor_join(&mut self) -> Option<tokio::task::JoinHandle<()>> {
727        self.actor_join.take()
728    }
729
730    /// Immediate abort — kills all tasks without draining.
731    ///
732    /// Algorithm lives in `lifecycle::application::context_lifecycle::abort_context`
733    /// (Tier C C2). Public signature is unchanged. The use-case routes
734    /// through `RouteOrderingPort` + `RouteDestructiveTeardownPort`; the
735    /// same underlying `RouteControllerHandle` is passed twice as both
736    /// trait objects.
737    pub async fn abort(&mut self) {
738        crate::lifecycle::application::context_lifecycle::abort_context(
739            &self.cancel_token,
740            &mut self.supervision_join,
741            &self.runtime,
742            &self.route_controller as &dyn crate::lifecycle::application::ports::RouteOrderingPort,
743            &self.route_controller
744                as &dyn crate::lifecycle::application::ports::RouteDestructiveTeardownPort,
745            &mut self.services,
746            self.health_registry.cancel_token(),
747            &mut self.actor_join,
748        )
749        .await
750    }
751
752    /// Check health status of all registered services and lifecycle services.
753    pub async fn health_check(&self) -> HealthReport {
754        use camel_api::HealthSource;
755        self.health_report().await
756    }
757
758    pub fn health_registry(&self) -> Arc<HealthCheckRegistry> {
759        Arc::clone(&self.health_registry)
760    }
761
762    /// Store a component config. Overwrites any previously stored config of the same type.
763    pub fn set_component_config<T: 'static + Send + Sync>(&mut self, config: T) {
764        self.component_configs
765            .insert(TypeId::of::<T>(), Box::new(config));
766    }
767
768    /// Retrieve a stored component config by type. Returns None if not stored.
769    pub fn get_component_config<T: 'static + Send + Sync>(&self) -> Option<&T> {
770        self.component_configs
771            .get(&TypeId::of::<T>())
772            .and_then(|b| b.downcast_ref::<T>())
773    }
774
775    // --- Component Metadata ---
776
777    /// Get a component's harvested metadata by URI scheme.
778    pub fn component_metadata(&self, scheme: &str) -> Option<ComponentMetadata> {
779        self.registry.lock().ok()?.get_metadata(scheme)
780    }
781
782    /// Get metadata for every registered component.
783    pub fn all_component_metadata(&self) -> Vec<ComponentMetadata> {
784        self.registry
785            .lock()
786            .expect("mutex poisoned: another thread panicked while holding this lock") // allow-unwrap
787            .all_metadata()
788    }
789
790    /// Get a metadata catalog handle implementing
791    /// [`ComponentMetadataCatalog`](camel_api::component_metadata::ComponentMetadataCatalog).
792    ///
793    /// The returned handle shares the same `Arc<Mutex<Registry>>` as the
794    /// context, so registrations made through one are visible through the
795    /// other.
796    pub fn metadata_catalog(
797        &self,
798    ) -> crate::component_metadata_catalog::RuntimeComponentMetadataCatalog {
799        crate::component_metadata_catalog::RuntimeComponentMetadataCatalog::new(Arc::clone(
800            &self.registry,
801        ))
802    }
803
804    // --- Route Template Registry (data-only) ---
805
806    /// Register a route template specification.
807    ///
808    /// Returns `Err(CamelError)` if a template with the same ID is already registered.
809    pub fn add_route_template(&self, spec: RouteTemplateSpec) -> Result<(), CamelError> {
810        self.template_registry.register(spec)
811    }
812
813    /// Retrieve a route template specification by its ID.
814    pub fn get_route_template(&self, id: &str) -> Option<RouteTemplateSpec> {
815        self.template_registry.get(id)
816    }
817
818    /// Return all registered template IDs.
819    pub fn template_ids(&self) -> Vec<String> {
820        self.template_registry.template_ids()
821    }
822
823    /// Record a newly instantiated template instance.
824    pub fn record_template_instance(&self, record: TemplateInstanceRecord) {
825        self.template_registry.record_instance(record)
826    }
827
828    /// Return all instance records for a given template ID.
829    pub fn template_instances(&self, template_id: &str) -> Vec<TemplateInstanceRecord> {
830        self.template_registry.instances(template_id)
831    }
832
833    // --- Idempotent Repository Registry ---
834
835    /// Register an idempotent repository.
836    ///
837    /// Returns `Err(RegistryError::AlreadyRegistered)` if a repository with
838    /// the same name is already registered.
839    pub fn register_idempotent_repository(
840        &mut self,
841        name: impl Into<String>,
842        repo: Arc<dyn camel_api::IdempotentRepository>,
843    ) -> Result<(), RegistryError> {
844        self.idempotent_repositories.register(name, repo)
845    }
846
847    /// Retrieve an idempotent repository by name.
848    pub fn idempotent_repository(
849        &self,
850        name: &str,
851    ) -> Option<Arc<dyn camel_api::IdempotentRepository>> {
852        self.idempotent_repositories.get(name)
853    }
854
855    // --- Claim Check Repository Registry ---
856
857    /// Register a claim check repository.
858    ///
859    /// Returns `Err(RegistryError::AlreadyRegistered)` if a repository with
860    /// the same name is already registered.
861    pub fn register_claim_check_repository(
862        &mut self,
863        name: impl Into<String>,
864        repo: Arc<dyn camel_api::ClaimCheckRepository>,
865    ) -> Result<(), RegistryError> {
866        self.claim_check_repositories.register(name, repo)
867    }
868
869    /// Retrieve a claim check repository by name.
870    pub fn claim_check_repository(
871        &self,
872        name: &str,
873    ) -> Option<Arc<dyn camel_api::ClaimCheckRepository>> {
874        self.claim_check_repositories.get(name)
875    }
876
877    // --- Cache Repository Registry ---
878
879    /// Register a cache repository.
880    ///
881    /// Returns `Err(RegistryError::AlreadyRegistered)` if a repository with
882    /// the same name is already registered.
883    pub fn register_cache_repository(
884        &mut self,
885        name: impl Into<String>,
886        repo: Arc<dyn camel_api::CacheRepository>,
887    ) -> Result<(), RegistryError> {
888        self.cache_repositories.register(name, repo)
889    }
890
891    /// Replace an existing cache repository, returning the evicted value.
892    ///
893    /// Returns `None` if no repository was registered under `name`.
894    pub fn replace_cache_repository(
895        &mut self,
896        name: impl Into<String>,
897        repo: Arc<dyn camel_api::CacheRepository>,
898    ) -> Option<Arc<dyn camel_api::CacheRepository>> {
899        self.cache_repositories.register_or_replace(name, repo)
900    }
901
902    /// Retrieve a cache repository by name.
903    pub fn cache_repository(&self, name: &str) -> Option<Arc<dyn camel_api::CacheRepository>> {
904        self.cache_repositories.get(name)
905    }
906
907    /// Access the shutdown cancellation token.
908    ///
909    /// Repositories that need to bind background sweep tasks to context
910    /// shutdown (e.g. `RedbCacheRepository`) can `child_token()` from this.
911    pub fn shutdown_token(&self) -> CancellationToken {
912        self.cancel_token.clone()
913    }
914}
915
916impl ComponentRegistrar for CamelContext {
917    fn register_component_dyn(&mut self, component: Arc<dyn Component>) {
918        let scheme = component.scheme().to_string();
919        self.registry
920            .lock()
921            .expect("mutex poisoned: another thread panicked while holding this lock") // allow-unwrap
922            .register(component);
923        trace!(scheme, "Registered component");
924    }
925}
926
927impl ComponentContext for CamelContext {
928    fn resolve_component(&self, scheme: &str) -> Option<Arc<dyn Component>> {
929        self.registry.lock().ok()?.get(scheme)
930    }
931
932    fn resolve_language(&self, name: &str) -> Option<Arc<dyn Language>> {
933        self.languages.lock().ok()?.get(name).cloned()
934    }
935
936    fn metrics(&self) -> Arc<dyn MetricsCollector> {
937        Arc::clone(&self.metrics)
938    }
939
940    fn health(&self) -> Arc<dyn camel_component_api::HealthCheckRegistry> {
941        // The concrete HealthCheckRegistry struct implements the trait via
942        // the impl added in health_registry.rs.
943        Arc::clone(&self.health_registry) as Arc<dyn camel_component_api::HealthCheckRegistry>
944    }
945
946    fn platform_service(&self) -> Arc<dyn PlatformService> {
947        Arc::clone(&self.platform_service)
948    }
949
950    fn register_route_health_check(
951        &self,
952        route_id: &str,
953        check: Arc<dyn camel_api::AsyncHealthCheck>,
954    ) {
955        self.health_registry.register_for_route(route_id, check);
956    }
957
958    fn unregister_route_health_check(&self, route_id: &str) {
959        self.health_registry.unregister_for_route(route_id);
960    }
961}
962
963#[async_trait::async_trait]
964impl camel_api::HealthSource for CamelContext {
965    async fn liveness(&self) -> camel_api::HealthStatus {
966        let has_failed = self
967            .services
968            .iter()
969            .any(|s| s.status() == camel_api::ServiceStatus::Failed);
970        if has_failed {
971            camel_api::HealthStatus::Unhealthy
972        } else {
973            camel_api::HealthStatus::Healthy
974        }
975    }
976
977    async fn readiness(&self) -> camel_api::HealthStatus {
978        let has_failed = self
979            .services
980            .iter()
981            .any(|s| s.status() == camel_api::ServiceStatus::Failed);
982        if has_failed {
983            return camel_api::HealthStatus::Unhealthy;
984        }
985        let has_stopped = self
986            .services
987            .iter()
988            .any(|s| s.status() == camel_api::ServiceStatus::Stopped);
989        if has_stopped {
990            return camel_api::HealthStatus::Degraded;
991        }
992        self.health_registry.check_all().await.status
993    }
994
995    async fn health_report(&self) -> camel_api::HealthReport {
996        let mut report = self.health_registry.check_all().await;
997        let mut worst = report.status;
998        for service in &self.services {
999            let svc_status = service.status();
1000            let health = match svc_status {
1001                camel_api::ServiceStatus::Started => camel_api::HealthStatus::Healthy,
1002                camel_api::ServiceStatus::Stopped => camel_api::HealthStatus::Degraded,
1003                camel_api::ServiceStatus::Failed => camel_api::HealthStatus::Unhealthy,
1004                // Forward-safe fail-closed: an unknown future ServiceStatus keeps
1005                // the pod NotReady (no traffic) rather than guessing Degraded.
1006                _ => camel_api::HealthStatus::Unhealthy,
1007            };
1008            if matches!(worst, camel_api::HealthStatus::Healthy)
1009                && matches!(
1010                    health,
1011                    camel_api::HealthStatus::Degraded | camel_api::HealthStatus::Unhealthy
1012                )
1013            {
1014                worst = health;
1015            }
1016            if matches!(worst, camel_api::HealthStatus::Degraded)
1017                && matches!(health, camel_api::HealthStatus::Unhealthy)
1018            {
1019                worst = health;
1020            }
1021            report.services.push(camel_api::ServiceHealth {
1022                name: service.name().to_string(),
1023                status: svc_status,
1024                message: None,
1025            });
1026        }
1027        report.status = worst;
1028        report
1029    }
1030
1031    async fn startup(&self) -> camel_api::HealthStatus {
1032        camel_api::HealthStatus::Healthy
1033    }
1034}
1035
1036#[cfg(test)]
1037#[path = "context_tests.rs"]
1038mod context_tests;