added test;

connectPorts only adds a producer to threadableStages, if it has not
been added yet

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

@@ -55,7 +55,9 @@ public final class ConfigurationContext {
 	 */
 	final <T> void connectPorts(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
 		if (sourcePort.getOwningStage().getInputPorts().length == 0) {
-			addThreadableStage(sourcePort.getOwningStage(), sourcePort.getOwningStage().getId());
+			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 " +

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

@@ -176,4 +176,23 @@ public class ExecutionTest {
 
 	}
 
+	@Test
+	public void threadNameing() {
+		NameConfig configuration = new NameConfig();
+		Execution<NameConfig> execution = new Execution<NameConfig>(configuration);
+		assertThat(configuration.stageWithNamedThread.getOwningThread().getName(), is("TestName"));
+	}
+
+	private class NameConfig extends Configuration {
+
+		public InitialElementProducer<Object> stageWithNamedThread;
+
+		public NameConfig() {
+			stageWithNamedThread = new InitialElementProducer<Object>(new Object());
+			addThreadableStage(stageWithNamedThread, "TestName");
+			connectPorts(stageWithNamedThread.getOutputPort(), new Sink().getInputPort());
+		}
+
+	}
+
 }