package de.tblsoft.search.pipeline.filter;

import de.tblsoft.search.pipeline.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

/**
 * Created by tbl on 04.11.17.
 */
public class ParallelFilter extends AbstractFilter {


    private List<Pipeline> pipelines = new ArrayList<>();

    private String sourcePipelineId;

    private ExecutorService executorService;

    public ParallelFilter(String sourcePipelineId) {
        this.sourcePipelineId = sourcePipelineId;
    }

    public void addPipeline(Pipeline pipeline) {
        pipelines.add(pipeline);
    }

    @Override
    public void init() {
        if(pipelines.size() > 0) {
            executorService = Executors.newFixedThreadPool(pipelines.size());
        }
    }

    @Override
    public PipelineContainer filter(PipelineContainer pipelineContainer) throws Exception {
        List<PipelineFutureTask<PipelineContainer>> futureTaskList = new ArrayList<>();

        for(Pipeline pipeline: pipelines) {
            futureTaskList.add(new PipelineFutureTask<>(new PipelineCallable(pipeline, pipelineContainer), pipeline));
        }


        for(PipelineFutureTask<PipelineContainer> futureTask :futureTaskList) {
            executorService.execute(futureTask);
        }


        List<PipelineContainer> results = new ArrayList<>();
        Pipeline currentPipeline = null;
        try {
            for(PipelineFutureTask<PipelineContainer> futureTask :futureTaskList) {
                currentPipeline = futureTask.getPipeline();
                PipelineContainer value = futureTask.getWithTimeout();
                results.add(value);
            }
        } catch (TimeoutException e) {
            if(currentPipeline != null) {
                pipelineContainer.error("The pipeline " + currentPipeline.getId() + " did not finished in " + currentPipeline.getTimeout() + " ms.");
            }
            pipelineContainer.error(e);
            PipelineExecuterService.failOnError(pipelineContainer);
        } catch (InterruptedException | ExecutionException e) {
            pipelineContainer.error(e);
            PipelineExecuterService.failOnError(pipelineContainer);
        }

        // TODO merge searchRequest object

        return pipelineContainer;
    }
}

@@ -1486,62 +1486,8 @@
 ();%0A
-        Pipeline currentPipeline = null;%0A        try %7B
 %0A   
@@ -1587,52 +1587,18 @@
     
-currentPipeline = futureTask.getPipeline();%0A
+try %7B%0A    
     
@@ -1680,16 +1680,20 @@
         
+    
 results.
@@ -1704,38 +1704,32 @@
 value);%0A        
-    %7D%0A
         %7D catch 
@@ -1767,38 +1767,66 @@
     
-if(currentPipeline != null) %7B%0A
+        Pipeline pipeline = futureTask.getPipeline();%0A    
     
@@ -1875,32 +1875,25 @@
 ipeline %22 + 
-currentP
+p
 ipeline.getI
@@ -1928,16 +1928,9 @@
 %22 + 
-currentP
+p
 ipel
@@ -1973,18 +1973,16 @@
     
-%7D%0A
         
     
@@ -1969,36 +1969,32 @@
                 
-    
 pipelineContaine
@@ -1997,32 +1997,40 @@
 ainer.error(e);%0A
+        
             Pipe
@@ -2073,32 +2073,40 @@
 lineContainer);%0A
+        
         %7D catch 
@@ -2157,32 +2157,40 @@
 ) %7B%0A            
+        
 pipelineContaine
@@ -2193,32 +2193,40 @@
 ainer.error(e);%0A
+        
             Pipe
@@ -2281,25 +2281,48 @@
 r);%0A        
-%7D
+        %7D%0A            %7D%0A
 %0A%0A        //

improve the error handling of the parallel filter