diff --git a/java/com.sap.sse.datamining.test/.classpath b/java/com.sap.sse.datamining.test/.classpath
index 098194ca4b7..f9609b26c9e 100644
--- a/java/com.sap.sse.datamining.test/.classpath
+++ b/java/com.sap.sse.datamining.test/.classpath
@@ -1,7 +1,7 @@
-
+
diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/TestAbstractDirectForwardingProcessingInstruction.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/TestAbstractDirectForwardingProcessingInstruction.java
new file mode 100644
index 00000000000..03438865712
--- /dev/null
+++ b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/TestAbstractDirectForwardingProcessingInstruction.java
@@ -0,0 +1,46 @@
+package com.sap.sse.datamining.test.components;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.junit.Test;
+
+import com.sap.sse.datamining.components.ProcessingInstruction;
+import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.AbstractDirectForwardProcessingInstruction;
+
+public class TestAbstractDirectForwardingProcessingInstruction {
+
+ private int calls = 0;
+
+ @Test
+ public void testThatTheResultIsDirectlyForwarded() {
+ Processor processor = new Processor() {
+ @Override
+ public void onElement(Integer element) {
+ calls++;
+ }
+ };
+ ProcessingInstruction instruction = createInstruction(processor);
+
+ instruction.run();
+ assertThat(calls, is(1));
+ }
+
+ private ProcessingInstruction createInstruction(
+ Processor resultReceiver) {
+ Integer input = 10;
+ Collection> resultReceivers = new ArrayList<>();
+ resultReceivers.add(resultReceiver);
+ ProcessingInstruction instruction =
+ new AbstractDirectForwardProcessingInstruction(input , resultReceivers ) {
+ @Override
+ protected Integer processInput(Integer input) { return 0; }
+ };
+ return instruction;
+ }
+
+}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java
new file mode 100644
index 00000000000..b708aa319c2
--- /dev/null
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java
@@ -0,0 +1,6 @@
+package com.sap.sse.datamining.components;
+
+
+public interface ProcessingInstruction extends Runnable {
+
+}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/Processor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/Processor.java
index 0252f24020d..783226244fb 100644
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/Processor.java
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/Processor.java
@@ -1,7 +1,7 @@
package com.sap.sse.datamining.components;
-public interface Processor {
+public interface Processor {
- public void onElement(ElementType element);
+ public void onElement(InputType element);
}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java
new file mode 100644
index 00000000000..d9f407b807e
--- /dev/null
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java
@@ -0,0 +1,39 @@
+package com.sap.sse.datamining.impl.components;
+
+import java.util.Collection;
+
+import com.sap.sse.datamining.components.ProcessingInstruction;
+import com.sap.sse.datamining.components.Processor;
+
+/**
+ * A processing construction, that processes the input and directly forwards
+ * it to the result receivers.
+ */
+public abstract class AbstractDirectForwardProcessingInstruction
+ implements ProcessingInstruction {
+
+ private final InputType input;
+ private ResultType result;
+ private final Collection> resultReceivers;
+
+ public AbstractDirectForwardProcessingInstruction(InputType input, Collection> resultReceivers) {
+ this.input = input;
+ this.resultReceivers = resultReceivers;
+ }
+
+ @Override
+ public void run() {
+ result = processInput(input);
+ forwardResult();
+ }
+
+ private void forwardResult() {
+ for (Processor resultReceiver : resultReceivers) {
+ resultReceiver.onElement(result);
+ }
+ }
+
+ protected abstract ResultType processInput(InputType input);
+
+
+}
diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java
index 5fdbec4a13b..10263231e89 100644
--- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java
+++ b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractParallelProcessor.java
@@ -8,7 +8,7 @@ import java.util.concurrent.RunnableFuture;
import com.sap.sse.datamining.components.Processor;
-public abstract class AbstractParallelProcessor implements Processor {
+public abstract class AbstractParallelProcessor implements Processor {
private Set> resultReceivers;
private Executor executor;
@@ -19,11 +19,11 @@ public abstract class AbstractParallelProcessor impleme
}
@Override
- public void onElement(ElementType element) {
- executor.execute(createFuture(element));
+ public void onElement(InputType element) {
+ executor.execute(createProcessingInstruction(element));
}
- protected abstract RunnableFuture createFuture(ElementType element);
+ protected abstract RunnableFuture createProcessingInstruction(InputType element);
protected Set> getResultReceivers() {
return resultReceivers;