removed Stage

src/main/java/teetime/framework/ThreadService.java

@@ -44,7 +44,7 @@ class ThreadService extends AbstractService<ThreadService> {
 	private final List<Thread> infiniteProducerThreads = Collections.synchronizedList(new LinkedList<Thread>());
 
 	private final SignalingCounter runnableCounter = new SignalingCounter();
-	private final Set<Stage> threadableStages = Collections.synchronizedSet(new HashSet<Stage>());
+	private final Set<AbstractStage> threadableStages = Collections.synchronizedSet(new HashSet<AbstractStage>());
 
 	private final Configuration configuration;
 
@@ -54,23 +54,23 @@ class ThreadService extends AbstractService<ThreadService> {
 
 	@Override
 	void onInitialize() {
-		Stage startStage = configuration.getStartStage();
+		AbstractStage startStage = configuration.getStartStage();
 
-		Set<Stage> newThreadableStages = initialize(startStage);
+		Set<AbstractStage> newThreadableStages = initialize(startStage);
 		startThreads(newThreadableStages);
 	}
 
-	void startStageAtRuntime(final Stage newStage) {
+	void startStageAtRuntime(final AbstractStage newStage) {
 		newStage.declareActive();
 
-		Set<Stage> newThreadableStages = initialize(newStage);
+		Set<AbstractStage> newThreadableStages = initialize(newStage);
 		startThreads(newThreadableStages);
 
 		sendStartingSignal(newThreadableStages);
 	}
 
 	// extracted for runtime use
-	private Set<Stage> initialize(final Stage startStage) {
+	private Set<AbstractStage> initialize(final AbstractStage startStage) {
 		if (startStage == null) {
 			throw new IllegalStateException("The start stage may not be null.");
 		}
@@ -81,7 +81,7 @@ class ThreadService extends AbstractService<ThreadService> {
 		Traverser traversor = new Traverser(stageCollector, Direction.BOTH);
 		traversor.traverse(startStage);
 
-		Set<Stage> newThreadableStages = stageCollector.getThreadableStages();
+		Set<AbstractStage> newThreadableStages = stageCollector.getThreadableStages();
 
 		threadableStages.addAll(newThreadableStages);
 		if (threadableStages.isEmpty()) {
@@ -98,14 +98,14 @@ class ThreadService extends AbstractService<ThreadService> {
 		A4StageAttributeSetter attributeSetter = new A4StageAttributeSetter(configuration, newThreadableStages);
 		attributeSetter.setAttributes();
 
-		for (Stage stage : newThreadableStages) {
+		for (AbstractStage stage : newThreadableStages) {
 			categorizeThreadableStage(stage);
 		}
 
 		return newThreadableStages;
 	}
 
-	private void categorizeThreadableStage(final Stage stage) {
+	private void categorizeThreadableStage(final AbstractStage stage) {
 		switch (stage.getTerminationStrategy()) {
 		case BY_INTERRUPT:
 			infiniteProducerThreads.add(stage.getOwningThread());
@@ -122,15 +122,15 @@ class ThreadService extends AbstractService<ThreadService> {
 		}
 	}
 
-	private void startThreads(final Set<Stage> threadableStages) {
-		for (Stage stage : threadableStages) {
+	private void startThreads(final Set<AbstractStage> threadableStages) {
+		for (AbstractStage stage : threadableStages) {
 			stage.getOwningThread().start();
 		}
 	}
 
-	private void sendStartingSignal(final Set<Stage> newThreadableStages) {
+	private void sendStartingSignal(final Set<AbstractStage> newThreadableStages) {
 		synchronized (newThreadableStages) {
-			for (Stage stage : newThreadableStages) {
+			for (AbstractStage stage : newThreadableStages) {
 				((TeeTimeThread) stage.getOwningThread()).sendStartingSignal();
 			}
 		}
@@ -146,9 +146,9 @@ class ThreadService extends AbstractService<ThreadService> {
 		abortStages(threadableStages);
 	}
 
-	private void abortStages(final Set<Stage> currentTreadableStages) {
+	private void abortStages(final Set<AbstractStage> currentTreadableStages) {
 		synchronized (currentTreadableStages) {
-			for (Stage stage : currentTreadableStages) {
+			for (AbstractStage stage : currentTreadableStages) {
 				stage.abort();
 			}
 		}
@@ -193,7 +193,7 @@ class ThreadService extends AbstractService<ThreadService> {
 		return exceptions;
 	}
 
-	Set<Stage> getThreadableStages() {
+	Set<AbstractStage> getThreadableStages() {
 		return threadableStages;
 	}
 

src/main/java/teetime/framework/Traverser.java

@@ -47,7 +47,7 @@ public class Traverser {
 		CONTINUE, STOP;
 	}
 
-	private final Set<Stage> visitedStages = new HashSet<Stage>();
+	private final Set<AbstractStage> visitedStages = new HashSet<AbstractStage>();
 
 	private final ITraverserVisitor traverserVisitor;
 	private final Direction direction;
@@ -61,7 +61,7 @@ public class Traverser {
 		this.direction = direction;
 	}
 
-	public void traverse(final Stage stage) {
+	public void traverse(final AbstractStage stage) {
 		VisitorBehavior behavior = traverserVisitor.visit(stage);
 		if (behavior == VisitorBehavior.STOP || !visitedStages.add(stage)) {
 			return;
@@ -93,7 +93,7 @@ public class Traverser {
 		}
 	}
 
-	public Set<Stage> getVisitedStages() {
+	public Set<AbstractStage> getVisitedStages() {
 		return visitedStages;
 	}
 }

src/main/java/teetime/framework/exceptionHandling/AbstractExceptionListener.java

@@ -21,12 +21,12 @@ import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 /**
  * Represents a minimalistic StageExceptionListener.
  * Listener which extend from this one, must a least implement this functionality.
- * This abstract class provides a Logger {@link #logger} and the method {@link #onStageException(Exception, Stage)} which is called on every raised exception.
+ * This abstract class provides a Logger {@link #logger} and the method {@link #onStageException(Exception, AbstractStage)} which is called on every raised exception.
  */
 public abstract class AbstractExceptionListener {
 
@@ -57,13 +57,13 @@ public abstract class AbstractExceptionListener {
 	 * @return
 	 * 		true, if the thread should be terminated, false otherwise
 	 */
-	public abstract FurtherExecution onStageException(Exception e, Stage throwingStage);
+	public abstract FurtherExecution onStageException(Exception e, AbstractStage throwingStage);
 
 	public List<Exception> getLoggedExceptions() {
 		return exceptionsList;
 	}
 
-	public FurtherExecution reportException(final Exception e, final Stage stage) {
+	public FurtherExecution reportException(final Exception e, final AbstractStage stage) {
 		if (logExceptions) {
 			exceptionsList.add(e);
 		}

src/main/java/teetime/framework/exceptionHandling/IgnoringExceptionListener.java

@@ -15,7 +15,7 @@
  */
 package teetime.framework.exceptionHandling;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 class IgnoringExceptionListener extends AbstractExceptionListener {
 
@@ -24,7 +24,7 @@ class IgnoringExceptionListener extends AbstractExceptionListener {
 	}
 
 	@Override
-	public FurtherExecution onStageException(final Exception e, final Stage throwingStage) {
+	public FurtherExecution onStageException(final Exception e, final AbstractStage throwingStage) {
 		return FurtherExecution.CONTINUE;
 	}
 }

src/main/java/teetime/framework/exceptionHandling/LoggingExceptionListener.java

@@ -15,7 +15,7 @@
  */
 package teetime.framework.exceptionHandling;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 class LoggingExceptionListener extends AbstractExceptionListener {
 
@@ -24,7 +24,7 @@ class LoggingExceptionListener extends AbstractExceptionListener {
 	}
 
 	@Override
-	public FurtherExecution onStageException(final Exception e, final Stage throwingStage) {
+	public FurtherExecution onStageException(final Exception e, final AbstractStage throwingStage) {
 		logger.warn("Exception occurred in " + throwingStage.getId(), e);
 		return FurtherExecution.CONTINUE;
 	}

src/main/java/teetime/framework/exceptionHandling/TerminatingExceptionListener.java

@@ -18,7 +18,7 @@ package teetime.framework.exceptionHandling;
 import java.util.ArrayList;
 import java.util.List;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 class TerminatingExceptionListener extends AbstractExceptionListener {
 
@@ -29,7 +29,7 @@ class TerminatingExceptionListener extends AbstractExceptionListener {
 	}
 
 	@Override
-	public FurtherExecution onStageException(final Exception e, final Stage throwingStage) {
+	public FurtherExecution onStageException(final Exception e, final AbstractStage throwingStage) {
 		if (logger.isWarnEnabled()) {
 			logger.warn("Exception occurred in " + throwingStage.getId(), e);
 		}

src/main/java/teetime/framework/signal/ISignal.java

@@ -19,11 +19,11 @@ import java.util.List;
 import java.util.Set;
 
 import teetime.framework.InputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 public interface ISignal {
 
-	void trigger(Stage stage) throws Exception;
+	void trigger(AbstractStage stage) throws Exception;
 
 	// Only used by the merger so far
 	boolean mayBeTriggered(Set<InputPort<?>> receivedInputPorts, List<InputPort<?>> allInputPorts);

src/main/java/teetime/framework/signal/StartingSignal.java

@@ -19,12 +19,12 @@ import java.util.List;
 import java.util.Set;
 
 import teetime.framework.InputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 public final class StartingSignal implements ISignal {
 
 	@Override
-	public void trigger(final Stage stage) throws Exception {
+	public void trigger(final AbstractStage stage) throws Exception {
 		stage.onStarting();
 	}
 

src/main/java/teetime/framework/signal/TerminatingSignal.java

@@ -19,12 +19,12 @@ import java.util.List;
 import java.util.Set;
 
 import teetime.framework.InputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 public final class TerminatingSignal implements ISignal {
 
 	@Override
-	public void trigger(final Stage stage) throws Exception {
+	public void trigger(final AbstractStage stage) throws Exception {
 		stage.onTerminating();
 	}
 

src/main/java/teetime/framework/signal/ValidatingSignal.java

@@ -20,7 +20,7 @@ import java.util.List;
 import java.util.Set;
 
 import teetime.framework.InputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 import teetime.framework.validation.InvalidPortConnection;
 
 public final class ValidatingSignal implements ISignal {
@@ -28,7 +28,7 @@ public final class ValidatingSignal implements ISignal {
 	private final List<InvalidPortConnection> invalidPortConnections = new LinkedList<InvalidPortConnection>();
 
 	@Override
-	public void trigger(final Stage stage) {
+	public void trigger(final AbstractStage stage) {
 		stage.onValidating(this.invalidPortConnections);
 	}
 

src/main/java/teetime/framework/test/InputHolder.java

@@ -16,18 +16,18 @@
 package teetime.framework.test;
 
 import teetime.framework.InputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 public final class InputHolder<I> {
 
 	private final StageTester stageTester;
-	private final Stage stage;
+	private final AbstractStage stage;
 	private final Iterable<Object> input;
 
 	private InputPort<Object> port;
 
 	@SuppressWarnings("unchecked")
-	InputHolder(final StageTester stageTester, final Stage stage, final Iterable<I> input) {
+	InputHolder(final StageTester stageTester, final AbstractStage stage, final Iterable<I> input) {
 		this.stageTester = stageTester;
 		this.stage = stage;
 		this.input = (Iterable<Object>) input;

src/main/java/teetime/framework/test/StageTester.java

@@ -22,7 +22,7 @@ import java.util.List;
 import teetime.framework.Configuration;
 import teetime.framework.Execution;
 import teetime.framework.ExecutionException;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 import teetime.framework.StageState;
 import teetime.stage.CollectorSink;
 import teetime.stage.InitialElementProducer;
@@ -36,13 +36,13 @@ public final class StageTester {
 
 	private final List<InputHolder<?>> inputHolders = new ArrayList<InputHolder<?>>();
 	private final List<OutputHolder<?>> outputHolders = new ArrayList<OutputHolder<?>>();
-	private final Stage stage;
+	private final AbstractStage stage;
 
-	private StageTester(final Stage stage) {
+	private StageTester(final AbstractStage stage) {
 		this.stage = stage;
 	}
 
-	public static StageTester test(final Stage stage) {
+	public static StageTester test(final AbstractStage stage) {
 		if (stage.getCurrentState() != StageState.CREATED) {
 			throw new AssertionError("This stage has already been tested in this test method. Move this test into a new test method.");
 		}
@@ -85,7 +85,7 @@ public final class StageTester {
 
 	private final class TestConfiguration extends Configuration {
 
-		public TestConfiguration(final List<InputHolder<?>> inputHolders, final Stage stage, final List<OutputHolder<?>> outputHolders) {
+		public TestConfiguration(final List<InputHolder<?>> inputHolders, final AbstractStage stage, final List<OutputHolder<?>> outputHolders) {
 			for (InputHolder<?> inputHolder : inputHolders) {
 				final InitialElementProducer<Object> producer = new InitialElementProducer<Object>(inputHolder.getInput());
 				connectPorts(producer.getOutputPort(), inputHolder.getPort());

src/main/java/teetime/stage/io/EveryXthPrinter.java

@@ -21,7 +21,7 @@ import java.util.List;
 import teetime.framework.AbstractCompositeStage;
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 import teetime.stage.EveryXthStage;
 import teetime.stage.basic.distributor.Distributor;
 import teetime.stage.basic.distributor.strategy.CopyByReferenceStrategy;
@@ -29,7 +29,7 @@ import teetime.stage.basic.distributor.strategy.CopyByReferenceStrategy;
 public final class EveryXthPrinter<T> extends AbstractCompositeStage {
 
 	private final Distributor<T> distributor;
-	private final List<Stage> lastStages = new ArrayList<Stage>();
+	private final List<AbstractStage> lastStages = new ArrayList<AbstractStage>();
 
 	public EveryXthPrinter(final int threshold) {
 		distributor = new Distributor<T>(new CopyByReferenceStrategy());
@@ -50,7 +50,7 @@ public final class EveryXthPrinter<T> extends AbstractCompositeStage {
 		return distributor.getNewOutputPort();
 	}
 
-	public Stage getFirstStage() {
+	public AbstractStage getFirstStage() {
 		return distributor;
 	}
 

src/main/java/teetime/util/framework/port/PortAction.java

@@ -15,9 +15,9 @@
  */
 package teetime.util.framework.port;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
-public interface PortAction<T extends Stage> {
+public interface PortAction<T extends AbstractStage> {
 
 	public abstract void execute(final T stage);
 

src/main/java/teetime/util/framework/port/PortActionHelper.java

@@ -23,7 +23,7 @@ import org.jctools.queues.spec.ConcurrentQueueSpec;
 import org.jctools.queues.spec.Ordering;
 import org.jctools.queues.spec.Preference;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 import teetime.util.framework.concurrent.queue.PCBlockingQueue;
 import teetime.util.framework.concurrent.queue.putstrategy.PutStrategy;
 import teetime.util.framework.concurrent.queue.putstrategy.YieldPutStrategy;
@@ -44,14 +44,14 @@ public final class PortActionHelper {
 		return portActions;
 	}
 
-	public static <T extends Stage> void checkForPendingPortActionRequest(final T stage, final BlockingQueue<PortAction<T>> portActions) {
+	public static <T extends AbstractStage> void checkForPendingPortActionRequest(final T stage, final BlockingQueue<PortAction<T>> portActions) {
 		PortAction<T> dynamicPortAction = portActions.poll();
 		if (null != dynamicPortAction) {
 			dynamicPortAction.execute(stage);
 		}
 	}
 
-	public static <T extends Stage> void checkBlockingForPendingPortActionRequest(final T stage, final BlockingQueue<PortAction<T>> portActions)
+	public static <T extends AbstractStage> void checkBlockingForPendingPortActionRequest(final T stage, final BlockingQueue<PortAction<T>> portActions)
 			throws InterruptedException {
 		PortAction<T> dynamicPortAction = portActions.take();
 		dynamicPortAction.execute(stage);

src/test/java/teetime/framework/StageTest.java → src/test/java/teetime/framework/AbstractStageTest.java

@@ -29,11 +29,11 @@ import teetime.stage.Cache;
 import teetime.stage.Counter;
 import teetime.stage.InitialElementProducer;
 
-public class StageTest {
+public class AbstractStageTest {
 
 	@Test
 	public void testId() {
-		Stage.clearInstanceCounters();
+		AbstractStage.clearInstanceCounters();
 
 		Counter<Object> counter0 = new Counter<Object>();
 		Counter<Object> counter1 = new Counter<Object>();
@@ -77,8 +77,6 @@ public class StageTest {
 
 		private final long delayInMs;
 
-		public boolean finished;
-
 		public DelayAndTerminate(final long delayInMs) {
 			super();
 			this.delayInMs = delayInMs;
@@ -89,8 +87,8 @@ public class StageTest {
 			try {
 				Thread.sleep(delayInMs);
 			} catch (InterruptedException e) {
+				throw new IllegalStateException(e);
 			}
-			finished = true;
 		}
 
 	}

src/test/java/teetime/framework/TraverserTest.java

@@ -45,7 +45,7 @@ public class TraverserTest {
 		Traverser traversor = new Traverser(new IntraStageCollector(tc.init));
 		traversor.traverse(tc.init);
 
-		Set<Stage> comparingStages = new HashSet<Stage>();
+		Set<AbstractStage> comparingStages = new HashSet<AbstractStage>();
 		comparingStages.add(tc.init);
 		comparingStages.add(tc.f2b);
 		comparingStages.add(tc.distributor);

src/test/java/teetime/framework/WaitStrategyConfiguration.java

@@ -28,10 +28,10 @@ class WaitStrategyConfiguration extends Configuration {
 
 	public WaitStrategyConfiguration(final long initialDelayInMs, final Object... elements) {
 
-		Stage producer = buildProducer(elements);
+		AbstractStage producer = buildProducer(elements);
 		producer.declareActive();
 
-		Stage consumer = buildConsumer(delay);
+		AbstractStage consumer = buildConsumer(delay);
 		consumer.declareActive();
 
 		Clock clock = buildClock(initialDelayInMs, delay);
@@ -47,7 +47,7 @@ class WaitStrategyConfiguration extends Configuration {
 		return clock;
 	}
 
-	private Stage buildProducer(final Object... elements) {
+	private AbstractStage buildProducer(final Object... elements) {
 		InitialElementProducer<Object> initialElementProducer = new InitialElementProducer<Object>(elements);
 		delay = new Delay<Object>();
 

src/test/java/teetime/framework/YieldStrategyConfiguration.java

@@ -28,7 +28,7 @@ class YieldStrategyConfiguration extends Configuration {
 		InitialElementProducer<Object> producer = buildProducer(elements);
 		producer.declareActive();
 
-		Stage consumer = buildConsumer(producer);
+		AbstractStage consumer = buildConsumer(producer);
 		consumer.declareActive();
 	}
 

src/test/java/teetime/framework/exceptionHandling/TestListener.java

@@ -15,7 +15,7 @@
  */
 package teetime.framework.exceptionHandling;
 
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
 
 public class TestListener extends AbstractExceptionListener {
 
@@ -26,7 +26,7 @@ public class TestListener extends AbstractExceptionListener {
 	private int numExceptionsInvoked;
 
 	@Override
-	public FurtherExecution onStageException(final Exception e, final Stage throwingStage) {
+	public FurtherExecution onStageException(final Exception e, final AbstractStage throwingStage) {
 		numExceptionsInvoked++;
 		if (numExceptionsInvoked == 2) {
 			return FurtherExecution.TERMINATE;