From d8e1b1b64b1784f02f29bb1fc6f93c9e0d1c6bb4 Mon Sep 17 00:00:00 2001
From: Lennart Hensler <LennartHensler@aol.com>
Date: Fri, 21 Feb 2014 21:18:58 +0100
Subject: [PATCH] Added ProcessingInstruction and an abstract implementation

---
 java/com.sap.sse.datamining.test/.classpath   |  2 +-
 ...DirectForwardingProcessingInstruction.java | 46 +++++++++++++++++++
 .../components/ProcessingInstruction.java     |  6 +++
 .../sse/datamining/components/Processor.java  |  4 +-
 ...actDirectForwardProcessingInstruction.java | 39 ++++++++++++++++
 .../components/AbstractParallelProcessor.java |  8 ++--
 6 files changed, 98 insertions(+), 7 deletions(-)
 create mode 100644 java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/components/TestAbstractDirectForwardingProcessingInstruction.java
 create mode 100644 java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/ProcessingInstruction.java
 create mode 100644 java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/AbstractDirectForwardProcessingInstruction.java

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 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <classpath>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.7"/>
 	<classpathentry kind="con" path="org.eclipse.pde.core.requiredPlugins"/>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.7"/>
 	<classpathentry kind="src" path="src"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>
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<Integer> processor = new Processor<Integer>() {
+			@Override
+			public void onElement(Integer element) {
+				calls++;
+			}
+		};
+		ProcessingInstruction<Integer, Integer> instruction = createInstruction(processor);
+		
+		instruction.run();
+		assertThat(calls, is(1));
+	}
+
+	private ProcessingInstruction<Integer, Integer> createInstruction(
+			Processor<Integer> resultReceiver) {
+		Integer input = 10;
+		Collection<Processor<Integer>> resultReceivers = new ArrayList<>();
+		resultReceivers.add(resultReceiver);
+		ProcessingInstruction<Integer, Integer> instruction =
+				new AbstractDirectForwardProcessingInstruction<Integer, Integer>(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<InputType, ResultType> 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<ElementType> {
+public interface Processor<InputType> {
 	
-	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<InputType, ResultType>
+			 implements ProcessingInstruction<InputType, ResultType> {
+	
+	private final InputType input;
+	private ResultType result;
+	private final Collection<Processor<ResultType>> resultReceivers;
+
+	public AbstractDirectForwardProcessingInstruction(InputType input, Collection<Processor<ResultType>> resultReceivers) {
+		this.input = input;
+		this.resultReceivers = resultReceivers;
+	}
+
+	@Override
+	public void run() {
+		result = processInput(input);
+		forwardResult();
+	}
+
+	private void forwardResult() {
+		for (Processor<ResultType> 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<ElementType, ResultType> implements Processor<ElementType> {
+public abstract class AbstractParallelProcessor<InputType, ResultType> implements Processor<InputType> {
 	
 	private Set<Processor<ResultType>> resultReceivers;
 	private Executor executor;
@@ -19,11 +19,11 @@ public abstract class AbstractParallelProcessor<ElementType, ResultType> impleme
 	}
 
 	@Override
-	public void onElement(ElementType element) {
-		executor.execute(createFuture(element));
+	public void onElement(InputType element) {
+		executor.execute(createProcessingInstruction(element));
 	}
 	
-	protected abstract RunnableFuture<ResultType> createFuture(ElementType element);
+	protected abstract RunnableFuture<ResultType> createProcessingInstruction(InputType element);
 
 	protected Set<Processor<ResultType>> getResultReceivers() {
 		return resultReceivers;