stepflow_step/
step.rs

1use stepflow_base::{generate_id_type, IdError, ObjectStoreContent};
2use stepflow_data::{StateData, var::VarId};
3
4generate_id_type!(StepId);
5
6#[derive(Debug)]
7/// A single step in a flow
8///
9/// A step is defined by its the required inputs to enter the step and the outputs it must fulfill to exit the step.
10/// Substeps allow for grouping of steps and are executing in order by default.
11pub struct Step {
12  pub id: StepId,
13  pub input_vars: Option<Vec<VarId>>,
14  pub output_vars: Vec<VarId>,
15
16  substep_step_ids: Option<Vec<StepId>>,
17}
18
19impl ObjectStoreContent for Step {
20    type IdType = StepId;
21
22    fn new_id(id_val: u16) -> Self::IdType {
23      StepId::new(id_val)
24    }
25
26    fn id(&self) -> &Self::IdType {
27      &self.id
28    }
29}
30
31impl Step {
32  /// Create a new step.
33  ///
34  /// If no inputs are required, pass in `None` for `input_vars`
35  pub fn new(id: StepId, input_vars: Option<Vec<VarId>>, output_vars: Vec<VarId>) -> Self {
36    Step {
37      id,
38      input_vars,
39      output_vars,
40      substep_step_ids: None,
41    }
42  }
43
44  #[cfg(test)]
45  pub fn test_new() -> Self {
46    Step::new(stepflow_test_util::test_id!(StepId), None, vec![])
47  }
48
49  pub fn get_input_vars(&self) -> &Option<Vec<VarId>> {
50    &self.input_vars
51  }
52
53  pub fn get_output_vars(&self) -> &Vec<VarId> {
54    &self.output_vars
55  }
56
57  /// Push a substep to the end of the current sub-steps
58  pub fn push_substep(&mut self, substep_step_id: StepId) {
59    match &mut self.substep_step_ids {
60      None => self.substep_step_ids = Some(vec![substep_step_id]),
61      Some(substep_step_ids) => substep_step_ids.push(substep_step_id),
62    }
63  }
64
65  /// Get the sub-step that directly follows `prev_substep_id`
66  pub fn next_substep(&self, prev_substep_id: &StepId) -> Option<&StepId> {
67    let mut skipped = false;
68    let mut iter = self.substep_step_ids
69      .as_ref()?
70      .iter()
71      .skip_while(|step_id| {
72        // find the prev, let it skip once, then stop
73        if skipped { 
74          return false;
75        }
76        if *step_id == prev_substep_id {
77          skipped = true;
78        }
79        true
80      });
81    iter.next()
82  }
83
84  pub fn first_substep(&self) -> Option<&StepId> {
85    self.substep_step_ids.as_ref()?.first()
86  }
87
88  /// Verifies that `inputs` fulfills the required inputs to enter the step
89  pub fn can_enter(&self, inputs: &StateData) -> Result<(), IdError<VarId>> {
90    // see if we're missing any inputs
91    if let Some(input_vars) = &self.input_vars {
92      let first_missing_input = input_vars.iter().find(|input_var_id| !inputs.contains(input_var_id));
93      if first_missing_input.is_some() {
94        return Err(IdError::IdMissing(first_missing_input.unwrap().clone()))
95      }
96    }
97
98    Ok(())
99  }
100
101  /// Verifies that `state_data` fulfills the required outputs to exit the step
102  pub fn can_exit(&self, state_data: &StateData) -> Result<(), IdError<VarId>> {
103    // see if we're missing any inputs
104    self.can_enter(state_data)?;
105
106    // see if we're missing any outputs
107    let first_missing_output = &self.output_vars.iter().find(|output_var_id| !state_data.contains(output_var_id));
108    if first_missing_output.is_some() {
109      return Err(IdError::IdMissing(first_missing_output.unwrap().clone()))
110    }
111
112    Ok(())
113  }
114}
115
116#[cfg(test)]
117mod tests {
118  use stepflow_base::ObjectStoreContent;
119  use super::{ Step };
120
121  #[test]
122  fn test_add_get_substep() {
123    // no substep
124    let mut step = Step::test_new();
125    assert_eq!(step.first_substep(), None);
126
127    // add one
128    let substep1 = Step::test_new();
129    step.push_substep(substep1.id().clone());
130    assert_eq!(step.first_substep().unwrap(), substep1.id());
131    assert_eq!(step.next_substep(&substep1.id()), None);
132
133    // add another
134    let substep2 = Step::test_new();
135    step.push_substep(substep2.id().clone());
136    assert_eq!(step.first_substep().unwrap(), substep1.id());
137    assert_eq!(step.next_substep(substep1.id()).unwrap(), substep2.id());
138    assert_eq!(step.next_substep(&substep2.id()), None);
139  }
140}
stepflow_step/step.rs

stepflow_step/
step.rs
