transformation_pipeline/
pipeline.rs

1use std::io;
2use std::clone::Clone;
3use stage::TransformationStage;
4use result::StageActions;
5
6pub struct TransformationPipeline<T: Clone> {
7
8    stages: Vec<Box<TransformationStage<T>>>,
9
10}
11
12impl<T: Clone> TransformationPipeline<T> {
13    pub fn new(stages: Vec<Box<TransformationStage<T>>>) -> TransformationPipeline<T> {
14        TransformationPipeline {
15            stages: stages,
16        }
17    }
18
19    pub fn run(&self, initial_data: T) -> Result<T, io::Error> {
20
21        let mut skip_stages: u8 = 0;
22        let mut current_stage: u8 = 0;
23        let mut current_data: T = initial_data;
24
25        for stage in &self.stages {
26
27            if skip_stages > 0 {
28                skip_stages = skip_stages - 1;
29                current_stage = current_stage + 1;
30                continue;
31            }
32
33            let stage_result: StageActions<T> = stage.run(current_data.clone())?;
34            match stage_result {
35                StageActions::Abort => {
36                    return Err(io::Error::new(io::ErrorKind::Other, "Stage aborted."));
37                },
38                StageActions::Skip => {
39                },
40                StageActions::Next(stage_result_data) => {
41                    current_data = stage_result_data;
42                },
43                StageActions::Jump(stage_count, stage_result_data) => {
44                    skip_stages = stage_count;
45                    current_data = stage_result_data;
46                },
47                StageActions::Finish(stage_result_data) => {
48                    return Ok(stage_result_data);
49                },
50            }
51
52            current_stage = current_stage + 1;
53        }
54        Ok(current_data)
55    }
56}
transformation_pipeline/pipeline.rs

transformation_pipeline/
pipeline.rs
