Added ProcessingInstruction and an abstract implementation

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>

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;
+	}
+
+}