diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestFilteringProcessors.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestFilteringProcessors.java index 0d313df4951..d4f5f353090 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestFilteringProcessors.java +++ b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestFilteringProcessors.java @@ -20,7 +20,7 @@ public class TestFilteringProcessors { /** * Contains the last received elements as key and the amount of the received times as value; */ - private Map lastReceivedElements; + private Map receivedElements; @Test public void testFilteringProcessor() { @@ -47,7 +47,7 @@ public class TestFilteringProcessors { private void verifyThatUnexpectedElementsHasNotBeenReceived(Collection unexpectedElements) { for (Integer unexpectedElement : unexpectedElements) { - assertThat("The unexpected element '" + unexpectedElement + "' was received", lastReceivedElements.containsKey(unexpectedElement), is(false)); + assertThat("The unexpected element '" + unexpectedElement + "' was received", receivedElements.containsKey(unexpectedElement), is(false)); } } @@ -75,22 +75,22 @@ public class TestFilteringProcessors { private void verifyThatExpectedElementsHasBeenReceived(Iterable expectedElements) { for (Integer expectedElement : expectedElements) { - assertThat("The expected element '" + expectedElement + "' wasn't received", lastReceivedElements.containsKey(expectedElement), is(true)); + assertThat("The expected element '" + expectedElement + "' wasn't received", receivedElements.containsKey(expectedElement), is(true)); } } @Before public void initializeReceivers() { - lastReceivedElements = new HashMap<>(); + receivedElements = new HashMap<>(); Processor receiver = new Processor() { @Override public void onElement(Integer element) { - if (!lastReceivedElements.containsKey(element)) { - lastReceivedElements.put(element, 0); + if (!receivedElements.containsKey(element)) { + receivedElements.put(element, 0); } - Integer elementAmount = lastReceivedElements.get(element) + 1; - lastReceivedElements.put(element, elementAmount); + Integer elementAmount = receivedElements.get(element) + 1; + receivedElements.put(element, elementAmount); } @Override public void finish() throws InterruptedException { diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestParallelExtractionProcessor.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestParallelExtractionProcessor.java new file mode 100644 index 00000000000..9cf7fbc10de --- /dev/null +++ b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/components/TestParallelExtractionProcessor.java @@ -0,0 +1,98 @@ +package com.sap.sse.datamining.components; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; + +import com.sap.sse.datamining.factories.FunctionFactory; +import com.sap.sse.datamining.functions.Function; +import com.sap.sse.datamining.impl.components.GroupedDataEntry; +import com.sap.sse.datamining.impl.components.ParallelGroupedElementsValueExtractionProcessor; +import com.sap.sse.datamining.shared.GroupKey; +import com.sap.sse.datamining.shared.impl.GenericGroupKey; +import com.sap.sse.datamining.test.components.util.Number; +import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil; +import com.sap.sse.datamining.test.util.FunctionTestsUtil; + +public class TestParallelExtractionProcessor { + + private Collection>> receivers; + private Map receivedValues; + + private Function getCrossSumFunction; + + @Before + public void initializeReceivers() { + receivedValues = new HashMap<>(); + + Processor> receiver = new Processor>() { + @Override + public void onElement(GroupedDataEntry element) { + receivedValues.put(element.getKey(), element.getDataEntry()); + } + @Override + public void finish() throws InterruptedException { + } + }; + + receivers = new HashSet<>(); + receivers.add(receiver); + } + + @Before + public void initializeExtractionFunction() { + Method extractionMethod = FunctionTestsUtil.getMethodFromClass(Number.class, "getCrossSum"); + getCrossSumFunction = FunctionFactory.createMethodWrappingFunction(extractionMethod); + } + + @Test + public void testValueExtraction() { + Processor> processor = new ParallelGroupedElementsValueExtractionProcessor(ConcurrencyTestsUtil.getExecutor(), receivers, getCrossSumFunction); + Collection> elements = createElements(); + processElements(processor, elements); + Collection> expectedReceivedValues = buildExpectedReceivedValues(elements); + verifyThatExpectedElementsHaveBeenReceived(expectedReceivedValues); + } + + private Collection> createElements() { + Collection> elements = new ArrayList<>(); + elements.add(new GroupedDataEntry(new GenericGroupKey(1), new Number(1))); + elements.add(new GroupedDataEntry(new GenericGroupKey(2), new Number(11))); + elements.add(new GroupedDataEntry(new GenericGroupKey(3), new Number(111))); + elements.add(new GroupedDataEntry(new GenericGroupKey(4), new Number(1111))); + return elements; + } + + private void processElements(Processor> processor, Collection> elements) { + for (GroupedDataEntry element : elements) { + processor.onElement(element); + } + } + + private Collection> buildExpectedReceivedValues(Collection> elements) { + Collection> expectedReceivedValues = new ArrayList<>(); + for (GroupedDataEntry element : elements) { + Integer value = getCrossSumFunction.tryToInvoke(element.getDataEntry()); + assertThat("The value couldn't be calculated from the function.", value != null, is(true)); + expectedReceivedValues.add(new GroupedDataEntry(element.getKey(), value)); + } + return expectedReceivedValues; + } + + private void verifyThatExpectedElementsHaveBeenReceived(Collection> expectedReceivedValues) { + for (GroupedDataEntry expectedReceivedValue : expectedReceivedValues) { + assertThat("The expected value '" + expectedReceivedValue + "' wasn't received.", receivedValues.containsKey(expectedReceivedValue.getKey()), is(true)); + assertThat(receivedValues.get(expectedReceivedValue.getKey()), is(expectedReceivedValue.getDataEntry())); + } + } + +} diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java new file mode 100644 index 00000000000..7c16515d155 --- /dev/null +++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/ParallelGroupedElementsValueExtractionProcessor.java @@ -0,0 +1,32 @@ +package com.sap.sse.datamining.impl.components; + +import java.util.Collection; +import java.util.concurrent.ThreadPoolExecutor; + +import com.sap.sse.datamining.components.Processor; +import com.sap.sse.datamining.functions.Function; + +public class ParallelGroupedElementsValueExtractionProcessor + extends AbstractSimpleParallelProcessor, GroupedDataEntry> { + + private final Function extractionFunction; + + public ParallelGroupedElementsValueExtractionProcessor(ThreadPoolExecutor executor, + Collection>> resultReceivers, + Function extractionFunction) { + super(executor, resultReceivers); + this.extractionFunction = extractionFunction; + } + + @Override + protected Runnable createInstruction(GroupedDataEntry element) { + return new AbstractDirectForwardProcessingInstruction, GroupedDataEntry>(element, getResultReceivers()) { + @Override + protected GroupedDataEntry doWork() { + FunctionReturnType value = extractionFunction.tryToInvoke(super.getInput().getDataEntry()); + return value != null ? new GroupedDataEntry(super.getInput().getKey(), value) : super.getInvalidResult(); + } + }; + } + +}