Simple pipeline implementation to supply fluent and readable DSLs for business workflows through using Java 8 functional features

PipelineStep.java

@FunctionalInterface
public interface PipelineStep<I, O, Ctx> {
    static <I, O, Ctx> PipelineStep<I, O, Ctx> of(PipelineStep<I, O, Ctx> source) {
        return source;
    }

    O execute(I value, Ctx context) throws Exception;

    default <R> PipelineStep<I, R, Ctx> pipe(PipelineStep<O, R, Ctx> next) {
        return (value, ctx) -> next.execute(execute(value, ctx), ctx);
    }
}

Usage.java

public class ProductImportPipeline {

    private final ParseExcelStep parseExcelStep;
    private final FilterInvalidRecordsStep filterInvalidRecordsStep;
    private final PersistRecordsStep persistRecordsStep;
    private final SendResultAsEmailStep sendResultAsEmailStep;

    public ProductImportPipeline(
                          ParseExcelStep parseExcelStep,
                          FilterInvalidRecordsStep filterInvalidRecordsStep,
                          PersistRecordsStep persistRecordsStep,
                          SendResultAsEmailStep sendResultAsEmailStep) {
        this.parseExcelStep = parseExcelStep;
        this.filterInvalidRecordsStep = filterInvalidRecordsStep;
        this.persistRecordsStep = persistRecordsStep;
        this.sendResultAsEmailStep = sendResultAsEmailStep;
    }

    public void begin(String excelLocation) throws Exception {
        PipelineStep
                .of(parseExcelStep)
                .pipe(filterInvalidRecordsStep)
                .pipe(persistRecordsStep)
                .pipe(sendResultAsMailStep)
                .execute(excelLocation, new ImportContext(/*fill data regarding whole pipeline*/));
    }
}

public class ParseExcelStep implements PipelineStep<String, Collection<ExcelRowModel>, ImportContext> {
    public Collection<ExcelRowModel> execute(String excelLocation, ImportContext context) {
      // parse excel ...
      return excelRows;
    }
}

public class FilterInvalidRecordsStep implements PipelineStep<Collection<ExcelRowModel>, Collection<ExcelRowModel>, ImportContext> {
    public Collection<ExcelRowModel> execute(Collection<ExcelRowModel> excelRows, ImportContext context) {
      // filter excelRows ...
      return filteredExcelRows;
    }
}

public class PersistRecordsStep implements PipelineStep<Collection<ExcelRowModel>, Void, ImportContext> {
    public Void execute(Collection<ExcelRowModel> excelRows, ImportContext context) {
      // persist filteredExcelRows ...
    }
}

public class SendResultAsMailStep implements PipelineStep<Void, Void, ImportContext> {
    public Void execute(Void value, ImportContext context) {
      // send notification email ...
    }
}