removed connectPorts from IPipe

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

@@ -26,36 +26,30 @@ public abstract class AbstractPipe implements IPipe {
 	 * this.getPipe().getTargetPort().getOwningStage()
 	 * </pre>
 	 */
-	protected Stage cachedTargetStage;
+	protected final Stage cachedTargetStage;
 
-	private InputPort<?> targetPort;
+	private final InputPort<?> targetPort;
 
 	protected <T> AbstractPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		this.targetPort = targetPort;
-		if (null != targetPort) { // BETTER remove this check if migration is completed
-			this.cachedTargetStage = targetPort.getOwningStage();
-		}
-		if (null != sourcePort) { // BETTER remove this check if migration is completed
-			sourcePort.setPipe(this);
+		if (sourcePort == null) {
+			throw new IllegalArgumentException("sourcePort may not be null");
 		}
-		if (null != targetPort) { // BETTER remove this check if migration is completed
-			targetPort.setPipe(this);
+		if (targetPort == null) {
+			throw new IllegalArgumentException("targetPort may not be null");
 		}
-	}
-
-	@Override
-	public InputPort<?> getTargetPort() {
-		return this.targetPort;
-	}
 
-	@Override
-	public <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
 		sourcePort.setPipe(this);
 		targetPort.setPipe(this);
+
 		this.targetPort = targetPort;
 		this.cachedTargetStage = targetPort.getOwningStage();
 	}
 
+	@Override
+	public InputPort<?> getTargetPort() {
+		return this.targetPort;
+	}
+
 	@Override
 	public final boolean hasMore() {
 		return !isEmpty();

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

@@ -16,7 +16,6 @@
 package teetime.framework.pipe;
 
 import teetime.framework.InputPort;
-import teetime.framework.OutputPort;
 import teetime.framework.signal.ISignal;
 
 /**
@@ -25,11 +24,8 @@ import teetime.framework.signal.ISignal;
  * @author Christian Wulf
  *
  */
-@SuppressWarnings("rawtypes")
 public final class DummyPipe implements IPipe {
 
-	public DummyPipe() {}
-
 	@Override
 	public boolean add(final Object element) {
 		return true;
@@ -63,9 +59,6 @@ public final class DummyPipe implements IPipe {
 	@Override
 	public void sendSignal(final ISignal signal) {}
 
-	@Override
-	public void connectPorts(final OutputPort sourcePort, final InputPort targetPort) {}
-
 	@Override
 	public void reportNewElement() {
 		// do nothing

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

@@ -16,7 +16,6 @@
 package teetime.framework.pipe;
 
 import teetime.framework.InputPort;
-import teetime.framework.OutputPort;
 import teetime.framework.signal.ISignal;
 
 /**
@@ -76,9 +75,6 @@ public interface IPipe {
 	 */
 	void sendSignal(ISignal signal);
 
-	@Deprecated
-	<T> void connectPorts(OutputPort<? extends T> sourcePort, InputPort<T> targetPort);
-
 	/**
 	 * Stages report new elements with this method.
 	 */

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

@@ -21,7 +21,7 @@ import teetime.framework.signal.ISignal;
 
 public class InstantiationPipe implements IPipe {
 
-	private final InputPort target;
+	private final InputPort<?> target;
 	private final int capacity;
 
 	public <T> InstantiationPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
@@ -76,12 +76,6 @@ public class InstantiationPipe implements IPipe {
 
 	}
 
-	@Override
-	public <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		// TODO Auto-generated method stub
-
-	}
-
 	@Override
 	public void reportNewElement() {
 		// TODO Auto-generated method stub

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

@@ -27,12 +27,6 @@ final class SingleElementPipe extends AbstractIntraThreadPipe {
 		super(sourcePort, targetPort);
 	}
 
-	@Deprecated
-	public static <T> void connect(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		final IPipe pipe = new SingleElementPipe(null, null);
-		pipe.connectPorts(sourcePort, targetPort);
-	}
-
 	@Override
 	public boolean add(final Object element) {
 		if (null == element) {

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

@@ -34,13 +34,6 @@ final class SpScPipe extends AbstractInterThreadPipe implements IMonitorablePipe
 		this.queue = new ObservableSpScArrayQueue<Object>(capacity);
 	}
 
-	@Deprecated
-	public static <T> IMonitorablePipe connect(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
-		final SpScPipe pipe = new SpScPipe(sourcePort, targetPort, capacity);
-		pipe.connectPorts(sourcePort, targetPort);
-		return pipe;
-	}
-
 	// BETTER introduce a QueueIsFullStrategy
 	@Override
 	public boolean add(final Object element) {

src/main/java/teetime/stage/basic/distributor/Distributor.java

@@ -26,9 +26,9 @@ import teetime.framework.OutputPort;
  * @param T
  *            the type of the input port and the output ports
  */
-public final class Distributor<T> extends AbstractConsumerStage<T> {
+public class Distributor<T> extends AbstractConsumerStage<T> {
 
-	private IDistributorStrategy strategy;
+	protected IDistributorStrategy strategy;
 
 	public Distributor() {
 		this(new RoundRobinStrategy2());

src/test/java/teetime/framework/pipe/SpScPipeTest.java

@@ -73,7 +73,7 @@ public class SpScPipeTest {
 		assertEquals(signals, secondSignals);
 	}
 
-	@Test(expected = NullPointerException.class)
+	@Test(expected = IllegalArgumentException.class)
 	public void testAdd() throws Exception {
 		SpScPipe pipe = new SpScPipe(null, null, 4);
 		assertFalse(pipe.add(null));

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

@@ -16,7 +16,6 @@
 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;
@@ -79,11 +78,6 @@ class MergerTestingPipe implements IPipe {
 		return null;
 	}
 
-	@Override
-	public <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-
-	}
-
 	@Override
 	public void reportNewElement() {