1use std::sync::Arc;
2use thiserror::Error;
3
4#[derive(Debug, Clone, PartialEq, Eq, Error)]
10#[non_exhaustive]
11pub enum ConfigValidationError {
12 #[error(
13 "aggregator config requires at least one completion bound (size, timeout, predicate, or interval)"
14 )]
15 AggregatorMissingCompletionBound,
16
17 #[error("aggregator requires at least one of max_buckets, completionTimeout, or bucket_ttl")]
22 AggregatorMissingMemoryBound,
23
24 #[error(
31 "aggregator Timeout completion requires bucket_ttl (memory-release bound for the timeout-task cap fallback)"
32 )]
33 AggregatorTimeoutRequiresTtl,
34
35 #[error("throttler max_requests must be > 0")]
36 ThrottlerMaxRequestsZero,
37
38 #[error("loop step must specify either 'count' or 'while', not both")]
39 LoopConflictingCountAndWhile,
40
41 #[error("loop step must specify either 'count' or 'while'")]
42 LoopMissingCountOrWhile,
43
44 #[error("SQL use_message_body_for_sql requires allow_dynamic_query=true")]
45 SqlDynamicQueryWithoutAllowDynamic,
46}
47
48#[derive(Debug, Clone, Error)]
50#[non_exhaustive]
51pub enum CamelError {
52 #[error("Component not found: {0}")]
53 ComponentNotFound(String),
54
55 #[error("Endpoint creation failed: {0}")]
56 EndpointCreationFailed(String),
57
58 #[error("Processor error: {0}")]
59 ProcessorError(String),
60
61 #[error("Processor error: {0}")]
64 ProcessorErrorWithSource(String, #[source] Arc<dyn std::error::Error + Send + Sync>),
65
66 #[error("Type conversion failed: {0}")]
67 TypeConversionFailed(String),
68
69 #[error("Invalid URI: {0}")]
70 InvalidUri(String),
71
72 #[error("Channel closed")]
73 ChannelClosed,
74
75 #[error("Route error: {0}")]
76 RouteError(String),
77
78 #[error("IO error: {0}")]
79 Io(String),
80
81 #[error("Dead letter channel failed: {0}")]
82 DeadLetterChannelFailed(String),
83
84 #[error("Circuit breaker open: {0}")]
85 CircuitOpen(String),
86
87 #[error("HTTP {method} {url} failed: {status_code} {status_text}")]
88 HttpOperationFailed {
89 method: String,
90 url: String,
91 status_code: u16,
92 status_text: String,
93 response_body: Option<String>,
94 },
95
96 #[error("Consumer stopping: semaphore closed during poll_ready")]
100 ConsumerStopping,
101
102 #[error("Configuration error: {0}")]
103 Config(String),
104
105 #[error("Configuration validation error: {0}")]
109 ConfigValidation(ConfigValidationError),
110
111 #[error("Body stream has already been consumed")]
112 AlreadyConsumed,
113
114 #[error("Stream size exceeded limit: {0}")]
115 StreamLimitExceeded(usize),
116
117 #[error("Unauthenticated: {0}")]
118 Unauthenticated(String),
119
120 #[error("Unauthorized: {0}")]
121 Unauthorized(String),
122
123 #[error("Validation failed: {0}")]
124 ValidationError(String),
125}
126
127impl CamelError {
128 pub fn classify(&self) -> &'static str {
129 #[allow(unreachable_patterns)]
130 match self {
131 Self::ComponentNotFound(_) => "component",
132 Self::EndpointCreationFailed(_) | Self::InvalidUri(_) => "endpoint",
133 Self::ProcessorError(_) | Self::ProcessorErrorWithSource(_, _) => "processor",
134 Self::TypeConversionFailed(_) | Self::AlreadyConsumed => "type_conversion",
135 Self::Io(_) => "io",
136 Self::RouteError(_) => "route",
137 Self::CircuitOpen(_) => "circuit_open",
138 Self::HttpOperationFailed { .. } => "http",
139 Self::Config(_) | Self::ConfigValidation(_) => "config",
140 Self::DeadLetterChannelFailed(_) => "dead_letter",
141 Self::ConsumerStopping => "consumer_stop",
142 Self::StreamLimitExceeded(_) => "stream",
143 Self::ChannelClosed => "channel",
144 Self::Unauthenticated(_) => "unauthenticated",
145 Self::Unauthorized(_) => "unauthorized",
146 Self::ValidationError(_) => "validation",
147 _ => "unknown",
148 }
149 }
150
151 pub fn variant_name(&self) -> &'static str {
160 match self {
161 Self::ComponentNotFound(_) => "ComponentNotFound",
162 Self::EndpointCreationFailed(_) => "EndpointCreationFailed",
163 Self::ProcessorError(_) => "ProcessorError",
164 Self::ProcessorErrorWithSource(_, _) => "ProcessorError",
165 Self::TypeConversionFailed(_) => "TypeConversionFailed",
166 Self::InvalidUri(_) => "InvalidUri",
167 Self::ChannelClosed => "ChannelClosed",
168 Self::RouteError(_) => "RouteError",
169 Self::Io(_) => "Io",
170 Self::DeadLetterChannelFailed(_) => "DeadLetterChannelFailed",
171 Self::CircuitOpen(_) => "CircuitOpen",
172 Self::HttpOperationFailed { .. } => "HttpOperationFailed",
173 Self::ConsumerStopping => "ConsumerStopping",
174 Self::Config(_) => "Config",
175 Self::ConfigValidation(_) => "ConfigValidation",
176 Self::AlreadyConsumed => "AlreadyConsumed",
177 Self::StreamLimitExceeded(_) => "StreamLimitExceeded",
178 Self::Unauthenticated(_) => "Unauthenticated",
179 Self::Unauthorized(_) => "Unauthorized",
180 Self::ValidationError(_) => "ValidationError",
181 }
182 }
183}
184
185impl From<std::io::Error> for CamelError {
186 fn from(err: std::io::Error) -> Self {
187 CamelError::Io(err.to_string())
188 }
189}
190
191impl From<crate::template::TemplateError> for CamelError {
192 fn from(err: crate::template::TemplateError) -> Self {
193 CamelError::Config(err.to_string())
194 }
195}
196
197impl From<ConfigValidationError> for CamelError {
198 fn from(e: ConfigValidationError) -> Self {
199 CamelError::ConfigValidation(e)
200 }
201}
202
203#[cfg(test)]
204mod tests {
205 use super::*;
206
207 fn all_error_samples() -> Vec<CamelError> {
208 vec![
209 CamelError::ComponentNotFound("x".to_string()),
210 CamelError::EndpointCreationFailed("x".to_string()),
211 CamelError::ProcessorError("x".to_string()),
212 CamelError::ProcessorErrorWithSource(
213 "x".to_string(),
214 Arc::new(std::io::Error::other("inner")),
215 ),
216 CamelError::TypeConversionFailed("x".to_string()),
217 CamelError::InvalidUri("x".to_string()),
218 CamelError::ChannelClosed,
219 CamelError::RouteError("x".to_string()),
220 CamelError::Io("x".to_string()),
221 CamelError::DeadLetterChannelFailed("x".to_string()),
222 CamelError::CircuitOpen("x".to_string()),
223 CamelError::HttpOperationFailed {
224 method: "GET".to_string(),
225 url: "https://example.com".to_string(),
226 status_code: 500,
227 status_text: "Internal Server Error".to_string(),
228 response_body: Some("error".to_string()),
229 },
230 CamelError::ConsumerStopping,
231 CamelError::Config("x".to_string()),
232 CamelError::ConfigValidation(ConfigValidationError::ThrottlerMaxRequestsZero),
233 CamelError::AlreadyConsumed,
234 CamelError::StreamLimitExceeded(42),
235 CamelError::Unauthenticated("token expired".to_string()),
236 CamelError::Unauthorized("missing admin role".to_string()),
237 CamelError::ValidationError("body does not match schema".to_string()),
238 ]
239 }
240
241 #[test]
242 fn test_http_operation_failed_display() {
243 let err = CamelError::HttpOperationFailed {
244 method: "GET".to_string(),
245 url: "https://example.com/test".to_string(),
246 status_code: 404,
247 status_text: "Not Found".to_string(),
248 response_body: Some("page not found".to_string()),
249 };
250 let msg = format!("{err}");
251 assert!(msg.contains("404"));
252 assert!(msg.contains("Not Found"));
253 }
254
255 #[test]
256 fn test_http_operation_failed_clone() {
257 let err = CamelError::HttpOperationFailed {
258 method: "POST".to_string(),
259 url: "https://api.example.com/users".to_string(),
260 status_code: 500,
261 status_text: "Internal Server Error".to_string(),
262 response_body: None,
263 };
264 let cloned = err.clone();
265 assert!(matches!(
266 cloned,
267 CamelError::HttpOperationFailed {
268 status_code: 500,
269 ..
270 }
271 ));
272 }
273
274 #[test]
275 fn test_classify_maps_all_variants() {
276 assert_eq!(
277 CamelError::ComponentNotFound("x".to_string()).classify(),
278 "component"
279 );
280 assert_eq!(
281 CamelError::EndpointCreationFailed("x".to_string()).classify(),
282 "endpoint"
283 );
284 assert_eq!(
285 CamelError::ProcessorError("x".to_string()).classify(),
286 "processor"
287 );
288 assert_eq!(
289 CamelError::TypeConversionFailed("x".to_string()).classify(),
290 "type_conversion"
291 );
292 assert_eq!(
293 CamelError::InvalidUri("x".to_string()).classify(),
294 "endpoint"
295 );
296 assert_eq!(CamelError::ChannelClosed.classify(), "channel");
297 assert_eq!(CamelError::RouteError("x".to_string()).classify(), "route");
298 assert_eq!(CamelError::Io("x".to_string()).classify(), "io");
299 assert_eq!(
300 CamelError::DeadLetterChannelFailed("x".to_string()).classify(),
301 "dead_letter"
302 );
303 assert_eq!(
304 CamelError::CircuitOpen("x".to_string()).classify(),
305 "circuit_open"
306 );
307 assert_eq!(
308 CamelError::HttpOperationFailed {
309 method: "GET".to_string(),
310 url: "https://example.com".to_string(),
311 status_code: 500,
312 status_text: "Internal Server Error".to_string(),
313 response_body: None,
314 }
315 .classify(),
316 "http"
317 );
318 assert_eq!(CamelError::Config("x".to_string()).classify(), "config");
319 assert_eq!(
320 CamelError::ConfigValidation(ConfigValidationError::ThrottlerMaxRequestsZero)
321 .classify(),
322 "config"
323 );
324 assert_eq!(CamelError::AlreadyConsumed.classify(), "type_conversion");
325 assert_eq!(CamelError::StreamLimitExceeded(42).classify(), "stream");
326 assert_eq!(
327 CamelError::ValidationError("bad".to_string()).classify(),
328 "validation"
329 );
330 }
331
332 #[test]
333 fn test_classify_output_is_ascii_and_short() {
334 for error in all_error_samples() {
335 let class = error.classify();
336 assert!(class.is_ascii());
337 assert!(class.len() <= 15, "class too long: {class}");
338 }
339 }
340
341 #[test]
342 fn test_auth_variants_classify() {
343 assert_eq!(
344 CamelError::Unauthenticated("x".to_string()).classify(),
345 "unauthenticated"
346 );
347 assert_eq!(
348 CamelError::Unauthorized("x".to_string()).classify(),
349 "unauthorized"
350 );
351 }
352
353 #[test]
354 fn test_validation_error_classify() {
355 assert_eq!(
356 CamelError::ValidationError("bad".to_string()).classify(),
357 "validation"
358 );
359 }
360
361 #[test]
362 fn test_auth_variants_are_clone() {
363 let err = CamelError::Unauthenticated("test".to_string());
364 let cloned = err.clone();
365 assert!(matches!(cloned, CamelError::Unauthenticated(_)));
366
367 let err2 = CamelError::Unauthorized("test".to_string());
368 let cloned2 = err2.clone();
369 assert!(matches!(cloned2, CamelError::Unauthorized(_)));
370 }
371}
372
373#[cfg(test)]
374mod variant_name_tests {
375 use super::{CamelError, ConfigValidationError};
376 use std::sync::Arc;
377
378 #[test]
383 fn variant_name_covers_all_variants() {
384 let cases: Vec<(CamelError, &str)> = vec![
385 (
386 CamelError::ComponentNotFound("x".into()),
387 "ComponentNotFound",
388 ),
389 (
390 CamelError::EndpointCreationFailed("x".into()),
391 "EndpointCreationFailed",
392 ),
393 (CamelError::ProcessorError("x".into()), "ProcessorError"),
394 (
395 CamelError::ProcessorErrorWithSource(
396 "x".into(),
397 Arc::new(std::io::Error::other("y")),
398 ),
399 "ProcessorError", ),
401 (
402 CamelError::TypeConversionFailed("x".into()),
403 "TypeConversionFailed",
404 ),
405 (CamelError::InvalidUri("x".into()), "InvalidUri"),
406 (CamelError::ChannelClosed, "ChannelClosed"),
407 (CamelError::RouteError("x".into()), "RouteError"),
408 (CamelError::Io("x".into()), "Io"),
409 (
410 CamelError::DeadLetterChannelFailed("x".into()),
411 "DeadLetterChannelFailed",
412 ),
413 (CamelError::CircuitOpen("x".into()), "CircuitOpen"),
414 (
415 CamelError::HttpOperationFailed {
416 method: "GET".into(),
417 url: "https://example.com".into(),
418 status_code: 500,
419 status_text: "Internal Server Error".into(),
420 response_body: None,
421 },
422 "HttpOperationFailed",
423 ),
424 (CamelError::Config("x".into()), "Config"),
425 (
426 CamelError::ConfigValidation(ConfigValidationError::ThrottlerMaxRequestsZero),
427 "ConfigValidation",
428 ),
429 (CamelError::AlreadyConsumed, "AlreadyConsumed"),
430 (CamelError::StreamLimitExceeded(42), "StreamLimitExceeded"),
431 (CamelError::Unauthenticated("x".into()), "Unauthenticated"),
432 (CamelError::Unauthorized("x".into()), "Unauthorized"),
433 (CamelError::ValidationError("bad".into()), "ValidationError"),
434 ];
435
436 for (err, expected) in cases {
437 assert_eq!(
438 err.variant_name(),
439 expected,
440 "variant_name mismatch for {:?}",
441 err
442 );
443 }
444 }
445}