dataflow_rs/engine/
workflow.rs

1use crate::engine::error::{DataflowError, Result};
2use crate::engine::task::Task;
3use serde::Deserialize;
4use serde_json::Value;
5use std::fs;
6use std::path::Path;
7
8/// Workflow represents a collection of tasks that execute sequentially
9#[derive(Clone, Debug, Deserialize)]
10pub struct Workflow {
11    pub id: String,
12    pub name: String,
13    #[serde(default)]
14    pub priority: u32,
15    pub description: Option<String>,
16    #[serde(default = "default_condition")]
17    pub condition: Value,
18    #[serde(skip)]
19    pub condition_index: Option<usize>,
20    pub tasks: Vec<Task>,
21}
22
23fn default_condition() -> Value {
24    Value::Bool(true)
25}
26
27impl Default for Workflow {
28    fn default() -> Self {
29        Self::new()
30    }
31}
32
33impl Workflow {
34    pub fn new() -> Self {
35        Workflow {
36            id: String::new(),
37            name: String::new(),
38            priority: 0,
39            description: None,
40            condition: Value::Bool(true),
41            condition_index: None,
42            tasks: Vec::new(),
43        }
44    }
45
46    /// Load workflow from JSON string
47    pub fn from_json(json_str: &str) -> Result<Self> {
48        serde_json::from_str(json_str).map_err(DataflowError::from_serde)
49    }
50
51    /// Load workflow from JSON file
52    pub fn from_file<P: AsRef<Path>>(path: P) -> Result<Self> {
53        let json_str = fs::read_to_string(path).map_err(DataflowError::from_io)?;
54
55        Self::from_json(&json_str)
56    }
57
58    /// Validate the workflow structure
59    pub fn validate(&self) -> Result<()> {
60        // Check required fields
61        if self.id.is_empty() {
62            return Err(DataflowError::Workflow(
63                "Workflow id cannot be empty".to_string(),
64            ));
65        }
66
67        if self.name.is_empty() {
68            return Err(DataflowError::Workflow(
69                "Workflow name cannot be empty".to_string(),
70            ));
71        }
72
73        // Check tasks
74        if self.tasks.is_empty() {
75            return Err(DataflowError::Workflow(
76                "Workflow must have at least one task".to_string(),
77            ));
78        }
79
80        // Validate that task IDs are unique
81        let mut task_ids = std::collections::HashSet::new();
82        for task in &self.tasks {
83            if !task_ids.insert(&task.id) {
84                return Err(DataflowError::Workflow(format!(
85                    "Duplicate task ID '{}' in workflow",
86                    task.id
87                )));
88            }
89        }
90
91        Ok(())
92    }
93}
dataflow_rs/engine/workflow.rs

dataflow_rs/engine/
workflow.rs
