added waitForInitializingSignal method

src/main/java/teetime/framework/AbstractInterThreadPipe.java

@@ -24,6 +24,7 @@ import org.jctools.queues.spec.Ordering;
 import org.jctools.queues.spec.Preference;
 
 import teetime.framework.signal.ISignal;
+import teetime.framework.signal.InitializingSignal;
 import teetime.util.concurrent.queue.PCBlockingQueue;
 import teetime.util.concurrent.queue.putstrategy.PutStrategy;
 import teetime.util.concurrent.queue.putstrategy.YieldPutStrategy;
@@ -69,6 +70,15 @@ public abstract class AbstractInterThreadPipe extends AbstractPipe {
 		cachedTargetStage.onSignal(signal, getTargetPort());
 	}
 
+	@Override
+	public final void waitForInitializingSignal() throws InterruptedException {
+		final ISignal signal = signalQueue.take();
+		if (!(signal instanceof InitializingSignal)) {
+			throw new IllegalStateException("Expected InitializingSignal, but was not the first arriving signal");
+		}
+		cachedTargetStage.onSignal(signal, getTargetPort());
+	}
+
 	@Override
 	public final boolean isClosed() {
 		return isClosed;

src/main/java/teetime/framework/AbstractIntraThreadPipe.java

@@ -56,4 +56,10 @@ public abstract class AbstractIntraThreadPipe extends AbstractPipe {
 	public void waitForStartSignal() throws InterruptedException {
 		// do nothing
 	}
+
+	@SuppressWarnings("PMD.EmptyMethodInAbstractClassShouldBeAbstract")
+	@Override
+	public void waitForInitializingSignal() throws InterruptedException {
+		// do nothing
+	}
 }

src/main/java/teetime/framework/InputPort.java

@@ -38,4 +38,8 @@ public final class InputPort<T> extends AbstractPort<T> {
 		pipe.waitForStartSignal();
 	}
 
+	public void waitForInitializingSignal() throws InterruptedException {
+		pipe.waitForInitializingSignal();
+	};
+
 }

src/main/java/teetime/framework/RunnableConsumerStage.java

@@ -43,14 +43,14 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
 	@SuppressWarnings("PMD.GuardLogStatement")
 	@Override
 	protected void beforeStageExecution(final Stage stage) throws InterruptedException {
-		logger.trace("Waiting for start signals..." + inputPorts);
+		logger.trace("Waiting for start signals... " + stage);
 		for (InputPort<?> inputPort : inputPorts) {
-			inputPort.waitForStartSignal();
+			inputPort.waitForInitializingSignal();
 		}
 		for (InputPort<?> inputPort : inputPorts) {
 			inputPort.waitForStartSignal();
 		}
-		logger.trace("Starting..." + stage);
+		logger.trace("Starting... " + stage);
 	}
 
 	@Override

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

@@ -86,6 +86,11 @@ public final class DummyPipe implements IPipe {
 
 	}
 
+	@Override
+	public void waitForInitializingSignal() throws InterruptedException {
+
+	}
+
 	@Override
 	public void close() {
 

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

@@ -92,6 +92,8 @@ public interface IPipe {
 
 	void waitForStartSignal() throws InterruptedException;
 
+	void waitForInitializingSignal() throws InterruptedException;
+
 	void close();
 
 }

src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java

@@ -102,6 +102,9 @@ class MergerTestingPipe implements IPipe {
 	@Override
 	public void waitForStartSignal() throws InterruptedException {}
 
+	@Override
+	public void waitForInitializingSignal() throws InterruptedException {}
+
 	@Override
 	public void close() {}