Use null as empty pipe element

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

@@ -36,20 +36,20 @@ public abstract class AbstractPipe<T> implements IPipe<T> {
 	private final InputPort<T> targetPort;
 
 	// FIXME each scheduler should set a default pipe scheduler by its own
-	private PipeScheduler scheduler = new PipeScheduler() {
+	private PipeScheduler<T> scheduler = new PipeScheduler<T>() {
 
 		@Override
-		public void onElementAdded(final AbstractSynchedPipe<?> pipe) {
+		public void onElementAdded(final AbstractSynchedPipe<T> pipe) {
 			// do nothing
 		}
 
 		@Override
-		public void onElementAdded(final AbstractUnsynchedPipe<?> pipe) {
+		public void onElementAdded(final AbstractUnsynchedPipe<T> pipe) {
 			// do nothing
 		}
 
 		@Override
-		public void onElementNotAdded(final AbstractSynchedPipe<?> pipe) {
+		public void onElementNotAdded(final AbstractSynchedPipe<T> pipe) {
 			// do nothing
 		}
 	};

src/main/java/teetime/framework/AbstractPort.java

@@ -26,7 +26,7 @@
  */
 public class AbstractPort<T> {
 
-	protected static final Object TERMINATE_ELEMENT = new Object();
+	protected final T TERMINATE_ELEMENT = null;
 
 	protected IPipe<T> pipe;
 	/**

src/main/java/teetime/framework/DivideAndConquerRecursivePipe.java

@@ -111,12 +111,12 @@ public void close() {
 	}
 
 	@Override
-	public boolean addNonBlocking(final Object element) {
+	public boolean addNonBlocking(final P element) {
 		return outputPipe.addNonBlocking(element);
 	}
 
 	@Override
-	public void add(final Object element) {
+	public void add(final P element) {
 		outputPipe.add(element);
 	}
 
@@ -131,7 +131,7 @@ public int size() {
 	}
 
 	@Override
-	public Object removeLast() {
+	public P removeLast() {
 		return outputPipe.removeLast();
 	}
 

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

@@ -72,7 +72,7 @@ public int size() {
 	}
 
 	@Override
-	public Object removeLast() {
+	public T removeLast() {
 		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 

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

@@ -46,7 +46,7 @@ public boolean addNonBlocking(final Object element) {
 	}
 
 	@Override
-	public Object removeLast() {
+	public T removeLast() {
 		return null;
 	}
 
@@ -61,12 +61,12 @@ public int size() {
 	}
 
 	@Override
-	public OutputPort<? extends Object> getSourcePort() {
+	public OutputPort<? extends T> getSourcePort() {
 		return null;
 	}
 
 	@Override
-	public InputPort<Object> getTargetPort() {
+	public InputPort<T> getTargetPort() {
 		return null;
 	}
 

src/main/java/teetime/framework/scheduling/globaltaskpool/BoundedMpMcSynchedPipe.java

@@ -33,25 +33,25 @@
  */
 class BoundedMpMcSynchedPipe<T> extends AbstractSynchedPipe<T> implements IMonitorablePipe {
 
-	private final MpmcArrayQueue<Object> queue;
+	private final MpmcArrayQueue<T> queue;
 
 	private transient long lastProducerIndex, lastConsumerIndex;
 
 	public BoundedMpMcSynchedPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int requestedCapacity) {
 		super(sourcePort, targetPort);
-		this.queue = new MpmcArrayQueue<Object>(requestedCapacity);
+		this.queue = new MpmcArrayQueue<T>(requestedCapacity);
 	}
 
 	@Override
-	public void add(final Object element) {
+	public void add(final T element) {
 		while (!this.queue.offer(element)) {
 			getScheduler().onElementNotAdded(this);
 		}
 		getScheduler().onElementAdded(this);
 	}
 
 	@Override
-	public boolean addNonBlocking(final Object element) {
+	public boolean addNonBlocking(final T element) {
 		return this.queue.offer(element);
 	}
 
@@ -66,7 +66,7 @@ public int size() {
 	}
 
 	@Override
-	public Object removeLast() {
+	public T removeLast() {
 		return this.queue.poll();
 	}
 

src/main/java/teetime/framework/scheduling/globaltaskpool/GlobalTaskPoolScheduling.java

@@ -45,7 +45,7 @@
  * @since 3.0
  *
  */
-public class GlobalTaskPoolScheduling implements TeeTimeScheduler, PipeScheduler, UncaughtExceptionHandler {
+public class GlobalTaskPoolScheduling<T> implements TeeTimeScheduler, PipeScheduler<T>, UncaughtExceptionHandler {
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(GlobalTaskPoolScheduling.class);
 	private static final StageFacade STAGE_FACADE = StageFacade.INSTANCE;
@@ -320,13 +320,13 @@ private void setScheduler(final AbstractStage stage) {
 	}
 
 	@Override
-	public void onElementAdded(final AbstractUnsynchedPipe<?> pipe) {
+	public void onElementAdded(final AbstractUnsynchedPipe<T> pipe) {
 		String message = String.format("This scheduler does not allow unsynched pipes: %s", pipe);
 		throw new IllegalStateException(message);
 	}
 
 	@Override
-	public void onElementAdded(final AbstractSynchedPipe<?> pipe) {
+	public void onElementAdded(final AbstractSynchedPipe<T> pipe) {
 		BoundedMpMcSynchedPipe<?> castedPipe = (BoundedMpMcSynchedPipe<?>) pipe;
 		long numPushes = castedPipe.getNumPushesSinceAppStart();
 		long lastNumPushes = castedPipe.getLastProducerIndex();
@@ -345,7 +345,7 @@ public void onElementAdded(final AbstractSynchedPipe<?> pipe) {
 	}
 
 	@Override
-	public void onElementNotAdded(final AbstractSynchedPipe<?> pipe) {
+	public void onElementNotAdded(final AbstractSynchedPipe<T> pipe) {
 		if (!taskPool.scheduleStage(pipe.getCachedTargetStage())) {
 			throw new IllegalStateException(String.format("onElementNotAdded: scheduling target stage failed for %s", pipe.getCachedTargetStage()));
 		}

src/main/java/teetime/framework/scheduling/pushpullmodel/A3PipeInstantiation.java

@@ -32,7 +32,7 @@
 /**
  * Automatically instantiates the correct pipes
  */
-class A3PipeInstantiation implements ITraverserVisitor, PipeScheduler {
+class A3PipeInstantiation<T> implements ITraverserVisitor, PipeScheduler<T> {
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(A3PipeInstantiation.class);
 
@@ -89,17 +89,17 @@ private <T> void instantiatePipe(final IPipe<T> pipe) {
 	}
 
 	@Override
-	public void onElementAdded(final AbstractUnsynchedPipe<?> pipe) {
+	public void onElementAdded(final AbstractUnsynchedPipe<T> pipe) {
 		pipe.getCachedTargetStage().executeByFramework();
 	}
 
 	@Override
-	public void onElementAdded(final AbstractSynchedPipe<?> pipe) {
+	public void onElementAdded(final AbstractSynchedPipe<T> pipe) {
 		// do nothing
 	}
 
 	@Override
-	public void onElementNotAdded(final AbstractSynchedPipe<?> pipe) {
+	public void onElementNotAdded(final AbstractSynchedPipe<T> pipe) {
 		// do nothing
 	}
 

src/test/java/teetime/stage/taskfarm/monitoring/extraction/ExtractionTestPipe.java

@@ -142,7 +142,7 @@ public boolean isEmpty() {
 	}
 
 	@Override
-	public Object removeLast() {
+	public T removeLast() {
 		return null;
 	}