rskit_dataset/
target.rs

1//! Target trait — publish collected data to a destination.
2
3use rskit_errors::{AppError, AppResult, ErrorCode};
4use serde::{Deserialize, Serialize};
5use std::path::Path;
6
7/// Result returned by a publish target.
8#[derive(Debug, Clone, Serialize, Deserialize)]
9pub struct PublishResult {
10    /// Target identifier.
11    pub target_name: String,
12    /// Published location.
13    pub location: String,
14    /// Number of files published or observed by the target.
15    pub files_published: usize,
16    /// Human-readable publish summary.
17    pub message: String,
18}
19
20/// Destination for collected dataset output.
21#[async_trait::async_trait]
22pub trait Target: Send + Sync {
23    /// Stable target name.
24    fn name(&self) -> &str;
25
26    /// Publish the dataset directory and optional metadata.
27    async fn publish(
28        &self,
29        directory: &Path,
30        metadata: Option<&std::collections::HashMap<String, String>>,
31    ) -> AppResult<PublishResult>;
32}
33
34/// Local filesystem target — data is already on disk.
35pub struct LocalTarget;
36
37#[async_trait::async_trait]
38impl Target for LocalTarget {
39    fn name(&self) -> &str {
40        "local"
41    }
42
43    async fn publish(
44        &self,
45        directory: &Path,
46        _metadata: Option<&std::collections::HashMap<String, String>>,
47    ) -> AppResult<PublishResult> {
48        let directory = directory.to_path_buf();
49        let file_count = tokio::task::spawn_blocking({
50            let directory = directory.clone();
51            move || {
52                let mut file_count = 0usize;
53                if directory.exists() {
54                    for entry in walkdir(&directory)? {
55                        if entry.is_file() {
56                            file_count += 1;
57                        }
58                    }
59                }
60                Ok::<usize, AppError>(file_count)
61            }
62        })
63        .await
64        .map_err(AppError::internal)??;
65        Ok(PublishResult {
66            target_name: self.name().to_string(),
67            location: directory.display().to_string(),
68            files_published: file_count,
69            message: format!("Data saved to {}", directory.display()),
70        })
71    }
72}
73
74fn walkdir(dir: &Path) -> AppResult<Vec<std::path::PathBuf>> {
75    let mut files = Vec::new();
76    if dir.is_dir() {
77        for entry in std::fs::read_dir(dir)
78            .map_err(|e| AppError::new(ErrorCode::Internal, format!("read dir failed: {e}")))?
79        {
80            let entry = entry
81                .map_err(|e| AppError::new(ErrorCode::Internal, format!("dir entry error: {e}")))?;
82            let path = entry.path();
83            if path.is_dir() {
84                files.extend(walkdir(&path)?);
85            } else {
86                files.push(path);
87            }
88        }
89    }
90    Ok(files)
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[tokio::test]
98    async fn local_target_counts_nested_files_and_missing_directory_as_empty() {
99        let dir = tempfile::tempdir().unwrap();
100        std::fs::create_dir(dir.path().join("nested")).unwrap();
101        std::fs::write(dir.path().join("a.txt"), b"a").unwrap();
102        std::fs::write(dir.path().join("nested").join("b.txt"), b"b").unwrap();
103
104        let target = LocalTarget;
105        let result = target.publish(dir.path(), None).await.unwrap();
106
107        assert_eq!(target.name(), "local");
108        assert_eq!(result.target_name, "local");
109        assert_eq!(result.files_published, 2);
110        assert!(result.message.contains("Data saved"));
111
112        let missing = dir.path().join("missing");
113        let result = target.publish(&missing, None).await.unwrap();
114        assert_eq!(result.files_published, 0);
115    }
116
117    #[test]
118    fn walkdir_rejects_file_as_directory_with_empty_listing() {
119        let dir = tempfile::tempdir().unwrap();
120        let file = dir.path().join("file.txt");
121        std::fs::write(&file, b"x").unwrap();
122
123        assert!(walkdir(&file).unwrap().is_empty());
124    }
125}
rskit_dataset/target.rs

rskit_dataset/
target.rs
