diff --git a/src/main/java/teetime/stage/basic/merger/Merger.java b/src/main/java/teetime/stage/basic/merger/Merger.java index 6afda057e455d38760c2073822730459ee939879..0634134c27045328a0abfdad894dd828c26310c8 100644 --- a/src/main/java/teetime/stage/basic/merger/Merger.java +++ b/src/main/java/teetime/stage/basic/merger/Merger.java @@ -54,19 +54,17 @@ public class Merger<T> extends AbstractStage { public void onSignal(final ISignal signal, final InputPort<?> inputPort) { this.logger.trace("Got signal: " + signal + " from input port: " + inputPort); - signal.trigger(this); + if (0 == finishedInputPorts) { + signal.trigger(this); + } + this.finishedInputPorts++; if (this.finishedInputPorts == this.getInputPorts().length) { this.outputPort.sendSignal(signal); + this.finishedInputPorts = 0; } } - @Override - public void onTerminating() throws Exception { - this.finishedInputPorts++; - super.onTerminating(); - } - public IMergerStrategy<T> getStrategy() { return this.strategy; }