camel_function/provider/
mod.rs

1use crate::pool::{RunnerHandle, RunnerPoolKey};
2use camel_api::{Exchange, function::*};
3use std::time::Duration;
4
5mod sealed {
6    pub trait Sealed {}
7}
8
9/// Health status of a function provider instance.
10/// Named `FunctionHealthStatus` to avoid collision with `camel_api::FunctionHealthStatus`.
11#[derive(Debug, Clone)]
12pub enum FunctionHealthStatus {
13    Healthy,
14    Unhealthy(String),
15}
16
17#[derive(Debug, thiserror::Error)]
18pub enum ProviderError {
19    #[error("spawn failed: {0}")]
20    SpawnFailed(String),
21    #[error("health check failed: {0}")]
22    HealthFailed(String),
23    #[error("register failed: {0}")]
24    RegisterFailed(String),
25    #[error("unregister failed: {0}")]
26    UnregisterFailed(String),
27    #[error("invoke failed: {0}")]
28    InvokeFailed(String),
29    #[error("shutdown failed: {0}")]
30    ShutdownFailed(String),
31    #[error("invalid provider config: {0}")]
32    InvalidConfig(String),
33    #[error("http client construction failed: {0}")]
34    HttpClientBuildFailed(String),
35    #[error("boot timeout")]
36    BootTimeout,
37}
38
39#[async_trait::async_trait]
40pub(crate) trait FunctionProvider: Send + Sync + sealed::Sealed {
41    async fn spawn(&self, key: &RunnerPoolKey) -> Result<RunnerHandle, ProviderError>;
42    async fn shutdown(&self, handle: RunnerHandle) -> Result<(), ProviderError>;
43    async fn health(&self, handle: &RunnerHandle) -> Result<FunctionHealthStatus, ProviderError>;
44    async fn register(
45        &self,
46        handle: &RunnerHandle,
47        def: &FunctionDefinition,
48    ) -> Result<(), ProviderError>;
49    async fn unregister(&self, handle: &RunnerHandle, id: &FunctionId)
50    -> Result<(), ProviderError>;
51    async fn invoke(
52        &self,
53        handle: &RunnerHandle,
54        id: &FunctionId,
55        ex: &Exchange,
56        timeout: Duration,
57    ) -> Result<ExchangePatch, ProviderError>;
58}
59
60pub mod container;
61pub mod fake {
62    use super::*;
63    use std::collections::{HashMap, HashSet};
64    use std::sync::atomic::{AtomicUsize, Ordering};
65    use std::sync::{Arc, Mutex};
66    use tokio_util::sync::CancellationToken;
67
68    #[derive(Debug, Clone, Default)]
69    pub struct FakeProviderConfig {
70        pub fail_on_spawn: bool,
71        pub fail_on_register: usize,
72        pub fail_on_health: bool,
73        pub fail_on_shutdown: bool,
74        pub invoke_response: Option<ExchangePatch>,
75        pub invoke_delay: Option<std::time::Duration>,
76    }
77
78    #[derive(Debug, Clone)]
79    pub enum FakeCall {
80        Spawn(RunnerPoolKey),
81        Shutdown(RunnerPoolKey),
82        Health(String),
83        Register(String, FunctionId),
84        Unregister(String, FunctionId),
85        Invoke(String, FunctionId),
86    }
87
88    pub struct FakeProvider {
89        pub config: Arc<Mutex<FakeProviderConfig>>,
90        pub calls: Arc<Mutex<Vec<FakeCall>>>,
91        pub registered: Arc<Mutex<HashMap<String, HashSet<FunctionId>>>>,
92        pub spawned: Arc<Mutex<Vec<RunnerPoolKey>>>,
93        pub shutdowns: Arc<Mutex<Vec<RunnerPoolKey>>>,
94        register_ok_count: Arc<Mutex<usize>>,
95        spawn_count: AtomicUsize,
96    }
97
98    impl FakeProvider {
99        pub fn new(config: FakeProviderConfig) -> Self {
100            Self {
101                config: Arc::new(Mutex::new(config)),
102                calls: Arc::new(Mutex::new(Vec::new())),
103                registered: Arc::new(Mutex::new(HashMap::new())),
104                spawned: Arc::new(Mutex::new(Vec::new())),
105                shutdowns: Arc::new(Mutex::new(Vec::new())),
106                register_ok_count: Arc::new(Mutex::new(0)),
107                spawn_count: AtomicUsize::new(0),
108            }
109        }
110
111        pub fn spawn_count(&self) -> usize {
112            self.spawn_count.load(Ordering::SeqCst)
113        }
114    }
115
116    impl super::sealed::Sealed for FakeProvider {}
117
118    #[async_trait::async_trait]
119    impl FunctionProvider for FakeProvider {
120        async fn spawn(&self, key: &RunnerPoolKey) -> Result<RunnerHandle, ProviderError> {
121            self.spawn_count.fetch_add(1, Ordering::SeqCst);
122            self.calls
123                .lock()
124                .expect("calls") // allow-unwrap
125                .push(FakeCall::Spawn(key.clone()));
126            self.spawned.lock().expect("spawned").push(key.clone()); // allow-unwrap
127            if self.config.lock().expect("config").fail_on_spawn {
128                // allow-unwrap
129                return Err(ProviderError::SpawnFailed("configured".into()));
130            }
131            Ok(RunnerHandle {
132                id: format!("fake-{}", key.runtime),
133                state: Arc::new(Mutex::new(crate::pool::RunnerState::Booting)),
134                cancel: CancellationToken::new(),
135            })
136        }
137
138        async fn shutdown(&self, handle: RunnerHandle) -> Result<(), ProviderError> {
139            self.calls
140                .lock()
141                .expect("calls") // allow-unwrap
142                .push(FakeCall::Shutdown(RunnerPoolKey {
143                    runtime: handle.id.replace("fake-", ""),
144                }));
145            self.shutdowns
146                .lock()
147                .expect("shutdowns") // allow-unwrap
148                .push(RunnerPoolKey {
149                    runtime: handle.id.replace("fake-", ""),
150                });
151            if self.config.lock().expect("config").fail_on_shutdown {
152                // allow-unwrap
153                return Err(ProviderError::ShutdownFailed(
154                    "configured shutdown failure".into(),
155                ));
156            }
157            Ok(())
158        }
159
160        async fn health(
161            &self,
162            handle: &RunnerHandle,
163        ) -> Result<FunctionHealthStatus, ProviderError> {
164            self.calls
165                .lock()
166                .expect("calls") // allow-unwrap
167                .push(FakeCall::Health(handle.id.clone()));
168            if self.config.lock().expect("config").fail_on_health {
169                // allow-unwrap
170                return Ok(FunctionHealthStatus::Unhealthy("configured".into()));
171            }
172            Ok(FunctionHealthStatus::Healthy)
173        }
174
175        async fn register(
176            &self,
177            handle: &RunnerHandle,
178            def: &FunctionDefinition,
179        ) -> Result<(), ProviderError> {
180            self.calls
181                .lock()
182                .expect("calls") // allow-unwrap
183                .push(FakeCall::Register(handle.id.clone(), def.id.clone()));
184            let mut count = self.register_ok_count.lock().expect("count"); // allow-unwrap
185            let cfg = self.config.lock().expect("config").clone(); // allow-unwrap
186            if cfg.fail_on_register > 0 && *count >= cfg.fail_on_register {
187                return Err(ProviderError::RegisterFailed("configured".into()));
188            }
189            *count += 1;
190            self.registered
191                .lock()
192                .expect("registered") // allow-unwrap
193                .entry(handle.id.clone())
194                .or_default()
195                .insert(def.id.clone());
196            Ok(())
197        }
198
199        async fn unregister(
200            &self,
201            handle: &RunnerHandle,
202            id: &FunctionId,
203        ) -> Result<(), ProviderError> {
204            self.calls
205                .lock()
206                .expect("calls") // allow-unwrap
207                .push(FakeCall::Unregister(handle.id.clone(), id.clone()));
208            if let Some(set) = self
209                .registered
210                .lock()
211                .expect("registered") // allow-unwrap
212                .get_mut(&handle.id)
213            {
214                set.remove(id);
215            }
216            Ok(())
217        }
218
219        async fn invoke(
220            &self,
221            handle: &RunnerHandle,
222            id: &FunctionId,
223            _ex: &Exchange,
224            _timeout: Duration,
225        ) -> Result<ExchangePatch, ProviderError> {
226            self.calls
227                .lock()
228                .expect("calls") // allow-unwrap
229                .push(FakeCall::Invoke(handle.id.clone(), id.clone()));
230            let exists = self
231                .registered
232                .lock()
233                .expect("registered") // allow-unwrap
234                .get(&handle.id)
235                .map(|s| s.contains(id))
236                .unwrap_or(false);
237            if !exists {
238                return Err(ProviderError::InvokeFailed("not registered".into()));
239            }
240            let cfg = self.config.lock().expect("config").clone(); // allow-unwrap
241            if let Some(delay) = cfg.invoke_delay {
242                tokio::time::sleep(delay).await;
243            }
244            Ok(cfg.invoke_response.unwrap_or_default())
245        }
246    }
247}
camel_function/provider/mod.rs

camel_function/provider/
mod.rs
