added source port to all pipe implementations

src/main/java/teetime/framework/AbstractPipe.java

@@ -28,6 +28,7 @@ public abstract class AbstractPipe implements IPipe {
 	 */
 	protected final Stage cachedTargetStage;
 
+	private final OutputPort<?> sourcePort;
 	private final InputPort<?> targetPort;
 
 	protected <T> AbstractPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
@@ -41,13 +42,19 @@ public abstract class AbstractPipe implements IPipe {
 		sourcePort.setPipe(this);
 		targetPort.setPipe(this);
 
+		this.sourcePort = sourcePort;
 		this.targetPort = targetPort;
 		this.cachedTargetStage = targetPort.getOwningStage();
 	}
 
+	@Override
+	public OutputPort<?> getSourcePort() {
+		return sourcePort;
+	}
+
 	@Override
 	public InputPort<?> getTargetPort() {
-		return this.targetPort;
+		return targetPort;
 	}
 
 	@Override

src/main/java/teetime/framework/pipe/DummyPipe.java

@@ -16,6 +16,7 @@
 package teetime.framework.pipe;
 
 import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
 import teetime.framework.signal.ISignal;
 
 /**
@@ -51,6 +52,11 @@ public final class DummyPipe implements IPipe {
 		return 0;
 	}
 
+	@Override
+	public OutputPort<?> getSourcePort() {
+		return null;
+	}
+
 	@Override
 	public InputPort<Object> getTargetPort() {
 		return null;

src/main/java/teetime/framework/pipe/IPipe.java

@@ -16,6 +16,7 @@
 package teetime.framework.pipe;
 
 import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
 import teetime.framework.signal.ISignal;
 
 /**
@@ -61,9 +62,12 @@ public interface IPipe {
 	Object removeLast();
 
 	/**
-	 * Retrieves the receiving port.
-	 *
-	 * @return InputPort which is connected to the pipe.
+	 * @return the output port that is connected to the pipe.
+	 */
+	OutputPort<?> getSourcePort();
+
+	/**
+	 * @return the input port that is connected to the pipe.
 	 */
 	InputPort<?> getTargetPort();
 

src/main/java/teetime/framework/pipe/InstantiationPipe.java

@@ -23,10 +23,12 @@ public class InstantiationPipe implements IPipe {
 
 	private static final String ERROR_MESSAGE = "This must not be called while executing the configuration";
 
+	private final OutputPort<?> sourcePort;
 	private final InputPort<?> targetPort;
 	private final int capacity;
 
 	public <T> InstantiationPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
+		this.sourcePort = sourcePort;
 		this.targetPort = targetPort;
 		this.capacity = capacity;
 		sourcePort.setPipe(this);
@@ -37,9 +39,14 @@ public class InstantiationPipe implements IPipe {
 		return capacity;
 	}
 
+	@Override
+	public OutputPort<?> getSourcePort() {
+		return sourcePort;
+	}
+
 	@Override
 	public InputPort<?> getTargetPort() {
-		return this.targetPort;
+		return targetPort;
 	}
 
 	@Override

src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java

@@ -16,6 +16,7 @@
 package teetime.stage.basic.merger;
 
 import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
 import teetime.framework.pipe.IPipe;
 import teetime.framework.signal.ISignal;
 import teetime.framework.signal.StartingSignal;
@@ -73,6 +74,11 @@ class MergerTestingPipe implements IPipe {
 		return null;
 	}
 
+	@Override
+	public OutputPort<?> getSourcePort() {
+		return null;
+	}
+
 	@Override
 	public InputPort<?> getTargetPort() {
 		return null;