Merge remote-tracking branch 'origin/master' into dynamic-distributor

.settings/edu.umd.cs.findbugs.core.prefs

 #FindBugs User Preferences
-#Thu Jun 18 09:21:56 CEST 2015
+#Mon Jun 22 16:34:51 CEST 2015
 detector_threshold=2
 effort=max
 excludefilter0=.fbExcludeFilterFile|true

pom.xml

@@ -4,7 +4,7 @@
 
 	<groupId>net.sourceforge.teetime</groupId>
 	<artifactId>teetime</artifactId>
-	<version>1.2-SNAPSHOT</version>
+	<version>2.0-SNAPSHOT</version>
 	<packaging>jar</packaging>
 
 	<name>TeeTime</name>

src/changes/changes.xml

@@ -11,10 +11,21 @@
 				TeeTime automatically
 				chooses the correct type of pipe for all connections.
 			</action>
+			<action dev="ntd" type="add">
+				Stages without any input port are
+				automatically executed in a dedicated thread.
+			</action>
 			<action dev="ntd" type="fix" issue="93">
 				Introduced a new concept
 				for composing stages.
 			</action>
+			<action dev="ntd" type="add" issue="171">
+				Configurations are now
+				built within an AnalysisContext which is passed on to nested
+				CompositeStages.
+				This removes any constraints on CompositeStages and
+				enables therefore multiple connections and multithreading.
+			</action>
 			<action dev="ntd" type="remove">
 				Marked Pair class as deprecated.
 			</action>

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

@@ -23,8 +23,34 @@ package teetime.framework;
  *
  *
  */
-public abstract class AbstractCompositeStage extends AnalysisConfiguration {
+public abstract class AbstractCompositeStage extends Configuration {
 
-	protected abstract Stage getFirstStage();
+	private final ConfigurationContext context;
+
+	public AbstractCompositeStage(final ConfigurationContext context) {
+		if (null == context) {
+			throw new IllegalArgumentException("Context may not be null.");
+		}
+		this.context = context;
+	}
+
+	@Override
+	protected <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
+		context.connectPorts(sourcePort, targetPort, capacity);
+	}
+
+	@Override
+	protected <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
+		connectPorts(sourcePort, targetPort, 4);
+	}
+
+	@Override
+	protected void addThreadableStage(final Stage stage) {
+		context.addThreadableStage(stage);
+	}
+
+	protected ConfigurationContext getContext() {
+		return context;
+	}
 
 }

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

+package teetime.framework;
+
+public abstract class Configuration {
+
+	protected abstract <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity);
+
+	protected abstract <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort);
+
+	protected abstract void addThreadableStage(final Stage stage);
+}

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

@@ -18,6 +18,9 @@ package teetime.framework;
 import java.util.HashSet;
 import java.util.Set;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import teetime.framework.pipe.IPipe;
 import teetime.framework.pipe.IPipeFactory;
 import teetime.framework.pipe.InstantiationPipe;
@@ -29,7 +32,9 @@ import teetime.framework.pipe.PipeFactoryRegistry.ThreadCommunication;
  * Represents a configuration of connected stages, which is needed to run a analysis.
  * Stages can be added by executing {@link #addThreadableStage(Stage)}.
  */
-public abstract class AnalysisConfiguration {
+public abstract class ConfigurationContext extends Configuration {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(ConfigurationContext.class);
 
 	private final Set<Stage> threadableStages = new HashSet<Stage>();
 
@@ -59,20 +64,10 @@ public abstract class AnalysisConfiguration {
 	 * @param stage
 	 *            A arbitrary stage, which will be added to the configuration and executed in a thread.
 	 */
+	@Override
 	protected final void addThreadableStage(final Stage stage) {
-		this.threadableStages.add(stage);
-	}
-
-	/**
-	 * Execute this method, to add a CompositeStage to the configuration, which should be executed in a own thread.
-	 *
-	 * @param stage
-	 *            A arbitrary CompositeStage, which will be added to the configuration and executed in a thread.
-	 */
-	protected final void addThreadableStage(final AbstractCompositeStage stage) {
-		this.threadableStages.add(stage.getFirstStage());
-		for (Stage threadableStage : stage.getThreadableStages()) {
-			this.addThreadableStage(threadableStage);
+		if (!this.threadableStages.add(stage)) {
+			LOGGER.warn("Stage " + stage.getId() + " was already marked as threadable stage.");
 		}
 	}
 
@@ -185,6 +180,7 @@ public abstract class AnalysisConfiguration {
 	 * @param <T>
 	 *            the type of elements to be sent
 	 */
+	@Override
 	protected final <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
 		connectPorts(sourcePort, targetPort, 4);
 	}
@@ -201,7 +197,15 @@ public abstract class AnalysisConfiguration {
 	 * @param <T>
 	 *            the type of elements to be sent
 	 */
+	@Override
 	protected final <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
+		if (sourcePort.getOwningStage().getInputPorts().length == 0 && !threadableStages.contains(sourcePort.getOwningStage())) {
+			addThreadableStage(sourcePort.getOwningStage());
+		}
+		if (sourcePort.getPipe() != null || targetPort.getPipe() != null) {
+			LOGGER.warn("Overwriting existing pipe while connecting stages " +
+					sourcePort.getOwningStage().getId() + " and " + targetPort.getOwningStage().getId() + ".");
+		}
 		new InstantiationPipe(sourcePort, targetPort, capacity);
 	}
 

src/main/java/teetime/framework/Analysis.java → src/main/java/teetime/framework/Execution.java

@@ -34,8 +34,8 @@ import teetime.framework.validation.AnalysisNotValidException;
 import teetime.util.Pair;
 
 /**
- * Represents an Analysis to which stages can be added and executed later.
- * This needs a {@link AnalysisConfiguration},
+ * Represents an Execution to which stages can be added and executed later.
+ * This needs a {@link ConfigurationContext},
  * in which the adding and configuring of stages takes place.
  * To start the analysis {@link #executeBlocking()} needs to be executed.
  * This class will automatically create threads and join them without any further commitment.
@@ -43,11 +43,11 @@ import teetime.util.Pair;
  * @author Christian Wulf, Nelson Tavares de Sousa
  *
  * @param <T>
- *            the type of the {@link AnalysisConfiguration}
+ *            the type of the {@link ConfigurationContext}
  */
-public final class Analysis<T extends AnalysisConfiguration> implements UncaughtExceptionHandler {
+public final class Execution<T extends ConfigurationContext> implements UncaughtExceptionHandler {
 
-	private static final Logger LOGGER = LoggerFactory.getLogger(Analysis.class);
+	private static final Logger LOGGER = LoggerFactory.getLogger(Execution.class);
 
 	private final T configuration;
 
@@ -64,32 +64,32 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 	private final List<RunnableProducerStage> producerRunnables = new LinkedList<RunnableProducerStage>();
 
 	/**
-	 * Creates a new {@link Analysis} that skips validating the port connections and uses the default listener.
+	 * Creates a new {@link Execution} that skips validating the port connections and uses the default listener.
 	 *
 	 * @param configuration
 	 *            to be used for the analysis
 	 */
-	public Analysis(final T configuration) {
+	public Execution(final T configuration) {
 		this(configuration, false, new IgnoringExceptionListenerFactory());
 	}
 
-	public Analysis(final T configuration, final boolean validationEnabled) {
+	public Execution(final T configuration, final boolean validationEnabled) {
 		this(configuration, validationEnabled, new IgnoringExceptionListenerFactory());
 	}
 
 	/**
-	 * Creates a new {@link Analysis} that skips validating the port connections and uses a specific listener.
+	 * Creates a new {@link Execution} that skips validating the port connections and uses a specific listener.
 	 *
 	 * @param configuration
 	 *            to be used for the analysis
 	 * @param factory
 	 *            specific listener for the exception handling
 	 */
-	public Analysis(final T configuration, final IExceptionListenerFactory factory) {
+	public Execution(final T configuration, final IExceptionListenerFactory factory) {
 		this(configuration, false, factory);
 	}
 
-	public Analysis(final T configuration, final boolean validationEnabled, final IExceptionListenerFactory factory) {
+	public Execution(final T configuration, final boolean validationEnabled, final IExceptionListenerFactory factory) {
 		this.configuration = configuration;
 		this.factory = factory;
 		if (validationEnabled) {
@@ -118,8 +118,8 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 	 *
 	 */
 	private final void init() {
-		AnalysisInstantiation analysisInstantiation = new AnalysisInstantiation(configuration);
-		analysisInstantiation.instantiatePipes();
+		ExecutionInstantiation executionInstantiation = new ExecutionInstantiation(configuration);
+		executionInstantiation.instantiatePipes();
 
 		final Set<Stage> threadableStageJobs = this.configuration.getThreadableStages();
 		if (threadableStageJobs.isEmpty()) {
@@ -194,7 +194,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 	/**
 	 * Calling this method will block the current thread, until the analysis terminates.
 	 *
-	 * @throws AnalysisException
+	 * @throws ExecutionException
 	 *             if at least one exception in one thread has occurred within the analysis. The exception contains the pairs of thread and throwable
 	 *
 	 * @since 1.1
@@ -209,7 +209,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 				thread.join();
 			}
 		} catch (InterruptedException e) {
-			LOGGER.error("Analysis has stopped unexpectedly", e);
+			LOGGER.error("Execution has stopped unexpectedly", e);
 			for (Thread thread : this.finiteProducerThreads) {
 				thread.interrupt();
 			}
@@ -224,7 +224,7 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 		}
 
 		if (!exceptions.isEmpty()) {
-			throw new AnalysisException(exceptions);
+			throw new ExecutionException(exceptions);
 		}
 	}
 
@@ -244,9 +244,9 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 	}
 
 	/**
-	 * This method will start the Analysis and block until it is finished.
+	 * This method will start the Execution and block until it is finished.
 	 *
-	 * @throws AnalysisException
+	 * @throws ExecutionException
 	 *             if at least one exception in one thread has occurred within the analysis. The exception contains the pairs of thread and throwable.
 	 *
 	 * @since 1.1
@@ -285,9 +285,9 @@ public final class Analysis<T extends AnalysisConfiguration> implements Uncaught
 	}
 
 	/**
-	 * Retrieves the Configuration which was used to add and arrange all stages needed for the Analysis
+	 * Retrieves the Configuration which was used to add and arrange all stages needed for the Execution
 	 *
-	 * @return the configuration used for the Analysis
+	 * @return the configuration used for the Execution
 	 */
 	public T getConfiguration() {
 		return this.configuration;

src/main/java/teetime/framework/AnalysisException.java → src/main/java/teetime/framework/ExecutionException.java

@@ -25,7 +25,7 @@ import teetime.util.Pair;
  *
  * @since 1.1
  */
-public class AnalysisException extends RuntimeException {
+public class ExecutionException extends RuntimeException {
 
 	/**
 	 *
@@ -34,7 +34,7 @@ public class AnalysisException extends RuntimeException {
 
 	private final Collection<Pair<Thread, Throwable>> exceptions;
 
-	public AnalysisException(final Collection<Pair<Thread, Throwable>> exceptions) {
+	public ExecutionException(final Collection<Pair<Thread, Throwable>> exceptions) {
 		super("Error(s) while running analysis. Check thrown exceptions.");
 		this.exceptions = exceptions;
 	}

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

+/**
+ * Copyright (C) 2015 Christian Wulf, Nelson Tavares de Sousa (http://teetime.sourceforge.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package teetime.framework;
 
 import java.util.HashMap;
@@ -13,22 +28,22 @@ import teetime.framework.pipe.SingleElementPipeFactory;
 import teetime.framework.pipe.SpScPipeFactory;
 import teetime.framework.pipe.UnboundedSpScPipeFactory;
 
-class AnalysisInstantiation {
+class ExecutionInstantiation {
 
-	private static final Logger LOGGER = LoggerFactory.getLogger(AnalysisInstantiation.class);
+	private static final Logger LOGGER = LoggerFactory.getLogger(ExecutionInstantiation.class);
 
 	private final IPipeFactory interBoundedThreadPipeFactory = new SpScPipeFactory();
 	private final IPipeFactory interUnboundedThreadPipeFactory = new UnboundedSpScPipeFactory();
 	private final IPipeFactory intraThreadPipeFactory = new SingleElementPipeFactory();
 
-	private final AnalysisConfiguration configuration;
+	private final ConfigurationContext configuration;
 
-	public AnalysisInstantiation(final AnalysisConfiguration configuration) {
+	public ExecutionInstantiation(final ConfigurationContext configuration) {
 		this.configuration = configuration;
 	}
 
 	@SuppressWarnings("rawtypes")
-	Integer colorAndConnectStages(final Integer i, final Map<Stage, Integer> colors, final Stage threadableStage, final AnalysisConfiguration configuration) {
+	Integer colorAndConnectStages(final Integer i, final Map<Stage, Integer> colors, final Stage threadableStage, final ConfigurationContext configuration) {
 		Integer createdConnections = new Integer(0);
 		Set<Stage> threadableStageJobs = configuration.getThreadableStages();
 		for (OutputPort outputPort : threadableStage.getOutputPorts()) {

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

@@ -21,95 +21,85 @@ import teetime.framework.signal.ISignal;
 
 public class InstantiationPipe implements IPipe {
 
-	private final InputPort<?> target;
+	private static final String ERROR_MESSAGE = "This must not be called while executing the configuration";
+
+	private final InputPort<?> targetPort;
 	private final int capacity;
 
 	public <T> InstantiationPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
-		this.target = targetPort;
+		this.targetPort = targetPort;
 		this.capacity = capacity;
 		sourcePort.setPipe(this);
+		targetPort.setPipe(this);
 	}
 
 	public int getCapacity() {
 		return capacity;
 	}
 
+	@Override
+	public InputPort<?> getTargetPort() {
+		return this.targetPort;
+	}
+
 	@Override
 	public boolean add(final Object element) {
-		// TODO Auto-generated method stub
-		return false;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public boolean addNonBlocking(final Object element) {
-		// TODO Auto-generated method stub
-		return false;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public boolean isEmpty() {
-		// TODO Auto-generated method stub
-		return false;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public int size() {
-		// TODO Auto-generated method stub
-		return 0;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public Object removeLast() {
-		// TODO Auto-generated method stub
-		return null;
-	}
-
-	@Override
-	public InputPort<?> getTargetPort() {
-		// TODO Auto-generated method stub
-		return this.target;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public void sendSignal(final ISignal signal) {
-		// TODO Auto-generated method stub
-
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public void reportNewElement() {
-		// TODO Auto-generated method stub
-
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public boolean isClosed() {
-		// TODO Auto-generated method stub
-		return false;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public boolean hasMore() {
-		// TODO Auto-generated method stub
-		return false;
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public void waitForStartSignal() throws InterruptedException {
-		// TODO Auto-generated method stub
-
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public void waitForInitializingSignal() throws InterruptedException {
-		// TODO Auto-generated method stub
-
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 	@Override
 	public void close() {
-		// TODO Auto-generated method stub
-
+		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
 }

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

@@ -19,9 +19,9 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
-import teetime.framework.Analysis;
-import teetime.framework.AnalysisConfiguration;
-import teetime.framework.AnalysisException;
+import teetime.framework.ConfigurationContext;
+import teetime.framework.Execution;
+import teetime.framework.ExecutionException;
 import teetime.framework.Stage;
 import teetime.framework.StageState;
 import teetime.stage.CollectorSink;
@@ -72,24 +72,23 @@ public final class StageTester {
 	/**
 	 * This method will start the test and block until it is finished.
 	 *
-	 * @throws AnalysisException
+	 * @throws ExecutionException
 	 *             if at least one exception in one thread has occurred within the analysis.
 	 *             The exception contains the pairs of thread and throwable.
 	 *
 	 */
 	public void start() {
-		final AnalysisConfiguration configuration = new Configuration(inputHolders, stage, outputHolders);
-		final Analysis<AnalysisConfiguration> analysis = new Analysis<AnalysisConfiguration>(configuration);
+		final ConfigurationContext configuration = new Configuration(inputHolders, stage, outputHolders);
+		final Execution<ConfigurationContext> analysis = new Execution<ConfigurationContext>(configuration);
 		analysis.executeBlocking();
 	}
 
-	private final class Configuration extends AnalysisConfiguration {
+	private final class Configuration extends ConfigurationContext {
 
 		public Configuration(final List<InputHolder<?>> inputHolders, final Stage stage, final List<OutputHolder<?>> outputHolders) {
 			for (InputHolder<?> inputHolder : inputHolders) {
 				final InitialElementProducer<Object> producer = new InitialElementProducer<Object>(inputHolder.getInput());
 				connectPorts(producer.getOutputPort(), inputHolder.getPort());
-				addThreadableStage(producer);
 			}
 
 			addThreadableStage(stage);

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

@@ -19,6 +19,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import teetime.framework.AbstractCompositeStage;
+import teetime.framework.ConfigurationContext;
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
 import teetime.framework.Stage;
@@ -31,7 +32,8 @@ public final class EveryXthPrinter<T> extends AbstractCompositeStage {
 	private final Distributor<T> distributor;
 	private final List<Stage> lastStages = new ArrayList<Stage>();
 
-	public EveryXthPrinter(final int threshold) {
+	public EveryXthPrinter(final int threshold, final ConfigurationContext context) {
+		super(context);
 		distributor = new Distributor<T>(new CopyByReferenceStrategy());
 		EveryXthStage<T> everyXthStage = new EveryXthStage<T>(threshold);
 		Printer<Integer> printer = new Printer<Integer>();
@@ -50,8 +52,7 @@ public final class EveryXthPrinter<T> extends AbstractCompositeStage {
 		return distributor.getNewOutputPort();
 	}
 
-	@Override
-	protected Stage getFirstStage() {
+	public Stage getFirstStage() {
 		return distributor;
 	}
 

src/main/java/teetime/stage/string/WordCounter.java

@@ -18,6 +18,7 @@ package teetime.stage.string;
 import java.util.ArrayList;
 
 import teetime.framework.AbstractCompositeStage;
+import teetime.framework.ConfigurationContext;
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
 import teetime.framework.Stage;
@@ -41,7 +42,8 @@ public final class WordCounter extends AbstractCompositeStage {
 	private final ArrayList<Stage> lastStages = new ArrayList<Stage>();
 
 	// The connection of the different stages is realized within the construction of a instance of this class.
-	public WordCounter() {
+	public WordCounter(final ConfigurationContext context) {
+		super(context);
 		this.lastStages.add(this.mapCounter);
 		final ToLowerCase toLowerCase = new ToLowerCase();
 
@@ -50,8 +52,7 @@ public final class WordCounter extends AbstractCompositeStage {
 		// connectStages(wordcharacterFilter.getOutputPort(), this.mapCounter.getInputPort());
 	}
 
-	@Override
-	protected Stage getFirstStage() {
+	public Stage getFirstStage() {
 		return this.tokenizer;
 	}
 

src/test/java/teetime/examples/cipher/CipherConfiguration.java

@@ -17,7 +17,7 @@ package teetime.examples.cipher;
 
 import java.io.File;
 
-import teetime.framework.AnalysisConfiguration;
+import teetime.framework.ConfigurationContext;
 import teetime.stage.CipherStage;
 import teetime.stage.CipherStage.CipherMode;
 import teetime.stage.InitialElementProducer;
@@ -26,7 +26,7 @@ import teetime.stage.ZipByteArray.ZipMode;
 import teetime.stage.io.ByteArrayFileWriter;
 import teetime.stage.io.File2ByteArray;
 
-public class CipherConfiguration extends AnalysisConfiguration {
+public class CipherConfiguration extends ConfigurationContext {
 
 	public CipherConfiguration(final String inputFile, final String outputFile, final String password) {
 		final File input = new File(inputFile);
@@ -47,8 +47,5 @@ public class CipherConfiguration extends AnalysisConfiguration {
 		connectPorts(decomp.getOutputPort(), decrypt.getInputPort());
 		connectPorts(decrypt.getOutputPort(), writer.getInputPort());
 
-		// this.getFiniteProducerStages().add(init);
-		this.addThreadableStage(init);
-
 	}
 }

src/test/java/teetime/examples/cipher/CipherTest.java

@@ -21,8 +21,8 @@ import java.io.IOException;
 import org.junit.Assert;
 import org.junit.Test;
 
-import teetime.framework.Analysis;
-import teetime.framework.AnalysisConfiguration;
+import teetime.framework.Execution;
+import teetime.framework.ConfigurationContext;
 
 import com.google.common.io.Files;
 
@@ -43,9 +43,9 @@ public class CipherTest {
 		final String outputFile = "src/test/resources/data/output.txt";
 		final String password = "Password";
 
-		final AnalysisConfiguration configuration = new CipherConfiguration(inputFile, outputFile, password);
-		final Analysis analysis = new Analysis(configuration);
-		analysis.executeBlocking();
+		final ConfigurationContext configuration = new CipherConfiguration(inputFile, outputFile, password);
+		final Execution execution = new Execution(configuration);
+		execution.executeBlocking();
 
 		Assert.assertTrue(Files.equal(new File(inputFile), new File(outputFile)));
 	}

src/test/java/teetime/examples/tokenizer/TokenizerConfiguration.java

@@ -17,7 +17,7 @@ package teetime.examples.tokenizer;
 
 import java.io.File;
 
-import teetime.framework.AnalysisConfiguration;
+import teetime.framework.ConfigurationContext;
 import teetime.stage.ByteArray2String;
 import teetime.stage.CipherStage;
 import teetime.stage.CipherStage.CipherMode;
@@ -28,7 +28,7 @@ import teetime.stage.ZipByteArray.ZipMode;
 import teetime.stage.io.File2ByteArray;
 import teetime.stage.string.Tokenizer;
 
-public class TokenizerConfiguration extends AnalysisConfiguration {
+public class TokenizerConfiguration extends ConfigurationContext {
 
 	private final Counter<String> counter;
 
@@ -50,7 +50,6 @@ public class TokenizerConfiguration extends AnalysisConfiguration {
 		connectPorts(b2s.getOutputPort(), tokenizer.getInputPort());
 		connectPorts(tokenizer.getOutputPort(), this.counter.getInputPort());
 
-		this.addThreadableStage(init);
 	}
 
 	public int getTokenCount() {

src/test/java/teetime/examples/tokenizer/TokenizerTest.java

@@ -22,7 +22,7 @@ import java.nio.charset.Charset;
 import org.junit.Assert;
 import org.junit.Test;
 
-import teetime.framework.Analysis;
+import teetime.framework.Execution;
 
 import com.google.common.io.Files;
 
@@ -42,8 +42,8 @@ public class TokenizerTest {
 		final String password = "Password";
 
 		final TokenizerConfiguration configuration = new TokenizerConfiguration(inputFile, password);
-		final Analysis analysis = new Analysis(configuration);
-		analysis.executeBlocking();
+		final Execution execution = new Execution(configuration);
+		execution.executeBlocking();
 
 		final String string = Files.toString(new File("src/test/resources/data/input.txt"), Charset.forName("UTF-8"));
 

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

@@ -33,51 +33,50 @@ import teetime.stage.InstanceOfFilter;
 import teetime.stage.basic.Sink;
 import teetime.util.StopWatch;
 
-public class AnalysisTest {
+public class ExecutionTest {
 
 	private static final long DELAY_IN_MS = 500;
 	private static final long ABSOLUTE_MAX_ERROR_IN_MS = 2;
 
-	private Analysis<TestConfig> analysis;
+	private Execution<TestConfig> execution;
 
 	@Before
 	public void before() {
 		TestConfig configuration = new TestConfig();
-		analysis = new Analysis<TestConfig>(configuration);
+		execution = new Execution<TestConfig>(configuration);
 	}
 
 	@Test
 	public void testExecuteNonBlocking() throws Exception {
 		StopWatch watch = new StopWatch();
 		watch.start();
-		analysis.executeNonBlocking();
+		execution.executeNonBlocking();
 		watch.end();
 
 		assertThat(watch.getDurationInMs(), is(lessThan(DELAY_IN_MS)));
-		assertFalse(analysis.getConfiguration().delay.finished);
+		assertFalse(execution.getConfiguration().delay.finished);
 
-		analysis.waitForTermination();
-		assertTrue(analysis.getConfiguration().delay.finished);
+		execution.waitForTermination();
+		assertTrue(execution.getConfiguration().delay.finished);
 	}
 
 	@Test
 	public void testExecuteBlocking() {
 		StopWatch watch = new StopWatch();
 		watch.start();
-		analysis.executeBlocking();
+		execution.executeBlocking();
 		watch.end();
 
 		assertThat(watch.getDurationInMs() + ABSOLUTE_MAX_ERROR_IN_MS, is(greaterThanOrEqualTo(DELAY_IN_MS)));
 	}
 
-	private static class TestConfig extends AnalysisConfiguration {
+	private static class TestConfig extends ConfigurationContext {
 		public final DelayAndTerminate delay;
 
 		public TestConfig() {
 			final InitialElementProducer<String> init = new InitialElementProducer<String>("Hello");
 			delay = new DelayAndTerminate(DELAY_IN_MS);
 			connectPorts(init.getOutputPort(), delay.getInputPort());
-			addThreadableStage(init);
 		}
 	}
 
@@ -105,20 +104,19 @@ public class AnalysisTest {
 
 	@Test
 	public void testInstantiatePipes() throws Exception {
-		Analysis<AnalysisTestConfig> interAnalysis = new Analysis<AnalysisTestConfig>(new AnalysisTestConfig(true));
+		Execution<AnalysisTestConfig> interAnalysis = new Execution<AnalysisTestConfig>(new AnalysisTestConfig(true));
 		assertThat(interAnalysis.getConfiguration().init.getOwningThread(), is(not(interAnalysis.getConfiguration().sink.getOwningThread())));
 
-		Analysis<AnalysisTestConfig> intraAnalysis = new Analysis<AnalysisTestConfig>(new AnalysisTestConfig(false));
+		Execution<AnalysisTestConfig> intraAnalysis = new Execution<AnalysisTestConfig>(new AnalysisTestConfig(false));
 		assertThat(intraAnalysis.getConfiguration().init.getOwningThread(), is(intraAnalysis.getConfiguration().sink.getOwningThread()));
 	}
 
-	private class AnalysisTestConfig extends AnalysisConfiguration {
+	private class AnalysisTestConfig extends ConfigurationContext {
 		public InitialElementProducer<Object> init = new InitialElementProducer<Object>();
 		public Sink<Object> sink = new Sink<Object>();
 
 		public AnalysisTestConfig(final boolean inter) {
 			connectPorts(init.getOutputPort(), sink.getInputPort());
-			addThreadableStage(init);
 			if (inter) {
 				addThreadableStage(sink);
 			}
@@ -133,10 +131,10 @@ public class AnalysisTest {
 		thrown.expect(IllegalStateException.class);
 		thrown.expectMessage("Crossing threads");
 		InvalidTestConfig configuration = new InvalidTestConfig();
-		new Analysis<InvalidTestConfig>(configuration);
+		new Execution<InvalidTestConfig>(configuration);
 	}
 
-	private class InvalidTestConfig extends AnalysisConfiguration {
+	private class InvalidTestConfig extends ConfigurationContext {
 		public InitialElementProducer<Object> init = new InitialElementProducer<Object>();
 		public InstanceOfFilter<Object, Object> iof = new InstanceOfFilter<Object, Object>(Object.class);
 		public Sink<Object> sink = new Sink<Object>();
@@ -145,9 +143,37 @@ public class AnalysisTest {
 			connectPorts(init.getOutputPort(), iof.getInputPort());
 			connectPorts(iof.getMatchedOutputPort(), sink.getInputPort());
 			connectPorts(init.createOutputPort(), sink.createInputPort());
-			addThreadableStage(init);
 			addThreadableStage(iof);
 		}
 	}
 
+	@Test
+	public void automaticallyAddHeadStages() {
+		AutomaticallyConfig context = new AutomaticallyConfig();
+		new Execution<ConfigurationContext>(context).executeBlocking();
+		assertTrue(context.executed);
+	}
+
+	private class AutomaticallyConfig extends ConfigurationContext {
+
+		public boolean executed;
+
+		public AutomaticallyConfig() {
+			AutomaticallyAddedStage aas = new AutomaticallyAddedStage();
+			Sink<Object> sink = new Sink<Object>();
+			connectPorts(aas.getOutputPort(), sink.getInputPort());
+		}
+
+		private class AutomaticallyAddedStage extends AbstractProducerStage<Object> {
+
+			@Override
+			protected void execute() {
+				executed = true;
+				terminate();
+			}
+
+		}
+
+	}
+
 }

src/test/java/teetime/framework/RunnableConsumerStageTest.java

@@ -34,11 +34,11 @@ public class RunnableConsumerStageTest {
 	public void testWaitingInfinitely() throws Exception {
 		RunnableConsumerStageTestConfiguration configuration = new RunnableConsumerStageTestConfiguration();
 
-		final Analysis analysis = new Analysis(configuration);
+		final Execution execution = new Execution(configuration);
 		final Thread thread = new Thread(new Runnable() {
 			@Override
 			public void run() {
-				start(analysis);
+				start(execution);
 			}
 		});
 		thread.start();
@@ -59,8 +59,8 @@ public class RunnableConsumerStageTest {
 	public void testCorrectStartAndTerminatation() throws Exception {
 		RunnableConsumerStageTestConfiguration configuration = new RunnableConsumerStageTestConfiguration(0, 1, 2, 3, 5);
 
-		final Analysis analysis = new Analysis(configuration);
-		start(analysis);
+		final Execution execution = new Execution(configuration);
+		start(execution);
 
 		assertEquals(5, configuration.getCollectedElements().size());
 	}
@@ -69,7 +69,7 @@ public class RunnableConsumerStageTest {
 	// public void testWaitingInfinitely() throws Exception {
 	// WaitStrategyConfiguration waitStrategyConfiguration = new WaitStrategyConfiguration(300, 42);
 	//
-	// final Analysis analysis = new Analysis(waitStrategyConfiguration);
+	// final Execution analysis = new Execution(waitStrategyConfiguration);
 	// Thread thread = new Thread(new Runnable() {
 	// @Override
 	// public void run() {
@@ -88,7 +88,7 @@ public class RunnableConsumerStageTest {
 	// public void testWaitingFinitely() throws Exception {
 	// WaitStrategyConfiguration waitStrategyConfiguration = new WaitStrategyConfiguration(300, 42);
 	//
-	// final Analysis analysis = new Analysis(waitStrategyConfiguration);
+	// final Execution analysis = new Execution(waitStrategyConfiguration);
 	// Thread thread = new Thread(new Runnable() {
 	// @Override
 	// public void run() {
@@ -109,19 +109,19 @@ public class RunnableConsumerStageTest {
 	public void testYieldRun() throws Exception {
 		YieldStrategyConfiguration waitStrategyConfiguration = new YieldStrategyConfiguration(42);
 
-		final Analysis analysis = new Analysis(waitStrategyConfiguration);
+		final Execution execution = new Execution(waitStrategyConfiguration);
 
-		start(analysis);
+		start(execution);
 
 		assertEquals(42, waitStrategyConfiguration.getCollectorSink().getElements().get(0));
 		assertEquals(1, waitStrategyConfiguration.getCollectorSink().getElements().size());
 	}
 
-	private void start(final Analysis analysis) {
+	private void start(final Execution execution) {
 		Collection<Pair<Thread, Throwable>> exceptions = new ArrayList<Pair<Thread, Throwable>>();
 		try {
-			analysis.executeBlocking();
-		} catch (AnalysisException e) {
+			execution.executeBlocking();
+		} catch (ExecutionException e) {
 			exceptions = e.getThrownExceptions();
 		}
 		for (Pair<Thread, Throwable> pair : exceptions) {

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

@@ -21,7 +21,7 @@ import java.util.List;
 import teetime.stage.CollectorSink;
 import teetime.stage.InitialElementProducer;
 
-public class RunnableConsumerStageTestConfiguration extends AnalysisConfiguration {
+public class RunnableConsumerStageTestConfiguration extends ConfigurationContext {
 
 	private final List<Integer> collectedElements = new ArrayList<Integer>();
 	private final CollectorSink<Integer> collectorSink;