taskflow_rs/executor/handlers/
file.rs

1use tokio::fs;
2use tracing::info;
3
4use crate::error::{Result, TaskFlowError};
5use crate::task::{Task, TaskHandler, TaskResult};
6
7pub struct FileTaskHandler;
8
9impl FileTaskHandler {
10    pub fn new() -> Self {
11        Self
12    }
13}
14
15#[async_trait::async_trait]
16impl TaskHandler for FileTaskHandler {
17    fn task_type(&self) -> &'static str {
18        "file_operation"
19    }
20
21    async fn execute(&self, task: &Task) -> Result<TaskResult> {
22        let operation = task
23            .definition
24            .payload
25            .get("operation")
26            .and_then(|v| v.as_str())
27            .ok_or_else(|| {
28                TaskFlowError::InvalidConfiguration("Missing operation type".to_string())
29            })?;
30
31        let path = task
32            .definition
33            .payload
34            .get("path")
35            .and_then(|v| v.as_str())
36            .ok_or_else(|| TaskFlowError::InvalidConfiguration("Missing file path".to_string()))?;
37
38        info!("Performing file operation {} on: {}", operation, path);
39
40        match operation {
41            "read" => self.read_file(path).await,
42            "write" => self.write_file(task, path).await,
43            "delete" => self.delete_file(path).await,
44            "copy" => self.copy_file(task, path).await,
45            "move" => self.move_file(task, path).await,
46            op => Err(TaskFlowError::InvalidConfiguration(format!(
47                "Unknown file operation: {}",
48                op
49            ))),
50        }
51    }
52}
53
54impl FileTaskHandler {
55    async fn read_file(&self, path: &str) -> Result<TaskResult> {
56        let content = fs::read_to_string(path)
57            .await
58            .map_err(|e| TaskFlowError::ExecutionError(e.to_string()))?;
59
60        Ok(TaskResult {
61            success: true,
62            output: Some(content),
63            error: None,
64            execution_time_ms: 0,
65            metadata: Default::default(),
66        })
67    }
68
69    async fn write_file(&self, task: &Task, path: &str) -> Result<TaskResult> {
70        let content = task
71            .definition
72            .payload
73            .get("content")
74            .and_then(|v| v.as_str())
75            .ok_or_else(|| TaskFlowError::InvalidConfiguration("Missing content".to_string()))?;
76
77        fs::write(path, content)
78            .await
79            .map_err(|e| TaskFlowError::ExecutionError(e.to_string()))?;
80
81        Ok(TaskResult {
82            success: true,
83            output: Some(format!("File written successfully: {}", path)),
84            error: None,
85            execution_time_ms: 0,
86            metadata: Default::default(),
87        })
88    }
89
90    async fn delete_file(&self, path: &str) -> Result<TaskResult> {
91        fs::remove_file(path)
92            .await
93            .map_err(|e| TaskFlowError::ExecutionError(e.to_string()))?;
94
95        Ok(TaskResult {
96            success: true,
97            output: Some(format!("File deleted: {}", path)),
98            error: None,
99            execution_time_ms: 0,
100            metadata: Default::default(),
101        })
102    }
103
104    async fn copy_file(&self, task: &Task, source_path: &str) -> Result<TaskResult> {
105        let dest_path = task
106            .definition
107            .payload
108            .get("destination")
109            .and_then(|v| v.as_str())
110            .ok_or_else(|| {
111                TaskFlowError::InvalidConfiguration("Missing destination path".to_string())
112            })?;
113
114        fs::copy(source_path, dest_path)
115            .await
116            .map_err(|e| TaskFlowError::ExecutionError(e.to_string()))?;
117
118        Ok(TaskResult {
119            success: true,
120            output: Some(format!("File copied from {} to {}", source_path, dest_path)),
121            error: None,
122            execution_time_ms: 0,
123            metadata: Default::default(),
124        })
125    }
126
127    async fn move_file(&self, task: &Task, source_path: &str) -> Result<TaskResult> {
128        let dest_path = task
129            .definition
130            .payload
131            .get("destination")
132            .and_then(|v| v.as_str())
133            .ok_or_else(|| {
134                TaskFlowError::InvalidConfiguration("Missing destination path".to_string())
135            })?;
136
137        fs::rename(source_path, dest_path)
138            .await
139            .map_err(|e| TaskFlowError::ExecutionError(e.to_string()))?;
140
141        Ok(TaskResult {
142            success: true,
143            output: Some(format!("File moved from {} to {}", source_path, dest_path)),
144            error: None,
145            execution_time_ms: 0,
146            metadata: Default::default(),
147        })
148    }
149}
taskflow_rs/executor/handlers/file.rs

taskflow_rs/executor/handlers/
file.rs
