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

camel_core/
context_builder.rs
