#192 working prototype

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

@@ -32,11 +32,8 @@ public abstract class AbstractCompositeStage {
 
 	private final ConfigurationContext context;
 
-	public AbstractCompositeStage(final ConfigurationContext context) {
-		if (null == context) {
-			throw new IllegalArgumentException("Context may not be null.");
-		}
-		this.context = context;
+	public AbstractCompositeStage() {
+		this.context = new ConfigurationContext();
 	}
 
 	protected ConfigurationContext getContext() {

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

@@ -25,8 +25,4 @@ package teetime.framework;
  */
 public abstract class Configuration extends AbstractCompositeStage {
 
-	public Configuration() {
-		super(new ConfigurationContext());
-	}
-
 }

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

@@ -29,7 +29,7 @@ import teetime.framework.pipe.InstantiationPipe;
  *
  * @since 2.0
  */
-public final class ConfigurationContext {
+public class ConfigurationContext {
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(ConfigurationContext.class);
 
@@ -63,7 +63,21 @@ public final class ConfigurationContext {
 			LOGGER.warn("Overwriting existing pipe while connecting stages " +
 					sourcePort.getOwningStage().getId() + " and " + targetPort.getOwningStage().getId() + ".");
 		}
+		mergeContexts(sourcePort.getOwningStage(), targetPort.getOwningStage());
 		new InstantiationPipe(sourcePort, targetPort, capacity);
 	}
 
+	final void mergeContexts(final Stage sourceStage, final Stage targetStage) {
+		if (!sourceStage.owningContext.equals(EmptyContext.getInstance())) {
+			this.threadableStages.putAll(sourceStage.owningContext.threadableStages);
+		} else {
+			sourceStage.owningContext = this;
+		}
+		if (!targetStage.owningContext.equals(EmptyContext.getInstance())) {
+			this.threadableStages.putAll(targetStage.owningContext.threadableStages);
+		} else {
+			targetStage.owningContext = this;
+		}
+	}
+
 }

src/main/java/teetime/framework/EmptyContext.java

+package teetime.framework;
+
+class EmptyContext extends ConfigurationContext {
+
+	private static final EmptyContext INSTANCE = new EmptyContext();
+
+	private EmptyContext() {
+
+	}
+
+	static EmptyContext getInstance() {
+		return EmptyContext.INSTANCE;
+	}
+
+}

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

@@ -48,6 +48,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. */
 	protected Thread owningThread;
 
+	protected ConfigurationContext owningContext = EmptyContext.getInstance();
+
 	protected Stage() {
 		this.id = this.createId();
 		this.logger = LoggerFactory.getLogger(this.getClass().getCanonicalName() + ":" + id);

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

@@ -33,7 +33,6 @@ public final class EveryXthPrinter<T> extends AbstractCompositeStage {
 	private final List<Stage> lastStages = new ArrayList<Stage>();
 
 	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>();

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

@@ -37,7 +37,6 @@ public final class WordCounter extends AbstractCompositeStage {
 	private final MappingCounter<String> mapCounter;
 
 	public WordCounter(final ConfigurationContext context) {
-		super(context);
 
 		this.tokenizer = new Tokenizer(" ");
 		final ToLowerCase toLowerCase = new ToLowerCase();

wiki @ 709c839c

+Subproject commit 709c839c447a50c93b37fcc633a01297115d4823