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

camel_core/
context_builder.rs
