camel_core/
context_builder.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3use tokio_util::sync::CancellationToken;
4
5use camel_api::{
6    CamelError, FunctionInvoker, Lifecycle, MetricsCollector, NoOpMetrics, NoopPlatformService,
7    PlatformService, SupervisionConfig,
8};
9use camel_language_api::Language;
10
11use super::context::{CamelContext, FromParts};
12use crate::cache::memory::MemoryCacheRepository;
13use crate::claim_check::memory_repository::MemoryClaimCheckRepository;
14use crate::health_registry::HealthCheckRegistry;
15use crate::idempotent::memory_repository::MemoryIdempotentRepository;
16use crate::lifecycle::adapters::RuntimeExecutionAdapter;
17use crate::lifecycle::adapters::controller_actor::{
18    RouteControllerHandle, spawn_controller_actor, spawn_supervision_task,
19};
20use crate::lifecycle::adapters::route_controller::{
21    DefaultRouteController, SharedLanguageRegistry,
22};
23use crate::lifecycle::application::ports::RuntimeExecutionPort;
24use crate::lifecycle::application::runtime_bus::RuntimeBus;
25use crate::registry::{CacheRegistry, ClaimCheckRegistry, IdempotentRegistry};
26use crate::shared::components::domain::Registry;
27use crate::startup_validation::ConfigCheck;
28use crate::template::TemplateRegistry;
29
30type ExecutionFactory =
31    Arc<dyn Fn(RouteControllerHandle) -> Arc<dyn RuntimeExecutionPort> + Send + Sync>;
32
33pub struct CamelContextBuilder {
34    registry: Option<Arc<std::sync::Mutex<Registry>>>,
35    languages: Option<SharedLanguageRegistry>,
36    metrics: Option<Arc<dyn MetricsCollector>>,
37    // Platform ports
38    platform_service: Option<Arc<dyn PlatformService>>,
39    supervision_config: Option<SupervisionConfig>,
40    runtime_store: Option<crate::lifecycle::adapters::InMemoryRuntimeStore>,
41    shutdown_timeout: std::time::Duration,
42    beans: Option<Arc<std::sync::Mutex<camel_bean::BeanRegistry>>>,
43    function_invoker: Option<Arc<dyn FunctionInvoker>>,
44    lifecycle_services: Vec<Box<dyn Lifecycle>>,
45    execution_factory: Option<ExecutionFactory>,
46    health_registry: Option<Arc<HealthCheckRegistry>>,
47    template_registry: Option<Arc<TemplateRegistry>>,
48}
49
50impl CamelContextBuilder {
51    pub fn new() -> Self {
52        Self {
53            registry: None,
54            languages: None,
55            metrics: None,
56            platform_service: None,
57            supervision_config: None,
58            runtime_store: None,
59            shutdown_timeout: std::time::Duration::from_secs(5),
60            beans: None,
61            function_invoker: None,
62            lifecycle_services: Vec::new(),
63            execution_factory: None,
64            health_registry: None,
65            template_registry: None,
66        }
67    }
68
69    pub fn registry(mut self, registry: Arc<std::sync::Mutex<Registry>>) -> Self {
70        self.registry = Some(registry);
71        self
72    }
73
74    pub fn languages(mut self, languages: SharedLanguageRegistry) -> Self {
75        self.languages = Some(languages);
76        self
77    }
78
79    pub fn with_execution_factory(
80        mut self,
81        factory: impl Fn(RouteControllerHandle) -> Arc<dyn RuntimeExecutionPort> + Send + Sync + 'static,
82    ) -> Self {
83        self.execution_factory = Some(Arc::new(factory));
84        self
85    }
86
87    pub fn metrics(mut self, metrics: Arc<dyn MetricsCollector>) -> Self {
88        self.metrics = Some(metrics);
89        self
90    }
91
92    /// Set a custom platform service.
93    pub fn platform_service(mut self, platform_service: Arc<dyn PlatformService>) -> Self {
94        self.platform_service = Some(platform_service);
95        self
96    }
97
98    pub fn supervision(mut self, config: SupervisionConfig) -> Self {
99        self.supervision_config = Some(config);
100        self
101    }
102
103    pub fn runtime_store(
104        mut self,
105        store: crate::lifecycle::adapters::InMemoryRuntimeStore,
106    ) -> Self {
107        self.runtime_store = Some(store);
108        self
109    }
110
111    pub fn shutdown_timeout(mut self, timeout: std::time::Duration) -> Self {
112        self.shutdown_timeout = timeout;
113        self
114    }
115
116    pub fn health_registry(mut self, registry: Arc<HealthCheckRegistry>) -> Self {
117        self.health_registry = Some(registry);
118        self
119    }
120
121    /// Inject a shared `BeanRegistry` for bean resolution across routes.
122    pub fn beans(mut self, beans: Arc<std::sync::Mutex<camel_bean::BeanRegistry>>) -> Self {
123        self.beans = Some(beans);
124        self
125    }
126
127    /// Register a lifecycle service (e.g., FunctionRuntimeService) at builder time.
128    ///
129    /// This is the recommended path for services that need to be wired into the
130    /// route controller before any routes are added. The function invoker (if any)
131    /// is extracted and passed to the `DefaultRouteController` during `build()`.
132    pub fn with_lifecycle<L: Lifecycle + 'static>(mut self, service: L) -> Self {
133        if let Some(collector) = service.as_metrics_collector() {
134            self.metrics = Some(collector);
135        }
136        if let Some(invoker) = service.as_function_invoker() {
137            self.function_invoker = Some(invoker);
138        }
139        self.lifecycle_services.push(Box::new(service));
140        self
141    }
142
143    /// Set a custom `TemplateRegistry` for route template storage.
144    ///
145    /// If not provided, a default empty registry is created during `build()`.
146    pub fn template_registry(mut self, registry: Arc<TemplateRegistry>) -> Self {
147        self.template_registry = Some(registry);
148        self
149    }
150
151    fn built_in_languages() -> SharedLanguageRegistry {
152        crate::language_registry::from_config(&camel_language_api::LanguagesConfig::default())
153    }
154
155    fn build_runtime(
156        controller: RouteControllerHandle,
157        store: crate::lifecycle::adapters::InMemoryRuntimeStore,
158        execution_factory: Option<ExecutionFactory>,
159        health_registry: Arc<HealthCheckRegistry>,
160        metrics: Arc<dyn MetricsCollector>,
161    ) -> Arc<RuntimeBus> {
162        let execution: Arc<dyn RuntimeExecutionPort> = if let Some(factory) = execution_factory {
163            factory(controller.clone())
164        } else {
165            Arc::new(RuntimeExecutionAdapter::new(controller))
166        };
167        Arc::new(
168            RuntimeBus::new(
169                Arc::new(store.clone()),
170                Arc::new(store.clone()),
171                Arc::new(store.clone()),
172                Arc::new(store.clone()),
173            )
174            .with_uow(Arc::new(store))
175            .with_execution(execution)
176            .with_health_registry(health_registry)
177            .with_metrics(metrics),
178        )
179    }
180
181    pub async fn build(self) -> Result<CamelContext, CamelError> {
182        let registry = self
183            .registry
184            .unwrap_or_else(|| Arc::new(std::sync::Mutex::new(Registry::new())));
185        let languages = self.languages.unwrap_or_else(Self::built_in_languages);
186        let simple_with_resolver: Arc<dyn Language> = Arc::new(
187            camel_language_simple::SimpleLanguage::with_resolver(Arc::new({
188                let languages = Arc::clone(&languages);
189                move |name| {
190                    languages
191                        .lock()
192                        .ok()
193                        .and_then(|registry| registry.get(name).cloned())
194                }
195            })),
196        );
197        languages
198            .lock()
199            .expect("mutex poisoned: another thread panicked while holding this lock") // allow-unwrap
200            .insert("simple".to_string(), simple_with_resolver);
201        let metrics = self.metrics.unwrap_or_else(|| Arc::new(NoOpMetrics));
202        let platform_service = self
203            .platform_service
204            .unwrap_or_else(|| Arc::new(NoopPlatformService::default()));
205        let health_registry = self.health_registry.unwrap_or_else(|| {
206            Arc::new(HealthCheckRegistry::new(std::time::Duration::from_secs(5)))
207        });
208
209        // Default idempotent repository registry with a built-in memory repo.
210        // Built BEFORE the controller so the same Arc can be shared between
211        // CamelContext (user-facing register API) and DefaultRouteController
212        // (compile-time repository-name resolution for the idempotent_consumer step).
213        let idempotent_repositories: crate::registry::SharedIdempotentRegistry = {
214            let reg = Arc::new(IdempotentRegistry::new());
215            let memory = Arc::new(MemoryIdempotentRepository::new("memory"));
216            // If registration fails (e.g. someone already registered "memory"),
217            // it's a programming error — unwrap is safe.
218            reg.register("memory", memory)
219                .expect("built-in memory idempotent repository registration must succeed"); // allow-unwrap
220            reg
221        };
222
223        // Default claim check repository registry with a built-in memory repo.
224        let claim_check_repositories: crate::registry::SharedClaimCheckRegistry = {
225            let reg = Arc::new(ClaimCheckRegistry::new());
226            let memory = Arc::new(MemoryClaimCheckRepository::new("memory"));
227            reg.register("memory", memory)
228                .expect("built-in memory claim check repository registration must succeed"); // allow-unwrap
229            reg
230        };
231
232        // Default cache repository registry with a built-in memory repo.
233        let cache_repositories: crate::registry::SharedCacheRegistry = {
234            let reg = Arc::new(CacheRegistry::new());
235            let memory = Arc::new(MemoryCacheRepository::new("memory", 10_000));
236            reg.register("memory", memory)
237                .expect("built-in memory cache repository registration must succeed"); // allow-unwrap
238            reg
239        };
240
241        let (controller, actor_join, supervision_join) =
242            if let Some(config) = self.supervision_config {
243                let (crash_tx, crash_rx) = tokio::sync::mpsc::channel(64);
244                let mut controller_impl = if let Some(ref beans) = self.beans {
245                    DefaultRouteController::with_languages_and_beans(
246                        Arc::clone(&registry),
247                        Arc::clone(&languages),
248                        Arc::clone(&platform_service),
249                        Arc::clone(beans),
250                    )
251                } else {
252                    DefaultRouteController::with_languages(
253                        Arc::clone(&registry),
254                        Arc::clone(&languages),
255                        Arc::clone(&platform_service),
256                    )
257                };
258                if let Some(invoker) = self.function_invoker.clone() {
259                    controller_impl = controller_impl.with_function_invoker(invoker);
260                }
261                controller_impl.set_idempotent_repositories(Arc::clone(&idempotent_repositories));
262                controller_impl.set_claim_check_repositories(Arc::clone(&claim_check_repositories));
263                controller_impl.set_cache_repositories(Arc::clone(&cache_repositories));
264                controller_impl.set_health_registry(Arc::clone(&health_registry));
265                controller_impl.set_crash_notifier(crash_tx);
266                let (controller, actor_join) = spawn_controller_actor(controller_impl);
267                let supervision_join = spawn_supervision_task(
268                    controller.clone(),
269                    config,
270                    Some(Arc::clone(&metrics)),
271                    crash_rx,
272                );
273                (controller, actor_join, Some(supervision_join))
274            } else {
275                let mut controller_impl = if let Some(ref beans) = self.beans {
276                    DefaultRouteController::with_languages_and_beans(
277                        Arc::clone(&registry),
278                        Arc::clone(&languages),
279                        Arc::clone(&platform_service),
280                        Arc::clone(beans),
281                    )
282                } else {
283                    DefaultRouteController::with_languages(
284                        Arc::clone(&registry),
285                        Arc::clone(&languages),
286                        Arc::clone(&platform_service),
287                    )
288                };
289                if let Some(invoker) = self.function_invoker.clone() {
290                    controller_impl = controller_impl.with_function_invoker(invoker);
291                }
292                controller_impl.set_idempotent_repositories(Arc::clone(&idempotent_repositories));
293                controller_impl.set_claim_check_repositories(Arc::clone(&claim_check_repositories));
294                controller_impl.set_cache_repositories(Arc::clone(&cache_repositories));
295                controller_impl.set_health_registry(Arc::clone(&health_registry));
296                let (controller, actor_join) = spawn_controller_actor(controller_impl);
297                (controller, actor_join, None)
298            };
299
300        let store = self.runtime_store.unwrap_or_default();
301        let runtime = Self::build_runtime(
302            controller.clone(),
303            store,
304            self.execution_factory,
305            Arc::clone(&health_registry),
306            Arc::clone(&metrics),
307        );
308        let runtime_handle: Arc<dyn camel_api::RuntimeHandle> = runtime.clone();
309        controller
310            .try_set_runtime_handle(runtime_handle)
311            .expect("controller actor mailbox should accept initial runtime handle"); // allow-unwrap
312
313        let template_registry = self
314            .template_registry
315            .unwrap_or_else(|| Arc::new(TemplateRegistry::new()));
316
317        Ok(CamelContext::from_parts(FromParts {
318            registry,
319            route_controller: controller,
320            _actor_join: actor_join,
321            supervision_join,
322            runtime,
323            cancel_token: CancellationToken::new(),
324            metrics,
325            platform_service,
326            languages,
327            shutdown_timeout: self.shutdown_timeout,
328            services: self.lifecycle_services,
329            health_registry,
330            component_configs: HashMap::new(),
331            function_invoker: self.function_invoker,
332            template_registry,
333            idempotent_repositories,
334            claim_check_repositories,
335            cache_repositories,
336            startup_checks: Vec::<Box<dyn ConfigCheck>>::new(),
337        }))
338    }
339}
340
341impl Default for CamelContextBuilder {
342    fn default() -> Self {
343        Self::new()
344    }
345}
346
347#[cfg(test)]
348mod tests {
349    use super::*;
350
351    #[test]
352    fn builder_default_has_sane_timeout() {
353        let builder = CamelContextBuilder::new();
354        assert_eq!(builder.shutdown_timeout, std::time::Duration::from_secs(5));
355    }
356
357    #[tokio::test]
358    async fn builder_registers_default_memory_idempotent_repository() {
359        let ctx = CamelContext::builder()
360            .build()
361            .await
362            .expect("build context");
363        let repo = ctx.idempotent_repository("memory");
364        assert!(
365            repo.is_some(),
366            "default 'memory' idempotent repository should be registered"
367        );
368    }
369
370    #[tokio::test]
371    async fn builder_registers_default_memory_cache_repository() {
372        let ctx = CamelContext::builder()
373            .build()
374            .await
375            .expect("build context");
376        let repo = ctx.cache_repository("memory");
377        assert!(
378            repo.is_some(),
379            "default 'memory' cache repository should be registered"
380        );
381        assert_eq!(repo.unwrap().name(), "memory");
382    }
383}
camel_core/context_builder.rs

camel_core/
context_builder.rs
