#207 added capacity to IPipe

src/main/java/teetime/framework/AbstractInterThreadPipe.java

@@ -38,8 +38,8 @@ public abstract class AbstractInterThreadPipe extends AbstractPipe {
 
 	private volatile boolean closed;
 
-	protected <T> AbstractInterThreadPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		super(sourcePort, targetPort);
+	protected <T> AbstractInterThreadPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
+		super(sourcePort, targetPort, capacity);
 		final Queue<ISignal> localSignalQueue = QueueFactory.newQueue(new ConcurrentQueueSpec(1, 1, 0, Ordering.FIFO, Preference.THROUGHPUT));
 		final PutStrategy<ISignal> putStrategy = new YieldPutStrategy<ISignal>();
 		final TakeStrategy<ISignal> takeStrategy = new SCParkTakeStrategy<ISignal>();

src/main/java/teetime/framework/AbstractIntraThreadPipe.java

@@ -21,8 +21,8 @@ public abstract class AbstractIntraThreadPipe extends AbstractPipe {
 
 	private boolean closed;
 
-	protected <T> AbstractIntraThreadPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		super(sourcePort, targetPort);
+	protected <T> AbstractIntraThreadPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
+		super(sourcePort, targetPort, capacity);
 	}
 
 	@Override

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

@@ -30,8 +30,10 @@ public abstract class AbstractPipe implements IPipe {
 
 	private final OutputPort<?> sourcePort;
 	private final InputPort<?> targetPort;
+	@SuppressWarnings("PMD.AvoidFieldNameMatchingMethodName")
+	private final int capacity;
 
-	protected <T> AbstractPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
+	protected <T> AbstractPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
 		if (sourcePort == null) {
 			throw new IllegalArgumentException("sourcePort may not be null");
 		}
@@ -44,16 +46,17 @@ public abstract class AbstractPipe implements IPipe {
 
 		this.sourcePort = sourcePort;
 		this.targetPort = targetPort;
+		this.capacity = capacity;
 		this.cachedTargetStage = targetPort.getOwningStage();
 	}
 
 	@Override
-	public OutputPort<?> getSourcePort() {
+	public final OutputPort<?> getSourcePort() {
 		return sourcePort;
 	}
 
 	@Override
-	public InputPort<?> getTargetPort() {
+	public final InputPort<?> getTargetPort() {
 		return targetPort;
 	}
 
@@ -61,4 +64,9 @@ public abstract class AbstractPipe implements IPipe {
 	public final boolean hasMore() {
 		return !isEmpty();
 	}
+
+	@Override
+	public final int capacity() {
+		return capacity;
+	}
 }

src/main/java/teetime/framework/ExecutionInstantiation.java

@@ -72,8 +72,8 @@ class ExecutionInstantiation {
 		}
 
 		if (threadableStages.contains(targetStage) && targetColor != color) {
-			if (pipe.getCapacity() != 0) {
-				interBoundedThreadPipeFactory.create(outputPort, pipe.getTargetPort(), pipe.getCapacity());
+			if (pipe.capacity() != 0) {
+				interBoundedThreadPipeFactory.create(outputPort, pipe.getTargetPort(), pipe.capacity());
 			} else {
 				interUnboundedThreadPipeFactory.create(outputPort, pipe.getTargetPort(), 4);
 			}

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

@@ -95,4 +95,9 @@ public final class DummyPipe implements IPipe {
 
 	}
 
+	@Override
+	public int capacity() {
+		return 0;
+	}
+
 }

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

@@ -23,6 +23,8 @@ public interface IMonitorablePipe {
 
 	int size();
 
+	int capacity();
+
 	long getPushThroughput();
 
 	long getPullThroughput();

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

@@ -50,10 +50,15 @@ public interface IPipe {
 	boolean isEmpty();
 
 	/**
-	 * @return the current number of elements
+	 * @return the current number of elements held by this pipe instance
 	 */
 	int size();
 
+	/**
+	 * @return the maximum number of elements possible to hold by this pipe instance
+	 */
+	int capacity();
+
 	/**
 	 * Retrieves the last element of the pipe and deletes it.
 	 *

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

@@ -35,7 +35,8 @@ public class InstantiationPipe implements IPipe {
 		targetPort.setPipe(this);
 	}
 
-	public int getCapacity() {
+	@Override
+	public int capacity() {
 		return capacity;
 	}
 

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

@@ -24,7 +24,7 @@ final class RelayTestPipe<T> extends AbstractInterThreadPipe {
 	private final ConstructorClosure<T> inputObjectCreator;
 
 	public RelayTestPipe(final int numInputObjects, final ConstructorClosure<T> inputObjectCreator) {
-		super(null, null);
+		super(null, null, Integer.MAX_VALUE);
 		this.numInputObjects = numInputObjects;
 		this.inputObjectCreator = inputObjectCreator;
 	}

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

@@ -24,7 +24,7 @@ final class SingleElementPipe extends AbstractIntraThreadPipe {
 	private Object element;
 
 	<T> SingleElementPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		super(sourcePort, targetPort);
+		super(sourcePort, targetPort, 1);
 	}
 
 	@Override

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

@@ -30,7 +30,7 @@ final class SpScPipe extends AbstractInterThreadPipe implements IMonitorablePipe
 	private int numWaits;
 
 	<T> SpScPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
-		super(sourcePort, targetPort);
+		super(sourcePort, targetPort, capacity);
 		this.queue = new ObservableSpScArrayQueue<Object>(capacity);
 	}
 

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

@@ -31,7 +31,7 @@ final class UnboundedSpScPipe extends AbstractInterThreadPipe {
 	private final Queue<Object> queue;
 
 	<T> UnboundedSpScPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
-		super(sourcePort, targetPort);
+		super(sourcePort, targetPort, Integer.MAX_VALUE);
 		ConcurrentQueueSpec specification = new ConcurrentQueueSpec(1, 1, 0, Ordering.FIFO, Preference.THROUGHPUT);
 		this.queue = QueueFactory.newQueue(specification);
 	}

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

@@ -69,6 +69,11 @@ class MergerTestingPipe implements IPipe {
 		return 0;
 	}
 
+	@Override
+	public int capacity() {
+		return 0;
+	}
+
 	@Override
 	public Object removeLast() {
 		return null;