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

camel_core/
context_builder.rs
