Merge remote-tracking branch 'origin/master' into DCParadigm

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

Merge remote-tracking branch 'origin/master' into DCParadigm
d8d09246 · Christian Wulf · 51187bff · 05df0801 · d8d09246 · d8d09246
Commit d8d09246 authored 10 years ago by Christian Wulf
--- a/pom.xml
+++ b/pom.xml
@@ -9,13 +9,13 @@
 	<name>TeeTime</name>
 	<description>TeeTime is a Pipe-and-Filter framework for Java</description>
-	<url>http://teetime.sourceforge.org</url>
+	<url>http://christianwulf.github.io/teetime/</url>
 	<inceptionYear>2015</inceptionYear>
 	<licenses>
 		<license>
 			<name>Apache License, Version 2.0</name>
-			<url>http://teetime.sourceforge.net/LICENSE.txt</url>
+			<url>http://christianwulf.github.io/teetime/LICENSE.txt</url>
 		</license>
 	</licenses>
@@ -310,12 +310,6 @@
 						<artifactId>doxia-module-markdown-teetime</artifactId>
 						<version>1.6</version>
 					</dependency>
-					<!-- add support for ssh/scp -->
-					<dependency>
-						<groupId>org.apache.maven.wagon</groupId>
-						<artifactId>wagon-ssh</artifactId>
-						<version>2.9-SNAPSHOT</version>
-					</dependency>
 					<!-- skin -->
 					<dependency>
 						<groupId>lt.velykis.maven.skins</groupId>

--- a/src/changes/changes.xml
+++ b/src/changes/changes.xml
@@ -6,7 +6,7 @@
 	</properties>
 	<body>
 		<release version="Snapshot" date="Daily basis"
-			description="Unstable preview of oncoming versions">
+			description="Unstable preview of oncoming versions - 2.0-SNAPSHOT">
 			<action dev="ntd" type="add" issue="33">
 				TeeTime automatically
 				chooses the correct type of pipe for all connections.
@@ -21,19 +21,24 @@
 			</action>
 			<action dev="ntd" type="add" issue="171">
 				Configurations are now
-				built within an AnalysisContext which is passed on to nested
+				built within the Configuration class 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="update">
+				Renamed Analysis to Execution
+			</action>
 			<action dev="ntd" type="remove">
-				Marked Pair class as deprecated.
+				Removed pair class.
 			</action>
 			<action dev="ntd" type="add" issue="154">
 				All stages will be
 				initialized before starting the analysis.
 			</action>
+			<action dev="ntd" type="add" issue="122">
+				Threads can be named for better debugging.
+			</action>
 		</release>
 		<release version="1.1.2" date="12.05.2015" description="Minor bugfixes for 1.1">

--- a/src/main/java/teetime/framework/AbstractCompositeStage.java
+++ b/src/main/java/teetime/framework/AbstractCompositeStage.java
@@ -49,6 +49,18 @@ public abstract class AbstractCompositeStage {
 		return dynamicContext;
 	}
+	/**
+	 * Execute this method, to add a stage to the configuration, which should be executed in a own thread.
+	 *
+	 * @param stage
+	 *            A arbitrary stage, which will be added to the configuration and executed in a thread.
+	 * @param threadName
+	 *            A string which can be used for debugging.
+	 */
+	protected final void addThreadableStage(final Stage stage, final String threadName) {
+		context.addThreadableStage(stage, threadName);
+	}
 	/**
 	 * Execute this method, to add a stage to the configuration, which should be executed in a own thread.
 	 *
@@ -56,7 +68,7 @@ public abstract class AbstractCompositeStage {
 	 *            A arbitrary stage, which will be added to the configuration and executed in a thread.
 	 */
 	protected final void addThreadableStage(final Stage stage) {
-		context.addThreadableStage(stage);
+		this.addThreadableStage(stage, stage.getId());
 	}
 	/**

--- a/src/main/java/teetime/framework/AbstractInterThreadPipe.java
+++ b/src/main/java/teetime/framework/AbstractInterThreadPipe.java
@@ -25,11 +25,12 @@ import org.jctools.queues.spec.Preference;
 import teetime.framework.signal.ISignal;
 import teetime.framework.signal.InitializingSignal;
-import teetime.util.concurrent.queue.PCBlockingQueue;
+import teetime.framework.signal.StartingSignal;
-import teetime.util.concurrent.queue.putstrategy.PutStrategy;
+import teetime.util.framework.concurrent.queue.PCBlockingQueue;
-import teetime.util.concurrent.queue.putstrategy.YieldPutStrategy;
+import teetime.util.framework.concurrent.queue.putstrategy.PutStrategy;
-import teetime.util.concurrent.queue.takestrategy.SCParkTakeStrategy;
+import teetime.util.framework.concurrent.queue.putstrategy.YieldPutStrategy;
-import teetime.util.concurrent.queue.takestrategy.TakeStrategy;
+import teetime.util.framework.concurrent.queue.takestrategy.SCParkTakeStrategy;
+import teetime.util.framework.concurrent.queue.takestrategy.TakeStrategy;
 public abstract class AbstractInterThreadPipe extends AbstractPipe {
@@ -65,16 +66,19 @@ public abstract class AbstractInterThreadPipe extends AbstractPipe {
 	}
 	@Override
-	public final void waitForStartSignal() throws InterruptedException {
+	public final void waitForInitializingSignal() throws InterruptedException {
 		final ISignal signal = signalQueue.take();
+		if (!(signal instanceof InitializingSignal)) {
+			throw new IllegalStateException("Expected InitializingSignal, but was " + signal.getClass().getSimpleName());
+		}
 		cachedTargetStage.onSignal(signal, getTargetPort());
 	}
 	@Override
-	public final void waitForInitializingSignal() throws InterruptedException {
+	public final void waitForStartSignal() throws InterruptedException {
 		final ISignal signal = signalQueue.take();
-		if (!(signal instanceof InitializingSignal)) {
+		if (!(signal instanceof StartingSignal)) {
-			throw new IllegalStateException("Expected InitializingSignal, but was not the first arriving signal");
+			throw new IllegalStateException("Expected StartingSignal, but was " + signal.getClass().getSimpleName());
 		}
 		cachedTargetStage.onSignal(signal, getTargetPort());
 	}

--- a/src/main/java/teetime/framework/AbstractPort.java
+++ b/src/main/java/teetime/framework/AbstractPort.java
@@ -19,47 +19,43 @@ import teetime.framework.pipe.IPipe;
 public abstract class AbstractPort<T> {
-	private final String portName;
-	protected IPipe pipe;
 	/**
 	 * The type of this port.
 	 * <p>
 	 * <i>Used to validate the connection between two ports at runtime.</i>
 	 * </p>
 	 */
-	protected final Class<T> type;
+	private final Class<T> type;
 	private final Stage owningStage;
+	private final String name;
-	public AbstractPort(final Class<T> type, final Stage owningStage, final String portName) {
+	protected IPipe pipe;
+	protected AbstractPort(final Class<T> type, final Stage owningStage, final String name) {
 		super();
-		this.portName = portName;
 		this.type = type;
 		this.owningStage = owningStage;
+		this.name = name;
 	}
-	public IPipe getPipe() {
+	public Class<T> getType() {
-		return this.pipe;
+		return this.type;
 	}
-	public void setPipe(final IPipe pipe) {
+	public Stage getOwningStage() {
-		this.pipe = pipe;
+		return owningStage;
 	}
-	public Class<T> getType() {
+	public IPipe getPipe() {
-		return this.type;
+		return this.pipe;
 	}
-	public final Stage getOwningStage() {
+	public void setPipe(final IPipe pipe) {
-		return owningStage;
+		this.pipe = pipe;
 	}
 	@Override
-	public final String toString() {
+	public String toString() {
-		if (portName == null) {
+		return (name != null) ? name : super.toString();
-			return super.toString();
-		} else {
-			return portName;
-		}
 	}
 }
--- a/src/main/java/teetime/framework/AbstractStage.java
+++ b/src/main/java/teetime/framework/AbstractStage.java
@@ -15,6 +15,7 @@
 */
 package teetime.framework;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -35,6 +36,9 @@ public abstract class AbstractStage extends Stage {
 	private OutputPort<?>[] outputPorts = new OutputPort<?>[0];
 	private StageState currentState = StageState.CREATED;
+	private final Set<OutputPortRemovedListener> outputPortRemovedListeners = new HashSet<OutputPortRemovedListener>();
+	private final Set<InputPortRemovedListener> inputPortsRemovedListeners = new HashSet<InputPortRemovedListener>();
 	@Override
 	public InputPort<?>[] getInputPorts() {
 		return inputPorts;
@@ -134,14 +138,11 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be received
 	 *
-	 * @return Newly added InputPort
+	 * @return the newly added InputPort
 	 *
 	 */
-	// * @deprecated Since 1.1. Use {@link #createInputPort(Class)} instead.
-	@SuppressWarnings("unchecked")
-	// @Deprecated
 	protected <T> InputPort<T> createInputPort() {
-		return (InputPort<T>) createInputPort(null, null);
+		return createInputPort(null, null);
 	}
 	/**
@@ -153,7 +154,7 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be received
 	 *
-	 * @return Newly added InputPort
+	 * @return the newly added InputPort
 	 */
 	protected <T> InputPort<T> createInputPort(final Class<T> type) {
 		return createInputPort(type, null);
@@ -167,14 +168,11 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be received
 	 *
-	 * @return Newly added InputPort
+	 * @return the newly added InputPort
 	 *
 	 */
-	// * @deprecated Since 1.1. Use {@link #createInputPort(Class)} instead.
-	@SuppressWarnings("unchecked")
-	// @Deprecated
 	protected <T> InputPort<T> createInputPort(final String name) {
-		return (InputPort<T>) createInputPort(null, name);
+		return createInputPort(null, name);
 	}
 	/**
@@ -187,7 +185,7 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be received
 	 *
-	 * @return Newly added InputPort
+	 * @return the newly added InputPort
 	 */
 	protected <T> InputPort<T> createInputPort(final Class<T> type, final String name) {
 		final InputPort<T> inputPort = new InputPort<T>(type, this, name);
@@ -201,14 +199,11 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be sent
 	 *
-	 * @return Newly added OutputPort
+	 * @return the newly added OutputPort
 	 *
 	 */
-	// * @deprecated Since 1.1. Use {@link #createOutputPort(Class)} instead.
-	@SuppressWarnings("unchecked")
-	// @Deprecated
 	protected <T> OutputPort<T> createOutputPort() {
-		return (OutputPort<T>) createOutputPort(null, null);
+		return createOutputPort(null, null);
 	}
 	/**
@@ -220,12 +215,10 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be sent
 	 *
-	 * @return Newly added OutputPort
+	 * @return the newly added OutputPort
 	 */
 	protected <T> OutputPort<T> createOutputPort(final Class<T> type) {
-		final OutputPort<T> outputPort = new OutputPort<T>(type, this, null);
+		return createOutputPort(type, null);
-		outputPorts = addElementToArray(outputPort, outputPorts);
-		return outputPort;
 	}
 	/**
@@ -237,14 +230,11 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be sent
 	 *
-	 * @return Newly added OutputPort
+	 * @return the newly added OutputPort
 	 *
 	 */
-	// * @deprecated Since 1.1. Use {@link #createOutputPort(Class)} instead.
-	@SuppressWarnings("unchecked")
-	// @Deprecated
 	protected <T> OutputPort<T> createOutputPort(final String name) {
-		return (OutputPort<T>) createOutputPort(null, name);
+		return createOutputPort(null, name);
 	}
 	/**
@@ -258,7 +248,7 @@ public abstract class AbstractStage extends Stage {
 	 * @param <T>
 	 *            the type of elements to be sent
 	 *
-	 * @return Newly added OutputPort
+	 * @return the newly added OutputPort
 	 */
 	protected <T> OutputPort<T> createOutputPort(final Class<T> type, final String name) {
 		final OutputPort<T> outputPort = new OutputPort<T>(type, this, name);
@@ -266,7 +256,7 @@ public abstract class AbstractStage extends Stage {
 		return outputPort;
 	}
-	private <T> T[] addElementToArray(final T element, final T[] srcArray) {
+	private <T, E extends T> T[] addElementToArray(final E element, final T[] srcArray) {
 		T[] newOutputPorts = Arrays.copyOf(srcArray, srcArray.length + 1);
 		newOutputPorts[srcArray.length] = element;
 		return newOutputPorts;
@@ -302,4 +292,68 @@ public abstract class AbstractStage extends Stage {
 		return TerminationStrategy.BY_SIGNAL;
 	}
+	protected <T> DynamicOutputPort<T> createDynamicOutputPort() {
+		final DynamicOutputPort<T> outputPort = new DynamicOutputPort<T>(null, this, outputPorts.length);
+		outputPorts = addElementToArray(outputPort, outputPorts);
+		return outputPort;
+	}
+	protected <T> DynamicInputPort<T> createDynamicInputPort() {
+		final DynamicInputPort<T> inputPort = new DynamicInputPort<T>(null, this, inputPorts.length);
+		inputPorts = addElementToArray(inputPort, inputPorts);
+		return inputPort;
+	}
+	@Override
+	protected void removeDynamicPort(final DynamicOutputPort<?> dynamicOutputPort) {
+		int index = dynamicOutputPort.getIndex();
+		List<OutputPort<?>> tempOutputPorts = new ArrayList<OutputPort<?>>(Arrays.asList(outputPorts));
+		OutputPort<?> removedOutputPort = tempOutputPorts.remove(index);
+		for (int i = index; i < tempOutputPorts.size(); i++) {
+			OutputPort<?> outputPort = tempOutputPorts.get(i);
+			if (outputPort instanceof DynamicOutputPort) {
+				((DynamicOutputPort<?>) outputPort).setIndex(i);
+			}
+		}
+		outputPorts = tempOutputPorts.toArray(new OutputPort[0]);
+		firePortRemoved(removedOutputPort);
+	}
+	private void firePortRemoved(final OutputPort<?> removedOutputPort) {
+		for (OutputPortRemovedListener listener : outputPortRemovedListeners) {
+			listener.onOutputPortRemoved(this, removedOutputPort);
+		}
+	}
+	protected final void addOutputPortRemovedListener(final OutputPortRemovedListener outputPortRemovedListener) {
+		outputPortRemovedListeners.add(outputPortRemovedListener);
+	}
+	@Override
+	protected void removeDynamicPort(final DynamicInputPort<?> dynamicInputPort) {
+		int index = dynamicInputPort.getIndex();
+		List<InputPort<?>> tempInputPorts = new ArrayList<InputPort<?>>(Arrays.asList(inputPorts));
+		InputPort<?> removedInputPort = tempInputPorts.remove(index);
+		for (int i = index; i < tempInputPorts.size(); i++) {
+			InputPort<?> inputPort = tempInputPorts.get(i);
+			if (inputPort instanceof DynamicInputPort) {
+				((DynamicInputPort<?>) inputPort).setIndex(i);
+			}
+		}
+		inputPorts = tempInputPorts.toArray(new InputPort[0]);
+		firePortRemoved(removedInputPort);
+	}
+	private void firePortRemoved(final InputPort<?> removedInputPort) {
+		for (InputPortRemovedListener listener : inputPortsRemovedListeners) {
+			listener.onInputPortRemoved(this, removedInputPort);
+		}
+	}
+	protected final void addInputPortRemovedListener(final InputPortRemovedListener outputPortRemovedListener) {
+		inputPortsRemovedListeners.add(outputPortRemovedListener);
+	}
 }
--- a/src/main/java/teetime/framework/Configuration.java
+++ b/src/main/java/teetime/framework/Configuration.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;
 /**

--- a/src/main/java/teetime/framework/ConfigurationContext.java
+++ b/src/main/java/teetime/framework/ConfigurationContext.java
@@ -15,8 +15,8 @@
 */
 package teetime.framework;
-import java.util.HashSet;
+import java.util.HashMap;
-import java.util.Set;
+import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -33,19 +33,19 @@ public final class ConfigurationContext {
 	private static final Logger LOGGER = LoggerFactory.getLogger(ConfigurationContext.class);
-	private final Set<Stage> threadableStages = new HashSet<Stage>();
+	private final Map<Stage, String> threadableStages = new HashMap<Stage, String>();
 	ConfigurationContext() {}
-	Set<Stage> getThreadableStages() {
+	Map<Stage, String> getThreadableStages() {
 		return this.threadableStages;
 	}
 	/**
 	 * @see AbstractCompositeStage#addThreadableStage(Stage)
 	 */
-	final void addThreadableStage(final Stage stage) {
+	final void addThreadableStage(final Stage stage, final String threadName) {
-		if (!this.threadableStages.add(stage)) {
+		if (this.threadableStages.put(stage, threadName) != null) {
 			LOGGER.warn("Stage " + stage.getId() + " was already marked as threadable stage.");
 		}
 	}
@@ -54,8 +54,10 @@ public final class ConfigurationContext {
 	 * @see AbstractCompositeStage#connectPorts(OutputPort, InputPort, int)
 	 */
 	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())) {
+		if (sourcePort.getOwningStage().getInputPorts().length == 0) {
-			addThreadableStage(sourcePort.getOwningStage());
+			if (!threadableStages.containsKey(sourcePort.getOwningStage())) {
+				addThreadableStage(sourcePort.getOwningStage(), sourcePort.getOwningStage().getId());
+			}
 		}
 		if (sourcePort.getPipe() != null || targetPort.getPipe() != null) {
 			LOGGER.warn("Overwriting existing pipe while connecting stages " +

--- a/src/main/java/teetime/framework/DynamicActuator.java
+++ b/src/main/java/teetime/framework/DynamicActuator.java
+package teetime.framework;
+public class DynamicActuator {
+	/**
+	 * @deprecated Use {@link #startWithinNewThread(Stage)} instead.
+	 */
+	@Deprecated
+	public AbstractRunnableStage wrap(final Stage stage) {
+		if (stage.getInputPorts().length > 0) {
+			return new RunnableConsumerStage(stage);
+		}
+		return new RunnableProducerStage(stage);
+	}
+	public Runnable startWithinNewThread(final Stage stage) {
+		Runnable runnable = wrap(stage);
+		Thread thread = new Thread(runnable);
+		thread.start();
+		return runnable;
+	}
+}
--- a/src/main/java/teetime/framework/DynamicInputPort.java
+++ b/src/main/java/teetime/framework/DynamicInputPort.java
+package teetime.framework;
+/**
+ *
+ * @author Christian Wulf
+ *
+ * @param <T>
+ *            the type of elements to be received
+ *
+ * @since 1.2
+ */
+public final class DynamicInputPort<T> extends InputPort<T> {
+	private int index;
+	DynamicInputPort(final Class<T> type, final Stage owningStage, final int index) {
+		super(type, owningStage, null);
+		this.index = index;
+	}
+	public int getIndex() {
+		return index;
+	}
+	public void setIndex(final int index) {
+		this.index = index;
+	}
+}
--- a/src/main/java/teetime/framework/DynamicOutputPort.java
+++ b/src/main/java/teetime/framework/DynamicOutputPort.java
+package teetime.framework;
+/**
+ *
+ * @author Christian Wulf
+ *
+ * @param <T>
+ *            the type of elements to be sent
+ *
+ * @since 1.2
+ */
+public final class DynamicOutputPort<T> extends OutputPort<T> {
+	private int index;
+	DynamicOutputPort(final Class<T> type, final Stage owningStage, final int index) {
+		super(type, owningStage, null);
+		this.index = index;
+	}
+	public int getIndex() {
+		return index;
+	}
+	public void setIndex(final int index) {
+		this.index = index;
+	}
+}
--- a/src/main/java/teetime/framework/Execution.java
+++ b/src/main/java/teetime/framework/Execution.java
@@ -19,6 +19,7 @@ import java.lang.Thread.UncaughtExceptionHandler;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -27,7 +28,7 @@ import org.slf4j.LoggerFactory;
 import teetime.framework.exceptionHandling.AbstractExceptionListener;
 import teetime.framework.exceptionHandling.IExceptionListenerFactory;
-import teetime.framework.exceptionHandling.IgnoringExceptionListenerFactory;
+import teetime.framework.exceptionHandling.TerminatingExceptionListenerFactory;
 import teetime.framework.signal.InitializingSignal;
 import teetime.framework.signal.ValidatingSignal;
 import teetime.framework.validation.AnalysisNotValidException;
@@ -72,11 +73,11 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 	 *            to be used for the analysis
 	 */
 	public Execution(final T configuration) {
-		this(configuration, false, new IgnoringExceptionListenerFactory());
+		this(configuration, false, new TerminatingExceptionListenerFactory());
 	}
 	public Execution(final T configuration, final boolean validationEnabled) {
-		this(configuration, validationEnabled, new IgnoringExceptionListenerFactory());
+		this(configuration, validationEnabled, new TerminatingExceptionListenerFactory());
 	}
 	/**
@@ -102,8 +103,8 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 	// BETTER validate concurrently
 	private void validateStages() {
-		final Set<Stage> threadableStageJobs = this.configuration.getContext().getThreadableStages();
+		final Map<Stage, String> threadableStageJobs = this.configuration.getContext().getThreadableStages();
-		for (Stage stage : threadableStageJobs) {
+		for (Stage stage : threadableStageJobs.keySet()) {
 			// // portConnectionValidator.validate(stage);
 			// }
@@ -123,7 +124,7 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 		ExecutionInstantiation executionInstantiation = new ExecutionInstantiation(configuration.getContext());
 		executionInstantiation.instantiatePipes();
-		final Set<Stage> threadableStageJobs = this.configuration.getContext().getThreadableStages();
+		final Set<Stage> threadableStageJobs = this.configuration.getContext().getThreadableStages().keySet();
 		if (threadableStageJobs.isEmpty()) {
 			throw new IllegalStateException("No stage was added using the addThreadableStage(..) method. Add at least one stage.");
 		}
@@ -182,7 +183,7 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 	private Thread createThread(final AbstractRunnableStage runnable, final String name) {
 		final Thread thread = new Thread(runnable);
 		thread.setUncaughtExceptionHandler(this);
-		thread.setName(name);
+		thread.setName(configuration.getContext().getThreadableStages().get(runnable.stage));
 		return thread;
 	}
@@ -203,10 +204,12 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 	 */
 	public void waitForTermination() {
 		try {
+			LOGGER.debug("Waiting for finiteProducerThreads");
 			for (Thread thread : this.finiteProducerThreads) {
 				thread.join();
 			}
+			LOGGER.debug("Waiting for consumerThreads");
 			for (Thread thread : this.consumerThreads) {
 				thread.join();
 			}
@@ -221,6 +224,7 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 			}
 		}
+		LOGGER.debug("Interrupting infiniteProducerThreads...");
 		for (Thread thread : this.infiniteProducerThreads) {
 			thread.interrupt();
 		}
@@ -300,7 +304,7 @@ public final class Execution<T extends Configuration> implements UncaughtExcepti
 		if (!executionInterrupted) {
 			executionInterrupted = true;
 			LOGGER.warn("Thread " + thread + " was interrupted. Terminating analysis now.");
-			for (Stage stage : configuration.getContext().getThreadableStages()) {
+			for (Stage stage : configuration.getContext().getThreadableStages().keySet()) {
 				if (stage.getOwningThread() != thread) {
 					if (stage.getTerminationStrategy() == TerminationStrategy.BY_SELF_DECISION) {
 						stage.terminate();

--- a/src/main/java/teetime/framework/ExecutionInstantiation.java
+++ b/src/main/java/teetime/framework/ExecutionInstantiation.java
@@ -45,7 +45,7 @@ class ExecutionInstantiation {
 	@SuppressWarnings("rawtypes")
 	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();
+		Set<Stage> threadableStageJobs = configuration.getThreadableStages().keySet();
 		for (OutputPort outputPort : threadableStage.getOutputPorts()) {
 			if (outputPort.pipe != null) {
 				if (outputPort.pipe instanceof InstantiationPipe) {
@@ -82,7 +82,7 @@ class ExecutionInstantiation {
 	void instantiatePipes() {
 		Integer i = new Integer(0);
 		Map<Stage, Integer> colors = new HashMap<Stage, Integer>();
-		Set<Stage> threadableStageJobs = configuration.getThreadableStages();
+		Set<Stage> threadableStageJobs = configuration.getThreadableStages().keySet();
 		Integer createdConnections = 0;
 		for (Stage threadableStage : threadableStageJobs) {
 			i++;

--- a/src/main/java/teetime/framework/InputPort.java
+++ b/src/main/java/teetime/framework/InputPort.java
@@ -24,7 +24,7 @@ package teetime.framework;
 *
 * @since 1.0
 */
-public final class InputPort<T> extends AbstractPort<T> {
+public class InputPort<T> extends AbstractPort<T> {
 	InputPort(final Class<T> type, final Stage owningStage, final String portName) {
 		super(type, owningStage, portName);

--- a/src/main/java/teetime/framework/InputPortRemovedListener.java
+++ b/src/main/java/teetime/framework/InputPortRemovedListener.java
+package teetime.framework;
+public interface InputPortRemovedListener {
+	void onInputPortRemoved(Stage stage, InputPort<?> removedInputPort);
+}
--- a/src/main/java/teetime/framework/OutputPort.java
+++ b/src/main/java/teetime/framework/OutputPort.java
@@ -27,7 +27,7 @@ import teetime.framework.signal.TerminatingSignal;
 *
 * @since 1.0
 */
-public final class OutputPort<T> extends AbstractPort<T> {
+public class OutputPort<T> extends AbstractPort<T> {
 	OutputPort(final Class<T> type, final Stage owningStage, final String portName) {
 		super(type, owningStage, portName);

--- a/src/main/java/teetime/framework/OutputPortRemovedListener.java
+++ b/src/main/java/teetime/framework/OutputPortRemovedListener.java
+package teetime.framework;
+public interface OutputPortRemovedListener {
+	void onOutputPortRemoved(Stage stage, OutputPort<?> removedOutputPort);
+}
--- a/src/main/java/teetime/framework/RunnableConsumerStage.java
+++ b/src/main/java/teetime/framework/RunnableConsumerStage.java
@@ -15,39 +15,30 @@
 */
 package teetime.framework;
-import teetime.framework.idle.IdleStrategy;
-import teetime.framework.idle.YieldStrategy;
 import teetime.framework.signal.ISignal;
 import teetime.framework.signal.TerminatingSignal;
 final class RunnableConsumerStage extends AbstractRunnableStage {
-	// cache the input ports here since getInputPorts() always returns a new copy
-	private final InputPort<?>[] inputPorts;
 	/**
-	 * Creates a new instance with the {@link YieldStrategy} as default idle strategy.
+	 * Creates a new instance.
 	 *
 	 * @param stage
 	 *            to execute within an own thread
 	 */
 	public RunnableConsumerStage(final Stage stage) {
-		this(stage, new YieldStrategy());
-	}
-	public RunnableConsumerStage(final Stage stage, final IdleStrategy idleStrategy) {
 		super(stage);
-		this.inputPorts = stage.getInputPorts(); // FIXME should getInputPorts() really be defined in Stage?
 	}
 	@SuppressWarnings("PMD.GuardLogStatement")
 	@Override
 	protected void beforeStageExecution() throws InterruptedException {
-		logger.trace("Waiting for start signals... " + stage);
+		logger.trace("Waiting for init signals... " + stage);
-		for (InputPort<?> inputPort : inputPorts) {
+		for (InputPort<?> inputPort : stage.getInputPorts()) {
 			inputPort.waitForInitializingSignal();
 		}
-		for (InputPort<?> inputPort : inputPorts) {
+		logger.trace("Waiting for start signals... " + stage);
+		for (InputPort<?> inputPort : stage.getInputPorts()) {
 			inputPort.waitForStartSignal();
 		}
 		logger.trace("Starting... " + stage);
@@ -63,7 +54,10 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
 	}
 	private void checkForTerminationSignal(final Stage stage) {
-		for (InputPort<?> inputPort : inputPorts) {
+		System.out.println("checkForTerminationSignal: " + stage);
+		// FIXME should getInputPorts() really be defined in Stage?
+		for (InputPort<?> inputPort : stage.getInputPorts()) {
+			System.out.println("\tclosed: " + inputPort.isClosed() + " (" + inputPort);
 			if (!inputPort.isClosed()) {
 				return;
 			}
@@ -75,7 +69,7 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
 	@Override
 	protected void afterStageExecution() {
 		final ISignal signal = new TerminatingSignal();
-		for (InputPort<?> inputPort : inputPorts) {
+		for (InputPort<?> inputPort : stage.getInputPorts()) {
 			stage.onSignal(signal, inputPort);
 		}
 	}

--- a/src/main/java/teetime/framework/RunnableProducerStage.java
+++ b/src/main/java/teetime/framework/RunnableProducerStage.java
@@ -21,12 +21,12 @@ import teetime.framework.signal.InitializingSignal;
 import teetime.framework.signal.StartingSignal;
 import teetime.framework.signal.TerminatingSignal;
-final class RunnableProducerStage extends AbstractRunnableStage {
+public final class RunnableProducerStage extends AbstractRunnableStage {
 	private final Semaphore startSemaphore = new Semaphore(0);
 	private final Semaphore initSemaphore = new Semaphore(0);
-	public RunnableProducerStage(final Stage stage) {
+	RunnableProducerStage(final Stage stage) {
 		super(stage);
 	}
@@ -57,11 +57,13 @@ final class RunnableProducerStage extends AbstractRunnableStage {
 		startSemaphore.release();
 	}
-	public void waitForInitializingSignal() throws InterruptedException {
+	private void waitForInitializingSignal() throws InterruptedException {
+		logger.trace("waitForInitializingSignal");
 		initSemaphore.acquire();
 	}
-	public void waitForStartingSignal() throws InterruptedException {
+	private void waitForStartingSignal() throws InterruptedException {
+		logger.trace("waitForStartingSignal");
 		startSemaphore.acquire();
 	}
 }
--- a/src/main/java/teetime/framework/Stage.java
+++ b/src/main/java/teetime/framework/Stage.java
@@ -148,4 +148,8 @@ public abstract class Stage {
 		this.exceptionHandler = exceptionHandler;
 	}
+	protected abstract void removeDynamicPort(DynamicOutputPort<?> dynamicOutputPort);
+	protected abstract void removeDynamicPort(DynamicInputPort<?> dynamicInputPort);
 }