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::runtime_bus::RuntimeBus;
23use crate::lifecycle::ports::RuntimeExecutionPort;
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    ) -> Arc<RuntimeBus> {
160        let execution: Arc<dyn RuntimeExecutionPort> = if let Some(factory) = execution_factory {
161            factory(controller.clone())
162        } else {
163            Arc::new(RuntimeExecutionAdapter::new(controller))
164        };
165        Arc::new(
166            RuntimeBus::new(
167                Arc::new(store.clone()),
168                Arc::new(store.clone()),
169                Arc::new(store.clone()),
170                Arc::new(store.clone()),
171            )
172            .with_uow(Arc::new(store))
173            .with_execution(execution)
174            .with_health_registry(health_registry),
175        )
176    }
177
178    pub async fn build(self) -> Result<CamelContext, CamelError> {
179        let registry = self
180            .registry
181            .unwrap_or_else(|| Arc::new(std::sync::Mutex::new(Registry::new())));
182        let languages = self.languages.unwrap_or_else(Self::built_in_languages);
183        let simple_with_resolver: Arc<dyn Language> = Arc::new(
184            camel_language_simple::SimpleLanguage::with_resolver(Arc::new({
185                let languages = Arc::clone(&languages);
186                move |name| {
187                    languages
188                        .lock()
189                        .ok()
190                        .and_then(|registry| registry.get(name).cloned())
191                }
192            })),
193        );
194        languages
195            .lock()
196            .expect("mutex poisoned: another thread panicked while holding this lock") // allow-unwrap
197            .insert("simple".to_string(), simple_with_resolver);
198        let metrics = self.metrics.unwrap_or_else(|| Arc::new(NoOpMetrics));
199        let platform_service = self
200            .platform_service
201            .unwrap_or_else(|| Arc::new(NoopPlatformService::default()));
202        let health_registry = self.health_registry.unwrap_or_else(|| {
203            Arc::new(HealthCheckRegistry::new(std::time::Duration::from_secs(5)))
204        });
205
206        // Default idempotent repository registry with a built-in memory repo.
207        // Built BEFORE the controller so the same Arc can be shared between
208        // CamelContext (user-facing register API) and DefaultRouteController
209        // (compile-time repository-name resolution for the idempotent_consumer step).
210        let idempotent_repositories: crate::registry::SharedIdempotentRegistry = {
211            let reg = Arc::new(IdempotentRegistry::new());
212            let memory = Arc::new(MemoryIdempotentRepository::new("memory"));
213            // If registration fails (e.g. someone already registered "memory"),
214            // it's a programming error — unwrap is safe.
215            reg.register("memory", memory)
216                .expect("built-in memory idempotent repository registration must succeed"); // allow-unwrap
217            reg
218        };
219
220        // Default claim check repository registry with a built-in memory repo.
221        let claim_check_repositories: crate::registry::SharedClaimCheckRegistry = {
222            let reg = Arc::new(ClaimCheckRegistry::new());
223            let memory = Arc::new(MemoryClaimCheckRepository::new("memory"));
224            reg.register("memory", memory)
225                .expect("built-in memory claim check repository registration must succeed"); // allow-unwrap
226            reg
227        };
228
229        let (controller, actor_join, supervision_join) =
230            if let Some(config) = self.supervision_config {
231                let (crash_tx, crash_rx) = tokio::sync::mpsc::channel(64);
232                let mut controller_impl = if let Some(ref beans) = self.beans {
233                    DefaultRouteController::with_languages_and_beans(
234                        Arc::clone(&registry),
235                        Arc::clone(&languages),
236                        Arc::clone(&platform_service),
237                        Arc::clone(beans),
238                    )
239                } else {
240                    DefaultRouteController::with_languages(
241                        Arc::clone(&registry),
242                        Arc::clone(&languages),
243                        Arc::clone(&platform_service),
244                    )
245                };
246                if let Some(invoker) = self.function_invoker.clone() {
247                    controller_impl = controller_impl.with_function_invoker(invoker);
248                }
249                controller_impl.set_idempotent_repositories(Arc::clone(&idempotent_repositories));
250                controller_impl.set_claim_check_repositories(Arc::clone(&claim_check_repositories));
251                controller_impl.set_health_registry(Arc::clone(&health_registry));
252                controller_impl.set_crash_notifier(crash_tx);
253                let (controller, actor_join) = spawn_controller_actor(controller_impl);
254                let supervision_join = spawn_supervision_task(
255                    controller.clone(),
256                    config,
257                    Some(Arc::clone(&metrics)),
258                    crash_rx,
259                );
260                (controller, actor_join, Some(supervision_join))
261            } else {
262                let mut controller_impl = if let Some(ref beans) = self.beans {
263                    DefaultRouteController::with_languages_and_beans(
264                        Arc::clone(&registry),
265                        Arc::clone(&languages),
266                        Arc::clone(&platform_service),
267                        Arc::clone(beans),
268                    )
269                } else {
270                    DefaultRouteController::with_languages(
271                        Arc::clone(&registry),
272                        Arc::clone(&languages),
273                        Arc::clone(&platform_service),
274                    )
275                };
276                if let Some(invoker) = self.function_invoker.clone() {
277                    controller_impl = controller_impl.with_function_invoker(invoker);
278                }
279                controller_impl.set_idempotent_repositories(Arc::clone(&idempotent_repositories));
280                controller_impl.set_claim_check_repositories(Arc::clone(&claim_check_repositories));
281                controller_impl.set_health_registry(Arc::clone(&health_registry));
282                let (controller, actor_join) = spawn_controller_actor(controller_impl);
283                (controller, actor_join, None)
284            };
285
286        let store = self.runtime_store.unwrap_or_default();
287        let runtime = Self::build_runtime(
288            controller.clone(),
289            store,
290            self.execution_factory,
291            Arc::clone(&health_registry),
292        );
293        let runtime_handle: Arc<dyn camel_api::RuntimeHandle> = runtime.clone();
294        controller
295            .try_set_runtime_handle(runtime_handle)
296            .expect("controller actor mailbox should accept initial runtime handle"); // allow-unwrap
297
298        let template_registry = self
299            .template_registry
300            .unwrap_or_else(|| Arc::new(TemplateRegistry::new()));
301
302        Ok(CamelContext::from_parts(FromParts {
303            registry,
304            route_controller: controller,
305            _actor_join: actor_join,
306            supervision_join,
307            runtime,
308            cancel_token: CancellationToken::new(),
309            metrics,
310            platform_service,
311            languages,
312            shutdown_timeout: self.shutdown_timeout,
313            services: self.lifecycle_services,
314            health_registry,
315            component_configs: HashMap::new(),
316            function_invoker: self.function_invoker,
317            template_registry,
318            idempotent_repositories,
319            claim_check_repositories,
320            startup_checks: Vec::<Box<dyn ConfigCheck>>::new(),
321        }))
322    }
323}
324
325impl Default for CamelContextBuilder {
326    fn default() -> Self {
327        Self::new()
328    }
329}
330
331#[cfg(test)]
332mod tests {
333    use super::*;
334
335    #[test]
336    fn builder_default_has_sane_timeout() {
337        let builder = CamelContextBuilder::new();
338        assert_eq!(builder.shutdown_timeout, std::time::Duration::from_secs(5));
339    }
340
341    #[tokio::test]
342    async fn builder_registers_default_memory_idempotent_repository() {
343        let ctx = CamelContext::builder()
344            .build()
345            .await
346            .expect("build context");
347        let repo = ctx.idempotent_repository("memory");
348        assert!(
349            repo.is_some(),
350            "default 'memory' idempotent repository should be registered"
351        );
352    }
353}
camel_core/context_builder.rs

camel_core/
context_builder.rs
