camel_api/
function.rs

1use crate::Exchange;
2
3#[derive(Debug, Clone, Hash, Eq, PartialEq)]
4pub struct FunctionId(pub String);
5
6impl std::fmt::Display for FunctionId {
7    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
8        f.write_str(&self.0)
9    }
10}
11
12impl FunctionId {
13    pub fn compute(runtime: &str, source: &str, timeout_ms_resolved: u64) -> Self {
14        let mut hasher = blake3::Hasher::new();
15        let rlen = (runtime.len() as u64).to_le_bytes();
16        hasher.update(&rlen);
17        hasher.update(runtime.as_bytes());
18        let slen = (source.len() as u64).to_le_bytes();
19        hasher.update(&slen);
20        hasher.update(source.as_bytes());
21        hasher.update(&timeout_ms_resolved.to_le_bytes());
22        let hash = hasher.finalize();
23        let truncated = &hash.as_bytes()[..16];
24        let hex: String = truncated.iter().map(|b| format!("{:02x}", b)).collect();
25        Self(hex)
26    }
27}
28
29#[derive(Debug, Clone)]
30pub struct FunctionDefinition {
31    pub id: FunctionId,
32    pub runtime: String,
33    pub source: String,
34    pub timeout_ms: u64,
35    pub route_id: Option<String>,
36    pub step_index: Option<usize>,
37}
38
39#[derive(Debug, Clone, Default)]
40pub struct ExchangePatch {
41    pub body: Option<PatchBody>,
42    pub headers_set: Vec<(String, serde_json::Value)>,
43    pub headers_removed: Vec<String>,
44    pub properties_set: Vec<(String, serde_json::Value)>,
45}
46
47#[derive(Debug, Clone)]
48#[non_exhaustive]
49pub enum PatchBody {
50    Text(String),
51    Json(serde_json::Value),
52    Empty,
53}
54
55#[derive(Debug, thiserror::Error)]
56#[non_exhaustive]
57pub enum FunctionInvocationError {
58    #[error("function {function_id} not registered on runtime")]
59    NotRegistered { function_id: FunctionId },
60    #[error("function {function_id} timed out after {timeout_ms}ms")]
61    Timeout {
62        function_id: FunctionId,
63        timeout_ms: u64,
64    },
65    #[error("runner unavailable: {reason}")]
66    RunnerUnavailable { reason: String },
67    #[error("user code failed: {message}")]
68    UserError {
69        function_id: FunctionId,
70        message: String,
71        stack: Option<String>,
72    },
73    #[error("transport error: {0}")]
74    Transport(String),
75    #[error("invalid patch: {0}")]
76    InvalidPatch(String),
77}
78
79#[derive(Debug, Default, Clone)]
80pub struct FunctionDiff {
81    pub added: Vec<(FunctionDefinition, Option<String>)>,
82    pub removed: Vec<(FunctionId, Option<String>)>,
83    pub unchanged: Vec<FunctionId>,
84}
85
86#[derive(Debug, Default, Clone)]
87pub struct PrepareToken {
88    pub registered: Vec<(FunctionDefinition, Option<String>)>,
89}
90
91pub trait FunctionInvokerSync: Send + Sync {
92    fn stage_pending(&self, def: FunctionDefinition, route_id: Option<&str>, generation: u64);
93    fn discard_staging(&self, generation: u64);
94    fn begin_reload(&self) -> u64;
95    fn function_refs_for_route(&self, route_id: &str) -> Vec<(FunctionId, Option<String>)>;
96    fn staged_refs_for_route(
97        &self,
98        route_id: &str,
99        generation: u64,
100    ) -> Vec<(FunctionId, Option<String>)>;
101    fn staged_defs_for_route(
102        &self,
103        route_id: &str,
104        generation: u64,
105    ) -> Vec<(FunctionDefinition, Option<String>)>;
106}
107
108#[async_trait::async_trait]
109pub trait FunctionInvoker: FunctionInvokerSync + Send + Sync {
110    async fn register(
111        &self,
112        def: FunctionDefinition,
113        route_id: Option<&str>,
114    ) -> Result<(), FunctionInvocationError>;
115    async fn unregister(
116        &self,
117        id: &FunctionId,
118        route_id: Option<&str>,
119    ) -> Result<(), FunctionInvocationError>;
120    async fn invoke(
121        &self,
122        id: &FunctionId,
123        exchange: &Exchange,
124    ) -> Result<ExchangePatch, FunctionInvocationError>;
125    async fn prepare_reload(
126        &self,
127        diff: FunctionDiff,
128        generation: u64,
129    ) -> Result<PrepareToken, FunctionInvocationError>;
130    async fn finalize_reload(
131        &self,
132        diff: &FunctionDiff,
133        generation: u64,
134    ) -> Result<(), FunctionInvocationError>;
135    async fn rollback_reload(
136        &self,
137        token: PrepareToken,
138        generation: u64,
139    ) -> Result<(), FunctionInvocationError>;
140    async fn commit_reload(
141        &self,
142        diff: FunctionDiff,
143        generation: u64,
144    ) -> Result<(), FunctionInvocationError> {
145        let _token = self.prepare_reload(diff.clone(), generation).await?;
146        self.finalize_reload(&diff, generation).await
147    }
148    async fn commit_staged(&self) -> Result<(), FunctionInvocationError>;
149}
camel_api/function.rs

camel_api/
function.rs
