Refactored the filtering retrieval processer to fit the standard usage

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestProcessorQuery.java

@@ -114,7 +114,7 @@ public class TestProcessorQuery {
         Collection<Function<?>> dimensions = new ArrayList<>();
         Function<Integer> getLengthFunction = FunctionFactory.createMethodWrappingFunction(FunctionTestsUtil.getMethodFromClass(Number.class, "getLength"));
         dimensions.add(getLengthFunction);
-        Processor<Iterable<Number>> lengthGrouper = new ParallelMultiDimensionalGroupingProcessor<>(executor, Arrays.asList(crossSumExtractor), dimensions);
+        Processor<Number> lengthGrouper = new ParallelMultiDimensionalGroupingProcessor<>(executor, Arrays.asList(crossSumExtractor), dimensions);
         
         FilterCriteria<Number> retrievalFilterCriteria = new FilterCriteria<Number>() {
             @Override
@@ -122,20 +122,11 @@ public class TestProcessorQuery {
                 return element.getValue() >= 10;
             }
         };
-        Processor<Iterable<Number>> filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor<Iterable<Number>, Number, Iterable<Number>>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(lengthGrouper), retrievalFilterCriteria) {
+        Processor<Iterable<Number>> filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor<Iterable<Number>, Number>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(lengthGrouper), retrievalFilterCriteria) {
             @Override
             protected Iterable<Number> retrieveData(Iterable<Number> element) {
                 return element;
             }
-            @Override
-            protected Callable<Iterable<Number>> createInstruction(final Number filteredPartialElement) {
-                return new Callable<Iterable<Number>>() {
-                    @Override
-                    public Iterable<Number> call() throws Exception {
-                        return Arrays.asList(filteredPartialElement);
-                    }
-                };
-            }
         };
         
         query.setFirstProcessor(filteringRetrievalProcessor);

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestAbstractFilteringRetrievalProcessor.java

@@ -6,7 +6,6 @@ import static org.junit.Assert.assertThat;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
-import java.util.concurrent.Callable;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -75,20 +74,11 @@ public class TestAbstractFilteringRetrievalProcessor {
                 return element >= 0;
             }
         };
-        filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor<Iterable<Integer>, Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(resultReceiver), filterCriteria) {
+        filteringRetrievalProcessor = new AbstractFilteringRetrievalProcessor<Iterable<Integer>, Integer>(ConcurrencyTestsUtil.getExecutor(), Arrays.asList(resultReceiver), filterCriteria) {
             @Override
             protected Iterable<Integer> retrieveData(Iterable<Integer> element) {
                 return element;
             }
-            @Override
-            protected Callable<Integer> createInstruction(final Integer filteredPartialElement) {
-                return new Callable<Integer>() {
-                    @Override
-                    public Integer call() throws Exception {
-                        return filteredPartialElement;
-                    }
-                };
-            }
         };
     }
     

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/TestParallelMultiDimensionalGroupingProcessor.java

@@ -5,7 +5,6 @@ import static org.junit.Assert.assertThat;
 
 import java.lang.reflect.Method;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 
 import org.junit.Before;
@@ -25,7 +24,7 @@ import com.sap.sse.datamining.test.util.FunctionTestsUtil;
 
 public class TestParallelMultiDimensionalGroupingProcessor {
     
-    private Processor<Iterable<Number>> processor;
+    private Processor<Number> processor;
     private Collection<Processor<GroupedDataEntry<Number>>> receivers;
 
     private GroupedDataEntry<Number> groupedElement;
@@ -79,7 +78,7 @@ public class TestParallelMultiDimensionalGroupingProcessor {
     @Test
     public void testGroupKeyGeneration() {
         Number number = new Number(1111);
-        processor.onElement(Arrays.asList(number));
+        processor.onElement(number);
         ConcurrencyTestsUtil.sleepFor(200); //Giving the processor time to finish the instruction
         verifyGroupedElement(number);
     }