diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java
index 2fe100e57402455723436b3bf04fa832b42809f3..ad33e815be84b88cf588b3c5cf44f4ae16060726 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/AbstractStage.java
@@ -1,7 +1,6 @@
package teetime.variant.methodcallWithPorts.framework.core;
import java.util.ArrayList;
-import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
@@ -10,6 +9,8 @@ import org.slf4j.LoggerFactory;
import teetime.variant.methodcallWithPorts.framework.core.pipe.DummyPipe;
import teetime.variant.methodcallWithPorts.framework.core.pipe.IPipe;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.validation.InvalidPortConnection;
public abstract class AbstractStage implements StageWithPort {
@@ -56,14 +57,6 @@ public abstract class AbstractStage implements StageWithPort {
return true;
}
- @Override
- public void onStart() {
- this.cachedInputPorts = this.inputPortList.toArray(new InputPort<?>[0]);
- this.cachedOutputPorts = this.outputPortList.toArray(new OutputPort<?>[0]);
-
- this.connectUnconnectedOutputPorts();
- }
-
@SuppressWarnings("unchecked")
private void connectUnconnectedOutputPorts() {
for (OutputPort<?> outputPort : this.cachedOutputPorts) {
@@ -74,11 +67,6 @@ public abstract class AbstractStage implements StageWithPort {
}
}
- protected void onFinished() {
- // empty default implementation
- this.onIsPipelineHead();
- }
-
protected InputPort<?>[] getInputPorts() {
return this.cachedInputPorts;
}
@@ -116,22 +104,31 @@ public abstract class AbstractStage implements StageWithPort {
*/
@Override
public void onSignal(final Signal signal, final InputPort<?> inputPort) {
- this.logger.debug("Got signal: " + signal + " from input port: " + inputPort);
-
- switch (signal) {
- case FINISHED:
- this.onFinished();
- break;
- default:
- this.logger.warn("Aborted sending signal " + signal + ". Reason: Unknown signal.");
- break;
- }
+ this.logger.trace("Got signal: " + signal + " from input port: " + inputPort);
+
+ signal.trigger(this);
for (OutputPort<?> outputPort : this.outputPortList) {
outputPort.sendSignal(signal);
}
}
+ public void onValidating(final List<InvalidPortConnection> invalidPortConnections) {
+ this.validateOutputPorts(invalidPortConnections);
+ }
+
+ public void onStarting() {
+ this.cachedInputPorts = this.inputPortList.toArray(new InputPort<?>[0]);
+ this.cachedOutputPorts = this.outputPortList.toArray(new OutputPort<?>[0]);
+
+ this.connectUnconnectedOutputPorts();
+ }
+
+ public void onTerminating() {
+ // empty default implementation
+ this.onIsPipelineHead();
+ }
+
protected <T> InputPort<T> createInputPort() {
InputPort<T> inputPort = new InputPort<T>(this);
// inputPort.setType(type); // TODO set type for input port
@@ -146,9 +143,8 @@ public abstract class AbstractStage implements StageWithPort {
return outputPort;
}
- public List<InvalidPortConnection> validateOutputPorts() {
- List<InvalidPortConnection> invalidOutputPortMessages = new LinkedList<InvalidPortConnection>();
-
+ @Override
+ public void validateOutputPorts(final List<InvalidPortConnection> invalidPortConnections) {
for (OutputPort<?> outputPort : this.getOutputPorts()) {
IPipe<?> pipe = outputPort.getPipe();
if (null != pipe) { // if output port is connected with another one
@@ -156,12 +152,10 @@ public abstract class AbstractStage implements StageWithPort {
Class<?> targetPortType = pipe.getTargetPort().getType();
if (null == sourcePortType || !sourcePortType.equals(targetPortType)) {
InvalidPortConnection invalidPortConnection = new InvalidPortConnection(outputPort, pipe.getTargetPort());
- invalidOutputPortMessages.add(invalidPortConnection);
+ invalidPortConnections.add(invalidPortConnection);
}
}
}
-
- return invalidOutputPortMessages;
}
@Override
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java
index b29c12464e1d329de06a92826b2a338f74968ce4..97d66d0521a617757332a4aee9fbf894584ffe36 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/OutputPort.java
@@ -1,5 +1,7 @@
package teetime.variant.methodcallWithPorts.framework.core;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
+
public class OutputPort<T> extends AbstractPort<T> {
/**
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java
index f20f891e025f205d6a46e8f93654088d98ae6058..018f39e01b56fe302a927dab29cd9e2f89d3356f 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Pipeline.java
@@ -5,13 +5,17 @@ import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.StartingSignal;
+import teetime.variant.methodcallWithPorts.framework.core.validation.InvalidPortConnection;
+
import kieker.common.logging.Log;
import kieker.common.logging.LogFactory;
/**
- *
+ *
* @author Christian Wulf
- *
+ *
* @param <FirstStage>
* @param <LastStage>
*/
@@ -28,13 +32,7 @@ public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageW
private final List<StageWithPort> intermediateStages = new LinkedList<StageWithPort>();
private LastStage lastStage;
- // BETTER remove the stage array and use the output ports instead for passing a signal to all stages in the same thread; what about multiple same signals due to
- // multiple input ports?
- private StageWithPort[] stages;
private StageWithPort parentStage;
- // private int startIndex;
-
- private int firstStageIndex;
// private final Set<StageWithPort<?, ?>> currentHeads = new HashSet<StageWithPort<?, ?>>();
@@ -70,7 +68,7 @@ public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageW
@Override
public void executeWithPorts() {
- StageWithPort headStage = this.stages[this.firstStageIndex];
+ StageWithPort headStage = this.firstStage;
// do {
headStage.executeWithPorts();
@@ -106,16 +104,16 @@ public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageW
// do nothing
}
- @Override
- public void onStart() {
+ @Deprecated
+ public void onStarting() {
int size = 1 + this.intermediateStages.size() + 1;
- this.stages = new StageWithPort[size];
- this.stages[0] = this.firstStage;
+ StageWithPort[] stages = new StageWithPort[size];
+ stages[0] = this.firstStage;
for (int i = 0; i < this.intermediateStages.size(); i++) {
StageWithPort stage = this.intermediateStages.get(i);
- this.stages[1 + i] = stage;
+ stages[1 + i] = stage;
}
- this.stages[this.stages.length - 1] = this.lastStage;
+ stages[stages.length - 1] = this.lastStage;
// for (int i = 0; i < this.stages.length; i++) {
// StageWithPort<?, ?> stage = this.stages[i];
@@ -129,8 +127,8 @@ public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageW
// }
// this.stages[this.stages.length - 1].setSuccessor(new EndStage<Object>());
- for (StageWithPort stage : this.stages) {
- stage.onStart();
+ for (StageWithPort stage : stages) {
+ stage.onSignal(new StartingSignal(), null);
}
}
@@ -146,14 +144,9 @@ public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageW
@Override
public boolean isReschedulable() {
- // return this.reschedulable;
return this.firstStage.isReschedulable();
}
- // public void setReschedulable(final boolean reschedulable) {
- // this.reschedulable = reschedulable;
- // }
-
@Override
public void onSignal(final Signal signal, final InputPort<?> inputPort) {
this.firstStage.onSignal(signal, inputPort);
@@ -167,4 +160,9 @@ public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageW
return this.lastStage;
}
+ @Override
+ public void validateOutputPorts(final List<InvalidPortConnection> invalidPortConnections) {
+ // do nothing
+ }
+
}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java
index f2ff6c73ddb016674710b52845b3b9b3712d40b9..54bc6088cbfde995d658db7d914cbfec629be2ff 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/RunnableStage.java
@@ -3,10 +3,15 @@ package teetime.variant.methodcallWithPorts.framework.core;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import teetime.variant.methodcallWithPorts.framework.core.signal.StartingSignal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.TerminatingSignal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.ValidatingSignal;
+
public class RunnableStage implements Runnable {
private final StageWithPort stage;
private final Logger logger;
+ private boolean validationEnabled;
public RunnableStage(final StageWithPort stage) {
this.stage = stage;
@@ -17,14 +22,25 @@ public class RunnableStage implements Runnable {
public void run() {
this.logger.debug("Executing runnable stage...");
+ if (this.validationEnabled) {
+ ValidatingSignal validatingSignal = new ValidatingSignal();
+ this.stage.onSignal(validatingSignal, null);
+ if (validatingSignal.getInvalidPortConnections().size() > 0) {
+ // throw new RuntimeException(message);
+ // TODO implement what to do on validation messages
+ }
+ }
+
try {
- this.stage.onStart();
+ StartingSignal startingSignal = new StartingSignal();
+ this.stage.onSignal(startingSignal, null);
do {
this.stage.executeWithPorts();
} while (this.stage.isReschedulable());
- this.stage.onSignal(Signal.FINISHED, null);
+ TerminatingSignal terminatingSignal = new TerminatingSignal();
+ this.stage.onSignal(terminatingSignal, null);
} catch (RuntimeException e) {
this.logger.error("Terminating thread due to the following exception: ", e);
@@ -33,4 +49,12 @@ public class RunnableStage implements Runnable {
this.logger.debug("Finished runnable stage. (" + this.stage.getId() + ")");
}
+
+ public boolean isValidationEnabled() {
+ return this.validationEnabled;
+ }
+
+ public void setValidationEnabled(final boolean validationEnabled) {
+ this.validationEnabled = validationEnabled;
+ }
}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Signal.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Signal.java
deleted file mode 100644
index 0e3eadce9f1065d4ed113c94ce300dea25c35eb0..0000000000000000000000000000000000000000
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/Signal.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package teetime.variant.methodcallWithPorts.framework.core;
-
-public enum Signal {
- FINISHED
-
-}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java
index 72563660a89d85f88d2fe8ab026bae77cd32c501..15733038a7566d2a5eb6cbb7ade4213af023aba7 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/StageWithPort.java
@@ -1,5 +1,10 @@
package teetime.variant.methodcallWithPorts.framework.core;
+import java.util.List;
+
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.validation.InvalidPortConnection;
+
public interface StageWithPort {
String getId();
@@ -20,7 +25,12 @@ public interface StageWithPort {
void onIsPipelineHead();
- void onStart();
-
void onSignal(Signal signal, InputPort<?> inputPort);
+
+ /**
+ *
+ * @param invalidPortConnections
+ * <i>(Passed as parameter for performance reasons)</i>
+ */
+ void validateOutputPorts(List<InvalidPortConnection> invalidPortConnections);
}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/DummyPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/DummyPipe.java
index ee90ec5d5b1b9eef370b4ca9eaab27362c9034d6..498ada08f792229067e685bcd9f9adf27cadf522 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/DummyPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/DummyPipe.java
@@ -2,7 +2,7 @@ package teetime.variant.methodcallWithPorts.framework.core.pipe;
import teetime.variant.methodcallWithPorts.framework.core.InputPort;
import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
/**
* A pipe implementation used to connect unconnected output ports.
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java
index e7d500b577a59362ee18f0704c032361d2ffe474..0bedb314592d62abcd5b8a2917cca130444cb5a2 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IPipe.java
@@ -2,7 +2,7 @@ package teetime.variant.methodcallWithPorts.framework.core.pipe;
import teetime.variant.methodcallWithPorts.framework.core.InputPort;
import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
public interface IPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IntraThreadPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IntraThreadPipe.java
index 59af28e52ad6164f781e0915196ad3db6d74ac08..116d2903a0b699c7ebcbfad158f5787cd2eddd75 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IntraThreadPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/IntraThreadPipe.java
@@ -1,6 +1,6 @@
package teetime.variant.methodcallWithPorts.framework.core.pipe;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
public abstract class IntraThreadPipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java
index ef95689501c1e1608ebd95444d40ea2efd3410de..8ceb292719e062cb6253b6d001d3f08ac2cbde93 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/pipe/SpScPipe.java
@@ -10,7 +10,7 @@ import org.jctools.queues.spec.Preference;
import teetime.variant.methodcallWithPorts.framework.core.InputPort;
import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
public class SpScPipe<T> extends AbstractPipe<T> {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/Signal.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/Signal.java
new file mode 100644
index 0000000000000000000000000000000000000000..93a47871075b0db76d559873e11f3f596cff5f13
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/Signal.java
@@ -0,0 +1,8 @@
+package teetime.variant.methodcallWithPorts.framework.core.signal;
+
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+
+public interface Signal {
+
+ void trigger(AbstractStage stage);
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/StartingSignal.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/StartingSignal.java
new file mode 100644
index 0000000000000000000000000000000000000000..9e82408c3ac1a802964c1cb67952025181f99b70
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/StartingSignal.java
@@ -0,0 +1,12 @@
+package teetime.variant.methodcallWithPorts.framework.core.signal;
+
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+
+public class StartingSignal implements Signal {
+
+ @Override
+ public void trigger(final AbstractStage stage) {
+ stage.onStarting();
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/TerminatingSignal.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/TerminatingSignal.java
new file mode 100644
index 0000000000000000000000000000000000000000..b0ea58c0a35fad8b54b52e68a3315724366bb356
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/TerminatingSignal.java
@@ -0,0 +1,12 @@
+package teetime.variant.methodcallWithPorts.framework.core.signal;
+
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+
+public class TerminatingSignal implements Signal {
+
+ @Override
+ public void trigger(final AbstractStage stage) {
+ stage.onTerminating();
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/ValidatingSignal.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/ValidatingSignal.java
new file mode 100644
index 0000000000000000000000000000000000000000..fad95a43646e4f6151a97de3b5d922fd2381cab7
--- /dev/null
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/signal/ValidatingSignal.java
@@ -0,0 +1,22 @@
+package teetime.variant.methodcallWithPorts.framework.core.signal;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
+import teetime.variant.methodcallWithPorts.framework.core.validation.InvalidPortConnection;
+
+public class ValidatingSignal implements Signal {
+
+ private final List<InvalidPortConnection> invalidPortConnections = new LinkedList<InvalidPortConnection>();
+
+ @Override
+ public void trigger(final AbstractStage stage) {
+ stage.onValidating(this.invalidPortConnections);
+ }
+
+ public List<InvalidPortConnection> getInvalidPortConnections() {
+ return invalidPortConnections;
+ }
+
+}
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/InvalidPortConnection.java b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/validation/InvalidPortConnection.java
similarity index 66%
rename from src/main/java/teetime/variant/methodcallWithPorts/framework/core/InvalidPortConnection.java
rename to src/main/java/teetime/variant/methodcallWithPorts/framework/core/validation/InvalidPortConnection.java
index d753230de7d44c703f58745720ef69165d126f2d..51239fedc5d763caceacdb89d34185c7a4a3dfe8 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/framework/core/InvalidPortConnection.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/framework/core/validation/InvalidPortConnection.java
@@ -1,4 +1,7 @@
-package teetime.variant.methodcallWithPorts.framework.core;
+package teetime.variant.methodcallWithPorts.framework.core.validation;
+
+import teetime.variant.methodcallWithPorts.framework.core.InputPort;
+import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
public class InvalidPortConnection {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementDelayMeasuringStage.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementDelayMeasuringStage.java
index 72213b5db89716943e76fc2b2ecf64fbf5715bef..2ab1fa47f75153d79de9db1a68db054d32d8b654 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementDelayMeasuringStage.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementDelayMeasuringStage.java
@@ -29,9 +29,9 @@ public class ElementDelayMeasuringStage<T> extends ConsumerStage<T> {
}
@Override
- public void onStart() {
+ public void onStarting() {
this.resetTimestamp(System.nanoTime());
- super.onStart();
+ super.onStarting();
}
private void computeElementDelay(final Long timestampInNs) {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java
index a73ec83c1d365380ffff46734d95632f590f162f..33ad7d5fb4fc544da48cefead2f003084879fcdc 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/ElementThroughputMeasuringStage.java
@@ -30,9 +30,9 @@ public class ElementThroughputMeasuringStage<T> extends ConsumerStage<T> {
}
@Override
- public void onStart() {
+ public void onStarting() {
this.resetTimestamp(System.nanoTime());
- super.onStart();
+ super.onStarting();
}
private void computeElementThroughput(final Long timestampInNs) {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java
index 574f822a2c71916d82ace1dce3ef0a6611ef4810..d0e281fdceb2469876c8857da0f77392c434c9a1 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/Relay.java
@@ -2,8 +2,8 @@ package teetime.variant.methodcallWithPorts.stage;
import teetime.variant.methodcallWithPorts.framework.core.InputPort;
import teetime.variant.methodcallWithPorts.framework.core.ProducerStage;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
import teetime.variant.methodcallWithPorts.framework.core.pipe.SpScPipe;
+import teetime.variant.methodcallWithPorts.framework.core.signal.TerminatingSignal;
public class Relay<T> extends ProducerStage<T> {
@@ -15,8 +15,7 @@ public class Relay<T> extends ProducerStage<T> {
public void execute() {
T element = this.inputPort.receive();
if (null == element) {
- // if (this.getInputPort().getPipe().isClosed()) {
- if (this.cachedCastedInputPipe.getSignal() == Signal.FINISHED) {
+ if (this.cachedCastedInputPipe.getSignal() instanceof TerminatingSignal) {
this.setReschedulable(false);
assert 0 == this.inputPort.getPipe().size();
}
@@ -27,16 +26,9 @@ public class Relay<T> extends ProducerStage<T> {
}
@Override
- public void onStart() {
+ public void onStarting() {
this.cachedCastedInputPipe = (SpScPipe<T>) this.inputPort.getPipe();
- super.onStart();
- }
-
- @Override
- public void onIsPipelineHead() {
- // if (this.getInputPort().getPipe().isClosed()) {
- // this.setReschedulable(false);
- // }
+ super.onStarting();
}
public InputPort<T> getInputPort() {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/basic/merger/Merger.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/basic/merger/Merger.java
index c28d0bcfca564986282b0c0898818a10e7498829..7f469d4f856faa5b953ee9539e3ee35a9150a372 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/basic/merger/Merger.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/basic/merger/Merger.java
@@ -19,16 +19,16 @@ package teetime.variant.methodcallWithPorts.stage.basic.merger;
import teetime.variant.methodcallWithPorts.framework.core.AbstractStage;
import teetime.variant.methodcallWithPorts.framework.core.InputPort;
import teetime.variant.methodcallWithPorts.framework.core.OutputPort;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
+import teetime.variant.methodcallWithPorts.framework.core.signal.Signal;
/**
- *
+ *
* This stage merges data from the input ports, by taking elements according to the chosen merge strategy and by putting them to the output port.
- *
+ *
* @author Christian Wulf
- *
+ *
* @since 1.10
- *
+ *
* @param <T>
* the type of the input ports and the output port
*/
@@ -61,16 +61,9 @@ public class Merger<T> extends AbstractStage {
@Override
public void onSignal(final Signal signal, final InputPort<?> inputPort) {
- this.logger.debug("Got signal: " + signal + " from input port: " + inputPort);
-
- switch (signal) {
- case FINISHED:
- this.onFinished();
- break;
- default:
- this.logger.warn("Aborted sending signal " + signal + ". Reason: Unknown signal.");
- break;
- }
+ this.logger.trace("Got signal: " + signal + " from input port: " + inputPort);
+
+ signal.trigger(this);
if (this.finishedInputPorts == this.getInputPorts().length) {
this.outputPort.sendSignal(signal);
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/io/DbReader.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/io/DbReader.java
index d8746827b4a62f42c51a8d4d2c52497f95dc9ea4..497daeabeeab9b589d95e9c1a7e5edb0960a5d77 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/io/DbReader.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/io/DbReader.java
@@ -49,7 +49,7 @@ public class DbReader extends ProducerStage<IMonitoringRecord> {
private volatile boolean running = true;
@Override
- public void onStart() {
+ public void onStarting() {
try {
Class.forName(this.driverClassname).newInstance();
} catch (final Exception ex) { // NOPMD NOCS (IllegalCatchCheck)
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/io/Printer.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/io/Printer.java
index 8cf9a9881f52aa7c2ca35394f3379d63d1fbda3a..3bed4e2bcb487cd7fc3a76667a11bb290e92c503 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/io/Printer.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/io/Printer.java
@@ -25,7 +25,7 @@ import teetime.variant.methodcallWithPorts.framework.core.ConsumerStage;
/**
* @author Matthias Rohr, Jan Waller, Nils Christian Ehmke
- *
+ *
* @since 1.10
*/
@Description("A filter to print objects to a configured stream")
@@ -86,14 +86,15 @@ public class Printer<T> extends ConsumerStage<T> {
}
@Override
- public void onStart() {
+ public void onStarting() {
+ super.onStarting();
this.initializeStream();
}
@Override
- protected void onFinished() {
+ public void onTerminating() {
this.closeStream();
- super.onFinished();
+ super.onTerminating();
}
private void initializeStream() {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/io/TCPReader.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/io/TCPReader.java
index 6dca74493295e6697ec3b6deeca1816548924ba7..decfc7f31cd057d5e60d96f95754e83706e5f5c5 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/io/TCPReader.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/io/TCPReader.java
@@ -76,13 +76,13 @@ public class TCPReader extends ProducerStage<IMonitoringRecord> {
}
@Override
- public void onStart() {
+ public void onStarting() {
this.recordFactory = new RecordFactory();
this.register();
this.tcpStringReader = new TCPStringReader(this.port2, this.stringRegistry);
this.tcpStringReader.start();
- super.onStart();
+ super.onStarting();
}
private void register() {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/TCPReaderSink.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/TCPReaderSink.java
index a09cdea37f89e0385b847a137f7b0a2c208d9fc0..af0c71fad9d84bf4ea78c4ea8093c30cd83432ad 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/TCPReaderSink.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/TCPReaderSink.java
@@ -78,7 +78,7 @@ public class TCPReaderSink extends ProducerStage<IMonitoringRecord> {
}
@Override
- public void onStart() {
+ public void onStarting() {
this.executorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
@@ -88,7 +88,7 @@ public class TCPReaderSink extends ProducerStage<IMonitoringRecord> {
this.tcpStringReader = new TCPStringReader(this.port2, this.stringRegistry);
this.tcpStringReader.start();
- super.onStart();
+ super.onStarting();
}
@Override
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/fileToRecord/BinaryFile2RecordFilter.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/fileToRecord/BinaryFile2RecordFilter.java
index 9119e31b564204a60259745c85de07e0999f9b16..6e9769c6d621a5d76216ea54bd3adcd0e5137517 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/fileToRecord/BinaryFile2RecordFilter.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/kieker/fileToRecord/BinaryFile2RecordFilter.java
@@ -58,9 +58,9 @@ public class BinaryFile2RecordFilter extends ConsumerStage<File> {
}
@Override
- public void onStart() {
+ public void onStarting() {
this.recordFromBinaryFileCreator = new RecordFromBinaryFileCreator(this.logger, this.classNameRegistryRepository);
- super.onStart();
+ super.onStarting();
}
public ClassNameRegistryRepository getClassNameRegistryRepository() {
diff --git a/src/main/java/teetime/variant/methodcallWithPorts/stage/stringBuffer/StringBufferFilter.java b/src/main/java/teetime/variant/methodcallWithPorts/stage/stringBuffer/StringBufferFilter.java
index d7c6f3c41af839de8156af0aa8cbc83738fb31e1..a566f7a5d3401cb706ec6752862e585d81f599c2 100644
--- a/src/main/java/teetime/variant/methodcallWithPorts/stage/stringBuffer/StringBufferFilter.java
+++ b/src/main/java/teetime/variant/methodcallWithPorts/stage/stringBuffer/StringBufferFilter.java
@@ -44,12 +44,12 @@ public class StringBufferFilter<T> extends ConsumerStage<T> {
}
@Override
- public void onStart() {
+ public void onStarting() {
for (final AbstractDataTypeHandler<?> handler : this.dataTypeHandlers) {
handler.setLogger(this.logger);
handler.setStringRepository(this.kiekerHashMap);
}
- super.onStart();
+ super.onStarting();
}
private T handle(final T object) {
diff --git a/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment17/MethodCallThroughputAnalysis17.java b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment17/MethodCallThroughputAnalysis17.java
index 652ccbdc2aa43ac9157612fd409cfa389ac1bc31..9a4c84ae7838a29932e653b9ad3f234d33cdb8d8 100644
--- a/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment17/MethodCallThroughputAnalysis17.java
+++ b/src/test/java/teetime/variant/methodcallWithPorts/examples/experiment17/MethodCallThroughputAnalysis17.java
@@ -24,11 +24,11 @@ import teetime.variant.explicitScheduling.examples.throughput.TimestampObject;
import teetime.variant.explicitScheduling.framework.core.Analysis;
import teetime.variant.methodcallWithPorts.framework.core.Pipeline;
import teetime.variant.methodcallWithPorts.framework.core.RunnableStage;
-import teetime.variant.methodcallWithPorts.framework.core.Signal;
import teetime.variant.methodcallWithPorts.framework.core.StageWithPort;
import teetime.variant.methodcallWithPorts.framework.core.pipe.IPipe;
import teetime.variant.methodcallWithPorts.framework.core.pipe.SpScPipe;
import teetime.variant.methodcallWithPorts.framework.core.pipe.UnorderedGrowablePipe;
+import teetime.variant.methodcallWithPorts.framework.core.signal.TerminatingSignal;
import teetime.variant.methodcallWithPorts.stage.CollectorSink;
import teetime.variant.methodcallWithPorts.stage.NoopFilter;
import teetime.variant.methodcallWithPorts.stage.ObjectProducer;
@@ -40,7 +40,7 @@ import teetime.variant.methodcallWithPorts.stage.basic.distributor.Distributor;
/**
* @author Christian Wulf
- *
+ *
* @since 1.10
*/
public class MethodCallThroughputAnalysis17 extends Analysis {
@@ -160,7 +160,7 @@ public class MethodCallThroughputAnalysis17 extends Analysis {
startPipe.add(this.inputObjectCreator.create());
}
// startPipe.close();
- startPipe.setSignal(Signal.FINISHED);
+ startPipe.setSignal(new TerminatingSignal());
UnorderedGrowablePipe.connect(relay.getOutputPort(), startTimestampFilter.getInputPort());
diff --git a/submodules/JCTools b/submodules/JCTools
index 75998aa20b7ec897ec321c1f94192de888f2dc6e..88e1e25f9519b250258c7e5ada30935975ab2d10 160000
--- a/submodules/JCTools
+++ b/submodules/JCTools
@@ -1 +1 @@
-Subproject commit 75998aa20b7ec897ec321c1f94192de888f2dc6e
+Subproject commit 88e1e25f9519b250258c7e5ada30935975ab2d10