camel_component_file/
health.rs

1use async_trait::async_trait;
2use camel_component_api::{AsyncHealthCheck, CamelError, CheckResult};
3use std::future::Future;
4use std::path::PathBuf;
5use std::pin::Pin;
6use std::time::Duration;
7
8type ProbeFuture = Pin<Box<dyn Future<Output = Result<(), CamelError>> + Send>>;
9
10trait FileHealthProbe: Send + Sync {
11    fn probe(&self) -> ProbeFuture;
12}
13
14struct DirectoryMetadataProbe {
15    path: PathBuf,
16    timeout: Duration,
17}
18
19impl DirectoryMetadataProbe {
20    fn new(path: PathBuf, timeout: Duration) -> Self {
21        Self { path, timeout }
22    }
23}
24
25impl FileHealthProbe for DirectoryMetadataProbe {
26    fn probe(&self) -> ProbeFuture {
27        let path = self.path.clone();
28        let timeout = self.timeout;
29        Box::pin(async move {
30            tokio::time::timeout(timeout, tokio::fs::metadata(&path))
31                .await
32                .map_err(|_| {
33                    CamelError::ProcessorError(format!(
34                        "health probe timed out for directory: {}",
35                        path.display()
36                    ))
37                })?
38                .map_err(|e| CamelError::ProcessorError(format!("directory metadata error: {}", e)))
39                .map(|_| ())
40        })
41    }
42}
43
44/// Health check for the file component.
45///
46/// Probes that the target directory exists via `tokio::fs::metadata()`.
47/// Reports `Healthy` when the directory is accessible.
48/// Reports `Degraded` if the directory is missing or the probe times out.
49pub struct FileHealthCheck {
50    probe: Option<Box<dyn FileHealthProbe>>,
51    timeout: Duration,
52}
53
54impl FileHealthCheck {
55    pub fn new(path: PathBuf) -> Self {
56        let timeout = Duration::from_secs(5);
57        Self {
58            probe: Some(Box::new(DirectoryMetadataProbe::new(path, timeout))),
59            timeout,
60        }
61    }
62
63    #[cfg(test)]
64    fn with_probe_for_tests(probe: Box<dyn FileHealthProbe>, timeout: Duration) -> Self {
65        Self {
66            probe: Some(probe),
67            timeout,
68        }
69    }
70}
71
72#[async_trait]
73impl AsyncHealthCheck for FileHealthCheck {
74    fn name(&self) -> &str {
75        "file"
76    }
77
78    async fn check(&self) -> CheckResult {
79        let Some(probe) = self.probe.as_ref() else {
80            return CheckResult::healthy(self.name());
81        };
82
83        match tokio::time::timeout(self.timeout, probe.probe()).await {
84            Ok(Ok(())) => CheckResult::healthy(self.name()),
85            Ok(Err(err)) => CheckResult::degraded(self.name(), &err.to_string()),
86            Err(_) => CheckResult::degraded(self.name(), "directory probe timed out"),
87        }
88    }
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94    use camel_component_api::HealthStatus;
95    use std::sync::Arc;
96
97    struct MockProbe {
98        responder: Arc<dyn Fn() -> ProbeFuture + Send + Sync>,
99    }
100
101    impl MockProbe {
102        fn new<F>(f: F) -> Self
103        where
104            F: Fn() -> ProbeFuture + Send + Sync + 'static,
105        {
106            Self {
107                responder: Arc::new(f),
108            }
109        }
110    }
111
112    impl FileHealthProbe for MockProbe {
113        fn probe(&self) -> ProbeFuture {
114            (self.responder)()
115        }
116    }
117
118    #[tokio::test]
119    async fn file_health_check_healthy_when_directory_exists() {
120        let dir = tempfile::tempdir().unwrap();
121        let check = FileHealthCheck::new(dir.path().to_path_buf());
122        let result = check.check().await;
123        assert_eq!(result.name, "file");
124        assert_eq!(result.status, HealthStatus::Healthy);
125        assert!(result.message.is_none());
126    }
127
128    #[tokio::test]
129    async fn file_health_check_degraded_when_directory_missing() {
130        let check = FileHealthCheck::new(PathBuf::from("/tmp/nonexistent_dir_12345"));
131        let result = check.check().await;
132        assert_eq!(result.name, "file");
133        assert_eq!(result.status, HealthStatus::Degraded);
134        assert!(
135            result
136                .message
137                .as_deref()
138                .is_some_and(|m| m.contains("directory metadata error"))
139        );
140    }
141
142    #[tokio::test]
143    async fn file_health_check_degraded_on_timeout() {
144        let probe = MockProbe::new(|| {
145            Box::pin(async {
146                tokio::time::sleep(Duration::from_millis(100)).await;
147                Ok(())
148            })
149        });
150        let check =
151            FileHealthCheck::with_probe_for_tests(Box::new(probe), Duration::from_millis(5));
152        let result = check.check().await;
153        assert_eq!(result.name, "file");
154        assert_eq!(result.status, HealthStatus::Degraded);
155        assert_eq!(result.message.as_deref(), Some("directory probe timed out"));
156    }
157}
camel_component_file/health.rs

camel_component_file/
health.rs
