java - Pipeline design pattern implementation

Question

This is a design question regarding the implementation of a Pipeline. The following is my naive implementation.

Interface for individual steps/stages in the pipeline:

public interface Step<T, U> {
    public U execute(T input);
}

Concrete implementations of steps/stages in pipeline:

public class StepOne implements Step<Integer, Integer> {
    @Override
    public Integer execute(Integer input) {
        return input + 100;
    }
}

public class StepTwo implements Step<Integer, Integer> {
    @Override
    public Integer execute(Integer input) {
        return input + 500;
    }
}

public class StepThree implements Step<Integer, String> {
    @Override
    public String execute(Integer input) {
        return "The final amount is " + input;
    }
}

The pipeline class will hold/register the steps in the pipeline and execute them one after the other:

public class Pipeline {
    private List<Step> pipelineSteps = new ArrayList<>();
    private Object firstStepInput = 100;

    public void addStep(Step step) {
        pipelineSteps.add(step);
    }

    public void execute() {
        for (Step step : pipelineSteps) {
            Object out = step.execute(firstStepInput);
            firstStepInput = out;
        }
   }
}

Diver program to execute the pipeline:

public class Main {
    public static void main(String[] args) {
        Pipeline pipeline = new Pipeline();
        pipeline.addStep(new StepOne());
        pipeline.addStep(new StepTwo());
        pipeline.addStep(new StepThree());

        pipeline.execute();
    } 
}

However, as you can see the naive implementation has many limitations.

One of the major ones is that since the requirement is that the output of each step could be of any type, the naive implementation is not type-safe (the execute method in the Pipeline class). If I happen to wire the steps in the pipeline incorrectly, the app will fail.

Can anyone help me design the solution by adding to what I have coded, or point me towards an already existing pattern to solve this?

question from:https://stackoverflow.com/questions/39947155/pipeline-design-pattern-implementation

Answer 1

I would focus on

If I happen to wire the steps in the pipeline incorrectly, the app will fail.

Yes, this is a problem. StepThree is the stranger here. I do not think one simple pattern might help, I do think it must be a combination of strategy and builder pattern. For example:

Pipeline<Integer,Integer> intPipe = new Pipeline<>();
intPipe = intPipe.add(new StepOne()); // increment 100
intPipe = intPipe.add(new StepTwo()); // increment 500
Pipeline<String, Integer> strPipe = intPipe.add(new StepThree()); // convert

Whereat Pipeline is like this:

public static class Pipeline<IN, OUT> {
   //...
   public<A> Pipeline<OUT,A> add(Step<IN,A> step) {
     pipelineSteps.add(step);
     return (Pipeline<OUT,A>)this;
   }
}

Using the fast-builder-syntax this might work:

Pipeline<String, Integer> pipe = new Pipeline<Integer, Integer>()
    .add(new StepOne()).add(new StepTwo()).add(new StepThree());

This should work since generics are not part of the bytecode.