Merge branch 'active-declaration' into 'master'

Active declaration

fixes #174
renamed addThreadableStage to declareActive and added a flag to Stage indicating if the current stage is active or not

See merge request !54

src/main/java/teetime/framework/AbstractCompositeStage.java

@@ -38,8 +38,8 @@ public abstract class AbstractCompositeStage {
 	 * @param stage
 	 *            A arbitrary stage, which will be added to the configuration and executed in a thread.
 	 */
-	protected final void addThreadableStage(final Stage stage) {
-		this.addThreadableStage(stage, stage.getId());
+	protected final void declareActive(final Stage stage) {
+		this.declareActive(stage, stage.getId());
 	}
 
 	/**
@@ -50,10 +50,11 @@ public abstract class AbstractCompositeStage {
 	 * @param threadName
 	 *            A string which can be used for debugging.
 	 */
-	protected void addThreadableStage(final Stage stage, final String threadName) {
+	protected void declareActive(final Stage stage, final String threadName) {
 		AbstractRunnableStage runnable = AbstractRunnableStage.create(stage);
 		Thread newThread = new TeeTimeThread(runnable, threadName);
 		stage.setOwningThread(newThread);
+		stage.setActive(true);
 	}
 
 	/**
@@ -85,7 +86,7 @@ public abstract class AbstractCompositeStage {
 	protected <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
 		if (sourcePort.getOwningStage().getInputPorts().size() == 0) {
 			if (sourcePort.getOwningStage().getOwningThread() == null) {
-				addThreadableStage(sourcePort.getOwningStage(), sourcePort.getOwningStage().getId());
+				declareActive(sourcePort.getOwningStage(), sourcePort.getOwningStage().getId());
 			}
 		}
 

src/main/java/teetime/framework/Configuration.java

@@ -56,9 +56,9 @@ public abstract class Configuration extends AbstractCompositeStage {
 	}
 
 	@Override
-	protected void addThreadableStage(final Stage stage, final String threadName) {
+	protected void declareActive(final Stage stage, final String threadName) {
 		startStage = stage; // memorize an arbitrary stage as starting point for traversing
-		super.addThreadableStage(stage, threadName);
+		super.declareActive(stage, threadName);
 	}
 
 	@Override

src/main/java/teetime/framework/ConfigurationContext.java

@@ -19,7 +19,7 @@ import java.util.Set;
 
 /**
  * Represents a context that is used by a configuration and composite stages to connect ports, for example.
- * Stages can be added by executing {@link #addThreadableStage(Stage)}.
+ * Stages can be added by executing {@link #declareActive(Stage)}.
  *
  * @since 2.0
  */

src/main/java/teetime/framework/Stage.java

@@ -50,6 +50,8 @@ public abstract class Stage {
 	/** The owning thread of this stage if this stage is directly executed by a {@link AbstractRunnableStage}, <code>null</code> otherwise. */
 	private Thread owningThread;
 
+	private boolean isActive;
+
 	private ConfigurationContext owningContext;
 
 	ConfigurationContext getOwningContext() {
@@ -181,4 +183,12 @@ public abstract class Stage {
 
 	protected abstract void removeDynamicPort(InputPort<?> inputPort);
 
+	public boolean isActive() {
+		return isActive;
+	}
+
+	void setActive(final boolean isActive) {
+		this.isActive = isActive;
+	}
+
 }

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

@@ -61,7 +61,7 @@ class ThreadService extends AbstractService<ThreadService> {
 	}
 
 	void startStageAtRuntime(final Stage newStage) {
-		configuration.addThreadableStage(newStage);
+		configuration.declareActive(newStage);
 
 		Set<Stage> newThreadableStages = initialize(newStage);
 		startThreads(newThreadableStages);

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

@@ -91,7 +91,7 @@ public final class StageTester {
 				connectPorts(producer.getOutputPort(), inputHolder.getPort());
 			}
 
-			addThreadableStage(stage);
+			declareActive(stage);
 
 			for (OutputHolder<?> outputHolder : outputHolders) {
 				final CollectorSink<Object> sink = new CollectorSink<Object>(outputHolder.getOutputElements());

src/test/java/teetime/examples/wordcounter/WordCounterConfiguration.java

@@ -83,7 +83,7 @@ public class WordCounterConfiguration extends Configuration {
 			connectPorts(distributor.getNewOutputPort(), threadableStage.getInputPort(), 1000);
 			connectPorts(wc.getOutputPort(), merger.getNewInputPort());
 			// Add WordCounter as a threadable stage, so it runs in its own thread
-			addThreadableStage(threadableStage.getInputPort().getOwningStage());
+			declareActive(threadableStage.getInputPort().getOwningStage());
 
 			distributorPorts.add(threadableStage.getInputPort());
 			mergerPorts.add(wc.getOutputPort());
@@ -95,8 +95,8 @@ public class WordCounterConfiguration extends Configuration {
 		connectPorts(merger.getOutputPort(), result.getInputPort());
 
 		// Add the first and last part to the threadable stages
-		addThreadableStage(init);
-		addThreadableStage(merger);
+		declareActive(init);
+		declareActive(merger);
 	}
 
 	public MonitoringThread getMonitoringThread() {

src/test/java/teetime/framework/AbstractCompositeStageTest.java

@@ -27,17 +27,17 @@ public class AbstractCompositeStageTest {
 	@Ignore
 	@Test
 	public void testNestedStages() {
-		Execution<NestesConfig> exec = new Execution<NestesConfig>(new NestesConfig());
+		Execution<NestedConf> exec = new Execution<NestedConf>(new NestedConf());
 		// assertThat(exec.getConfiguration().getContext().getThreadableStages().size(), is(3));
 	}
 
-	private class NestesConfig extends Configuration {
+	private class NestedConf extends Configuration {
 
 		private final InitialElementProducer<Object> init;
 		private final Sink sink;
 		private final TestNestingCompositeStage compositeStage;
 
-		public NestesConfig() {
+		public NestedConf() {
 			init = new InitialElementProducer<Object>(new Object());
 			sink = new Sink();
 			compositeStage = new TestNestingCompositeStage();
@@ -52,7 +52,7 @@ public class AbstractCompositeStageTest {
 		private final Counter firstCounter = new Counter();
 
 		public TestCompositeOneStage() {
-			addThreadableStage(firstCounter);
+			declareActive(firstCounter);
 		}
 
 	}
@@ -63,7 +63,7 @@ public class AbstractCompositeStageTest {
 		private final Counter secondCounter = new Counter();
 
 		public TestCompositeTwoStage() {
-			addThreadableStage(firstCounter);
+			declareActive(firstCounter);
 			connectPorts(firstCounter.getOutputPort(), secondCounter.getInputPort());
 		}
 

src/test/java/teetime/framework/ExecutionTest.java

@@ -118,7 +118,7 @@ public class ExecutionTest {
 		public AnalysisTestConfig(final boolean inter) {
 			connectPorts(init.getOutputPort(), sink.getInputPort());
 			if (inter) {
-				addThreadableStage(sink);
+				declareActive(sink);
 			}
 		}
 	}
@@ -143,7 +143,7 @@ public class ExecutionTest {
 			connectPorts(init.getOutputPort(), iof.getInputPort());
 			connectPorts(iof.getMatchedOutputPort(), sink.getInputPort());
 			connectPorts(init.createOutputPort(), sink.createInputPort());
-			addThreadableStage(iof);
+			declareActive(iof);
 		}
 	}
 
@@ -191,7 +191,7 @@ public class ExecutionTest {
 			stageWithNamedThread = new InitialElementProducer<Object>(new Object());
 			Sink<Object> sink = new Sink<Object>();
 
-			addThreadableStage(stageWithNamedThread, "TestName");
+			declareActive(stageWithNamedThread, "TestName");
 
 			connectPorts(stageWithNamedThread.getOutputPort(), sink.getInputPort());
 		}

src/test/java/teetime/framework/RunnableConsumerStageTestConfiguration.java

@@ -30,11 +30,11 @@ public class RunnableConsumerStageTestConfiguration extends Configuration {
 	public RunnableConsumerStageTestConfiguration(final Integer... inputElements) {
 		InitialElementProducer<Integer> producer = new InitialElementProducer<Integer>(inputElements);
 		if (inputElements.length > 0) {
-			addThreadableStage(producer);
+			declareActive(producer);
 		}
 
 		CollectorSink<Integer> collectorSink = new CollectorSink<Integer>(collectedElements);
-		addThreadableStage(collectorSink);
+		declareActive(collectorSink);
 
 		// Can not use createPorts, as the if condition above will lead to an exception
 		new SpScPipeFactory().create(producer.getOutputPort(), collectorSink.getInputPort());

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

@@ -18,7 +18,9 @@ package teetime.framework;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -53,6 +55,13 @@ public class StageTest {
 		assertEquals(tc.init.exceptionListener, tc.delay.exceptionListener);
 	}
 
+	@Test
+	public void testActiveFlag() {
+		TestConfig config = new TestConfig();
+		assertTrue(config.init.isActive());
+		assertFalse(config.delay.isActive());
+	}
+
 	private static class TestConfig extends Configuration {
 		public final DelayAndTerminate delay;
 		public InitialElementProducer<String> init;

src/test/java/teetime/framework/TerminationTest.java

@@ -55,11 +55,11 @@ public class TerminationTest {
 				connectPorts(init.getOutputPort(), firstProp.getInputPort());
 				connectPorts(firstProp.getOutputPort(), sinkStage.getInputPort(), capacity);
 				connectPorts(sinkStage.getOutputPort(), finalProp.getInputPort());
-				addThreadableStage(sinkStage);
+				declareActive(sinkStage);
 			} else {
 				Sink<Integer> sink = new Sink<Integer>();
 				connectPorts(init.getOutputPort(), sink.getInputPort(), capacity);
-				addThreadableStage(sink);
+				declareActive(sink);
 			}
 		}
 

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

@@ -86,14 +86,14 @@ public class TraverserTest {
 				connectPorts(distributor.getNewOutputPort(), wc.getInputPort());
 				connectPorts(wc.getOutputPort(), merger.getNewInputPort());
 				// Add WordCounter as a threadable stage, so it runs in its own thread
-				addThreadableStage(wc.getInputPort().getOwningStage());
+				declareActive(wc.getInputPort().getOwningStage());
 			}
 
 			// Connect the stages of the last part
 			connectPorts(merger.getOutputPort(), result.getInputPort());
 
 			// Add the first and last part to the threadable stages
-			addThreadableStage(merger);
+			declareActive(merger);
 		}
 
 	}

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

@@ -29,13 +29,13 @@ class WaitStrategyConfiguration extends Configuration {
 	public WaitStrategyConfiguration(final long initialDelayInMs, final Object... elements) {
 
 		Stage producer = buildProducer(elements);
-		addThreadableStage(producer);
+		declareActive(producer);
 
 		Stage consumer = buildConsumer(delay);
-		addThreadableStage(consumer);
+		declareActive(consumer);
 
 		Clock clock = buildClock(initialDelayInMs, delay);
-		addThreadableStage(clock);
+		declareActive(clock);
 	}
 
 	private Clock buildClock(final long initialDelayInMs, final Delay<Object> delay) {

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

@@ -26,10 +26,10 @@ class YieldStrategyConfiguration extends Configuration {
 	public YieldStrategyConfiguration(final Object... elements) {
 
 		InitialElementProducer<Object> producer = buildProducer(elements);
-		addThreadableStage(producer);
+		declareActive(producer);
 
 		Stage consumer = buildConsumer(producer);
-		addThreadableStage(consumer);
+		declareActive(consumer);
 	}
 
 	private InitialElementProducer<Object> buildProducer(final Object... elements) {

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

@@ -33,8 +33,8 @@ public class ExceptionTestConfiguration extends Configuration {
 		connectPorts(first.getOutputPort(), second.getInputPort());
 		// this.addThreadableStage(new ExceptionTestStage());
 
-		this.addThreadableStage(second);
-		this.addThreadableStage(third);
+		this.declareActive(second);
+		this.declareActive(third);
 	}
 
 }

src/test/java/teetime/stage/basic/distributor/dynamic/DynamicDistributorTest.java

@@ -137,8 +137,8 @@ public class DynamicDistributorTest {
 			connectPorts(initialElementProducer.getOutputPort(), distributor.getInputPort());
 			connectPorts(distributor.getNewOutputPort(), collectorSink.getInputPort());
 
-			addThreadableStage(distributor);
-			addThreadableStage(collectorSink);
+			declareActive(distributor);
+			declareActive(collectorSink);
 
 			for (PortAction<DynamicDistributor<T>> a : inputActions) {
 				distributor.addPortActionRequest(a);

src/test/java/teetime/stage/basic/merger/dynamic/DynamicMergerTest.java

@@ -118,7 +118,7 @@ public class DynamicMergerTest {
 			connectPorts(initialElementProducer.getOutputPort(), merger.getNewInputPort());
 			connectPorts(merger.getOutputPort(), collectorSink.getInputPort());
 
-			addThreadableStage(merger);
+			declareActive(merger);
 
 			for (PortAction<DynamicMerger<T>> a : inputActions) {
 				boolean added = merger.addPortActionRequest(a);