camel_component_exec/
endpoint.rs1use std::sync::Arc;
8
9use camel_api::{BoxProcessor, CamelError};
10use camel_component_api::{Consumer, Endpoint, ProducerContext, RuntimeObservability};
11use tokio::sync::Semaphore;
12
13use crate::config::ExecGlobalConfig;
14use crate::producer::ExecProducer;
15
16pub struct ExecEndpoint {
22 uri: String,
23 profile_name: String,
24 global: Arc<ExecGlobalConfig>,
25}
26
27impl ExecEndpoint {
28 pub fn new(uri: String, profile_name: String, global: Arc<ExecGlobalConfig>) -> Self {
29 Self {
30 uri,
31 profile_name,
32 global,
33 }
34 }
35}
36
37impl Endpoint for ExecEndpoint {
38 fn uri(&self) -> &str {
39 &self.uri
40 }
41
42 fn create_consumer(
43 &self,
44 _rt: Arc<dyn RuntimeObservability>,
45 ) -> Result<Box<dyn Consumer>, CamelError> {
46 Err(CamelError::InvalidUri(
47 "exec component is producer-only".into(),
48 ))
49 }
50
51 fn create_producer(
52 &self,
53 rt: Arc<dyn RuntimeObservability>,
54 ctx: &ProducerContext,
55 ) -> Result<BoxProcessor, CamelError> {
56 let profile = self
57 .global
58 .profile(&self.profile_name)
59 .ok_or_else(|| {
60 CamelError::EndpointCreationFailed(format!(
61 "exec profile {:?} not found",
62 self.profile_name
63 ))
64 })?
65 .clone();
66
67 let route_id = ctx.route_id().unwrap_or("unknown").to_string();
68 let semaphore = Arc::new(Semaphore::new(profile.concurrency));
69 let host_env: std::collections::HashMap<String, String> = std::env::vars().collect();
70
71 let producer = ExecProducer {
72 profile: Arc::new(profile),
73 global: Arc::clone(&self.global),
74 route_id,
75 host_env,
76 semaphore,
77 rt: Some(rt),
78 };
79
80 Ok(BoxProcessor::new(producer))
81 }
82}
83
84#[cfg(test)]
85mod tests {
86 use std::sync::Arc;
87
88 use camel_api::CamelError;
89 use camel_component_api::{Component, NoOpComponentContext, RuntimeObservability};
90
91 use crate::ExecComponent;
92 use crate::config::ExecGlobalConfig;
93
94 fn component_with_echo() -> ExecComponent {
95 let cfg: ExecGlobalConfig = toml::from_str(
96 r#"
97[[profiles]]
98name = "echo"
99executable = "echo"
100"#,
101 )
102 .unwrap();
103 ExecComponent::new(cfg)
104 }
105
106 fn extract_err<T>(result: Result<T, CamelError>) -> CamelError {
107 match result {
108 Err(e) => e,
109 Ok(_) => panic!("expected Err, got Ok"),
110 }
111 }
112
113 #[test]
114 fn create_endpoint_rejects_missing_prefix() {
115 let c = component_with_echo();
116 let ctx = NoOpComponentContext;
117 let err = extract_err(c.create_endpoint("notecho:echo", &ctx));
118 assert!(
119 matches!(&err, CamelError::InvalidUri(_)),
120 "expected InvalidUri, got {err:?}"
121 );
122 assert!(
123 err.to_string().contains("exec:"),
124 "should mention 'exec:': {err}"
125 );
126 }
127
128 #[test]
129 fn create_endpoint_rejects_empty_profile() {
130 let c = component_with_echo();
131 let ctx = NoOpComponentContext;
132 let err = extract_err(c.create_endpoint("exec:", &ctx));
133 assert!(
134 matches!(&err, CamelError::InvalidUri(_)),
135 "expected InvalidUri, got {err:?}"
136 );
137 assert!(
138 err.to_string().contains("profile name"),
139 "should mention 'profile name': {err}"
140 );
141 }
142
143 #[test]
144 fn create_endpoint_strips_query_string() {
145 let c = component_with_echo();
146 let ctx = NoOpComponentContext;
147 let ep = c
148 .create_endpoint("exec:echo?foo=bar", &ctx)
149 .expect("should succeed with query string");
150 assert_eq!(ep.uri(), "exec:echo?foo=bar");
151 }
152
153 #[test]
154 fn create_endpoint_rejects_unknown_profile() {
155 let c = component_with_echo();
156 let ctx = NoOpComponentContext;
157 let err = extract_err(c.create_endpoint("exec:nope", &ctx));
158 assert!(
159 matches!(&err, CamelError::EndpointCreationFailed(_)),
160 "expected EndpointCreationFailed, got {err:?}"
161 );
162 assert!(
163 err.to_string().contains("not configured"),
164 "should mention 'not configured': {err}"
165 );
166 }
167
168 #[test]
169 fn create_endpoint_valid_profile_ok() {
170 let c = component_with_echo();
171 let ctx = NoOpComponentContext;
172 let ep = c
173 .create_endpoint("exec:echo", &ctx)
174 .expect("should succeed");
175 assert_eq!(ep.uri(), "exec:echo");
176 }
177
178 #[test]
179 fn endpoint_is_producer_only() {
180 let c = component_with_echo();
181 let ctx = NoOpComponentContext;
182 let ep = c
183 .create_endpoint("exec:echo", &ctx)
184 .expect("should succeed");
185 let rt: Arc<dyn RuntimeObservability> = Arc::new(NoOpComponentContext);
186 let err = extract_err(ep.create_consumer(rt));
187 assert!(
188 matches!(&err, CamelError::InvalidUri(_)),
189 "expected InvalidUri, got {err:?}"
190 );
191 assert!(
192 err.to_string().contains("producer-only"),
193 "should mention 'producer-only': {err}"
194 );
195 }
196}