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

camel_function/provider/
mod.rs
