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
33pub 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_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 startup_checks: Vec<Box<dyn ConfigCheck>>,
65}
66
67pub(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#[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 #[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 #[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)] 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 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 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 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 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 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 pub async fn set_error_handler(&mut self, config: ErrorHandlerConfig) {
446 let _ = self.route_controller.set_error_handler(config).await;
447 }
448
449 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 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 pub async fn set_tracer_config(&mut self, config: TracerConfig) {
472 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 pub async fn with_tracing(mut self) -> Self {
487 self.set_tracing(true).await;
488 self
489 }
490
491 pub async fn with_tracer_config(mut self, config: TracerConfig) -> Self {
495 self.set_tracer_config(config).await;
496 self
497 }
498
499 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 pub fn register_component<C: Component + 'static>(&mut self, component: C) {
528 self.register_component_dyn(Arc::new(component));
529 }
530
531 pub fn add_startup_check(&mut self, check: Box<dyn ConfigCheck>) {
540 self.startup_checks.push(check);
541 }
542
543 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"); if languages.contains_key(&name) {
560 return Err(LanguageRegistryError::AlreadyRegistered { name });
561 }
562 languages.insert(name, Arc::from(lang));
563 Ok(())
564 }
565
566 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"); languages.get(name).cloned()
573 }
574
575 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 pub fn registry(&self) -> std::sync::MutexGuard<'_, Registry> {
596 self.registry
597 .lock()
598 .expect("mutex poisoned: another thread panicked while holding this lock") }
600
601 pub fn registry_arc(&self) -> Arc<std::sync::Mutex<Registry>> {
603 Arc::clone(&self.registry)
604 }
605
606 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 pub fn metrics(&self) -> Arc<dyn MetricsCollector> {
619 Arc::clone(&self.metrics)
620 }
621
622 pub fn platform_service(&self) -> Arc<dyn PlatformService> {
624 Arc::clone(&self.platform_service)
625 }
626
627 pub fn readiness_gate(&self) -> Arc<dyn ReadinessGate> {
629 self.platform_service.readiness_gate()
630 }
631
632 pub fn platform_identity(&self) -> PlatformIdentity {
634 self.platform_service.identity()
635 }
636
637 pub fn leadership(&self) -> Arc<dyn camel_api::LeadershipService> {
639 self.platform_service.leadership()
640 }
641
642 pub fn runtime(&self) -> Arc<dyn camel_api::RuntimeHandle> {
644 self.runtime.clone()
645 }
646
647 pub fn producer_context(&self) -> camel_api::ProducerContext {
649 camel_api::ProducerContext::new().with_runtime(self.runtime())
650 }
651
652 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 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 pub async fn stop(&mut self) -> Result<(), CamelError> {
690 self.stop_timeout(self.shutdown_timeout).await
691 }
692
693 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 pub fn shutdown_timeout(&self) -> std::time::Duration {
715 self.shutdown_timeout
716 }
717
718 pub fn set_shutdown_timeout(&mut self, timeout: std::time::Duration) {
720 self.shutdown_timeout = timeout;
721 }
722
723 #[cfg(test)]
726 pub(crate) fn take_actor_join(&mut self) -> Option<tokio::task::JoinHandle<()>> {
727 self.actor_join.take()
728 }
729
730 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 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 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 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 pub fn component_metadata(&self, scheme: &str) -> Option<ComponentMetadata> {
779 self.registry.lock().ok()?.get_metadata(scheme)
780 }
781
782 pub fn all_component_metadata(&self) -> Vec<ComponentMetadata> {
784 self.registry
785 .lock()
786 .expect("mutex poisoned: another thread panicked while holding this lock") .all_metadata()
788 }
789
790 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 pub fn add_route_template(&self, spec: RouteTemplateSpec) -> Result<(), CamelError> {
810 self.template_registry.register(spec)
811 }
812
813 pub fn get_route_template(&self, id: &str) -> Option<RouteTemplateSpec> {
815 self.template_registry.get(id)
816 }
817
818 pub fn template_ids(&self) -> Vec<String> {
820 self.template_registry.template_ids()
821 }
822
823 pub fn record_template_instance(&self, record: TemplateInstanceRecord) {
825 self.template_registry.record_instance(record)
826 }
827
828 pub fn template_instances(&self, template_id: &str) -> Vec<TemplateInstanceRecord> {
830 self.template_registry.instances(template_id)
831 }
832
833 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 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 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 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 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 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 pub fn cache_repository(&self, name: &str) -> Option<Arc<dyn camel_api::CacheRepository>> {
904 self.cache_repositories.get(name)
905 }
906
907 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") .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 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 _ => 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;