diff --git a/.classpath b/.classpath
index 9022062d8dfa64335dc564f900678538dac21bd5..31223dc742592f137f39dc384bc12c47bdaecc00 100644
--- a/.classpath
+++ b/.classpath
@@ -23,8 +23,16 @@
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
- <classpathentry including="**/*.java" kind="src" path="src/main/resources"/>
- <classpathentry including="**/*.java" kind="src" path="src/test/resources"/>
+ <classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
+ <attributes>
+ <attribute name="maven.pomderived" value="true"/>
+ </attributes>
+ </classpathentry>
+ <classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources">
+ <attributes>
+ <attribute name="maven.pomderived" value="true"/>
+ </attributes>
+ </classpathentry>
<classpathentry kind="src" path="conf"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>
diff --git a/.settings/edu.umd.cs.findbugs.core.prefs b/.settings/edu.umd.cs.findbugs.core.prefs
index af7d163d7df5db16611a9d3fa014f685b53e43d2..aa459302fcd46c3d10841046d2ebbb6fdf30d3d7 100644
--- a/.settings/edu.umd.cs.findbugs.core.prefs
+++ b/.settings/edu.umd.cs.findbugs.core.prefs
@@ -1,5 +1,5 @@
#FindBugs User Preferences
-#Wed Jun 24 10:24:19 CEST 2015
+#Thu Jul 30 15:19:22 CEST 2015
detector_threshold=3
effort=max
excludefilter0=.fbExcludeFilterFile|true
diff --git a/src/main/java/teetime/framework/Pipeline.java b/src/main/java/teetime/framework/Pipeline.java
index f818703d477586c8feda385281aa0c24b1e4953b..efe021216f497e95b7f14a66cde655dea4df029c 100644
--- a/src/main/java/teetime/framework/Pipeline.java
+++ b/src/main/java/teetime/framework/Pipeline.java
@@ -26,17 +26,17 @@ import java.util.List;
* the type of the last stage in this pipeline
*/
// TODO Consider to move it in the framework
-public final class Pipeline<L extends Stage> extends AbstractCompositeStage {
+public final class Pipeline<L extends AbstractStage> extends AbstractCompositeStage {
- private final Stage firstStage;
+ private final AbstractStage firstStage;
private final List<L> lastStages = new LinkedList<L>();
- public Pipeline(final Stage firstStage, final L lastStage) {
+ public Pipeline(final AbstractStage firstStage, final L lastStage) {
this.firstStage = firstStage;
this.lastStages.add(lastStage);
}
- public Stage getFirstStage() {
+ public AbstractStage getFirstStage() {
return firstStage;
}
diff --git a/src/main/java/teetime/stage/io/filesystem/Dir2RecordsFilter.java b/src/main/java/teetime/stage/io/filesystem/Dir2RecordsFilter.java
index 66189863d431c24d18111993c52372f8aec48148..37671f222e0f1a5213deed77227e99f755d9b694 100644
--- a/src/main/java/teetime/stage/io/filesystem/Dir2RecordsFilter.java
+++ b/src/main/java/teetime/stage/io/filesystem/Dir2RecordsFilter.java
@@ -20,7 +20,7 @@ import java.io.File;
import teetime.framework.AbstractCompositeStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
import teetime.stage.FileExtensionSwitch;
import teetime.stage.basic.merger.Merger;
import teetime.stage.className.ClassNameRegistryCreationFilter;
@@ -85,7 +85,7 @@ public final class Dir2RecordsFilter extends AbstractCompositeStage {
this.classNameRegistryCreationFilter = classNameRegistryCreationFilter;
}
- public Stage getFirstStage() {
+ public AbstractStage getFirstStage() {
return classNameRegistryCreationFilter;
}
diff --git a/src/main/java/teetime/stage/io/filesystem/format/binary/DirWithBin2RecordFilter.java b/src/main/java/teetime/stage/io/filesystem/format/binary/DirWithBin2RecordFilter.java
index bed16b27225276be0c940dafd871d4a22cf79641..26f72fc33f5c5a4a7a4339fa77a710ad6938d3f9 100644
--- a/src/main/java/teetime/stage/io/filesystem/format/binary/DirWithBin2RecordFilter.java
+++ b/src/main/java/teetime/stage/io/filesystem/format/binary/DirWithBin2RecordFilter.java
@@ -20,7 +20,7 @@ import java.io.File;
import teetime.framework.AbstractCompositeStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
import teetime.stage.className.ClassNameRegistryCreationFilter;
import teetime.stage.className.ClassNameRegistryRepository;
import teetime.stage.io.filesystem.format.binary.file.BinaryFile2RecordFilter;
@@ -53,7 +53,7 @@ public class DirWithBin2RecordFilter extends AbstractCompositeStage {
this.classNameRegistryRepository = classNameRegistryRepository;
}
- protected Stage getFirstStage() {
+ protected AbstractStage getFirstStage() {
return classNameRegistryCreationFilter;
}
diff --git a/src/main/java/teetime/stage/io/filesystem/format/text/file/DatFile2RecordFilter.java b/src/main/java/teetime/stage/io/filesystem/format/text/file/DatFile2RecordFilter.java
index d2ae9fe419e84f2cf24ee74f044274fa0aa0882f..0bdc1c0027f48185e3403fb6d210772c4efe296d 100644
--- a/src/main/java/teetime/stage/io/filesystem/format/text/file/DatFile2RecordFilter.java
+++ b/src/main/java/teetime/stage/io/filesystem/format/text/file/DatFile2RecordFilter.java
@@ -20,7 +20,7 @@ import java.io.File;
import teetime.framework.AbstractCompositeStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
+import teetime.framework.AbstractStage;
import teetime.stage.className.ClassNameRegistryRepository;
import teetime.stage.io.File2TextLinesFilter;
@@ -43,7 +43,7 @@ public class DatFile2RecordFilter extends AbstractCompositeStage {
connectPorts(file2TextLinesFilter.getOutputPort(), textLine2RecordFilter.getInputPort());
}
- public Stage getFirstStage() {
+ public AbstractStage getFirstStage() {
return file2TextLinesFilter;
}
diff --git a/src/performancetest/java/teetime/examples/recordReader/RecordReaderAnalysisTest.java b/src/performancetest/java/teetime/examples/recordReader/RecordReaderAnalysisTest.java
index 395da32495071c867cbb22ec1375fc617c851f52..8b558ac5e74a8233e12e364650aee1f9e8cce110 100644
--- a/src/performancetest/java/teetime/examples/recordReader/RecordReaderAnalysisTest.java
+++ b/src/performancetest/java/teetime/examples/recordReader/RecordReaderAnalysisTest.java
@@ -23,13 +23,13 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import teetime.framework.Execution;
-import teetime.util.StopWatch;
-
import kieker.common.record.IMonitoringRecord;
import kieker.common.record.controlflow.OperationExecutionRecord;
import kieker.common.record.misc.KiekerMetadataRecord;
+import teetime.framework.Execution;
+import teetime.util.StopWatch;
+
/**
* @author Christian Wulf
*
@@ -54,7 +54,7 @@ public class RecordReaderAnalysisTest {
public void performAnalysis() {
final RecordReaderConfiguration configuration = new RecordReaderConfiguration();
- Execution analysis = new Execution(configuration);
+ Execution<RecordReaderConfiguration> analysis = new Execution<RecordReaderConfiguration>(configuration);
this.stopWatch.start();
analysis.executeBlocking();
diff --git a/src/performancetest/java/teetime/examples/recordReader/RecordReaderConfiguration.java b/src/performancetest/java/teetime/examples/recordReader/RecordReaderConfiguration.java
index f50b52208325806eafa34f92f4d230ebd6f1e173..d68de9b9d180257b4e323d8660cdf3a995f066b2 100644
--- a/src/performancetest/java/teetime/examples/recordReader/RecordReaderConfiguration.java
+++ b/src/performancetest/java/teetime/examples/recordReader/RecordReaderConfiguration.java
@@ -19,15 +19,15 @@ import java.io.File;
import java.util.LinkedList;
import java.util.List;
+import kieker.common.record.IMonitoringRecord;
+
+import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
-import teetime.framework.Stage;
import teetime.stage.CollectorSink;
import teetime.stage.InitialElementProducer;
import teetime.stage.className.ClassNameRegistryRepository;
import teetime.stage.io.filesystem.Dir2RecordsFilter;
-import kieker.common.record.IMonitoringRecord;
-
/**
* @author Christian Wulf
*
@@ -41,12 +41,7 @@ public class RecordReaderConfiguration extends Configuration {
this.buildConfiguration();
}
- private void buildConfiguration() {
- final Stage producerPipeline = this.buildProducerPipeline();
- declareActive(producerPipeline);
- }
-
- private Stage buildProducerPipeline() {
+ private AbstractStage buildConfiguration() {
ClassNameRegistryRepository classNameRegistryRepository = new ClassNameRegistryRepository();
File logDir = new File("src/test/data/bookstore-logs");
// create stages
diff --git a/src/performancetest/java/teetime/examples/traceReading/TcpTraceLoggingExtAnalysisConfiguration.java b/src/performancetest/java/teetime/examples/traceReading/TcpTraceLoggingExtAnalysisConfiguration.java
index 1e145c27f2d2a16cb0c28630437dc5d0a48efab0..85e6215cbde9710ab1fbcd249682eb13e18c57b5 100644
--- a/src/performancetest/java/teetime/examples/traceReading/TcpTraceLoggingExtAnalysisConfiguration.java
+++ b/src/performancetest/java/teetime/examples/traceReading/TcpTraceLoggingExtAnalysisConfiguration.java
@@ -17,9 +17,11 @@ package teetime.examples.traceReading;
import java.util.List;
+import kieker.common.record.IMonitoringRecord;
+
+import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
import teetime.framework.Pipeline;
-import teetime.framework.Stage;
import teetime.stage.Clock;
import teetime.stage.Counter;
import teetime.stage.ElementThroughputMeasuringStage;
@@ -27,8 +29,6 @@ import teetime.stage.basic.Sink;
import teetime.stage.basic.distributor.Distributor;
import teetime.stage.io.network.TcpReaderStage;
-import kieker.common.record.IMonitoringRecord;
-
public class TcpTraceLoggingExtAnalysisConfiguration extends Configuration {
private Counter<IMonitoringRecord> recordCounter;
@@ -40,9 +40,7 @@ public class TcpTraceLoggingExtAnalysisConfiguration extends Configuration {
private void init() {
final Pipeline<Distributor<Long>> clockPipeline = this.buildClockPipeline(1000);
- declareActive(clockPipeline.getFirstStage());
- final Stage tcpPipeline = this.buildTcpPipeline(clockPipeline.getLastStage());
- declareActive(tcpPipeline);
+ this.buildTcpPipeline(clockPipeline.getLastStage());
}
private Pipeline<Distributor<Long>> buildClockPipeline(final long intervalDelayInMs) {
@@ -56,7 +54,7 @@ public class TcpTraceLoggingExtAnalysisConfiguration extends Configuration {
return new Pipeline<Distributor<Long>>(clockStage, distributor);
}
- private Stage buildTcpPipeline(final Distributor<Long> previousClockStage) {
+ private AbstractStage buildTcpPipeline(final Distributor<Long> previousClockStage) {
TcpReaderStage tcpReader = new TcpReaderStage();
this.recordCounter = new Counter<IMonitoringRecord>();
this.recordThroughputStage = new ElementThroughputMeasuringStage<IMonitoringRecord>();
@@ -69,6 +67,8 @@ public class TcpTraceLoggingExtAnalysisConfiguration extends Configuration {
connectPorts(previousClockStage.getNewOutputPort(), this.recordThroughputStage.getTriggerInputPort(), 10);
+ recordThroughputStage.declareActive();
+
return tcpReader;
}
diff --git a/src/performancetest/java/teetime/examples/traceReconstruction/TcpTraceReconstructionConf.java b/src/performancetest/java/teetime/examples/traceReconstruction/TcpTraceReconstructionConf.java
index c05eb5a65729e20d4d07c2612b7ea6ae8d0376f7..2b2d2caa340f64de60b6b3b47f275a302537e935 100644
--- a/src/performancetest/java/teetime/examples/traceReconstruction/TcpTraceReconstructionConf.java
+++ b/src/performancetest/java/teetime/examples/traceReconstruction/TcpTraceReconstructionConf.java
@@ -17,9 +17,12 @@ package teetime.examples.traceReconstruction;
import java.util.List;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.record.flow.IFlowRecord;
+
+import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
import teetime.framework.Pipeline;
-import teetime.framework.Stage;
import teetime.stage.Clock;
import teetime.stage.Counter;
import teetime.stage.ElementThroughputMeasuringStage;
@@ -32,9 +35,6 @@ import teetime.stage.trace.traceReconstruction.EventBasedTraceFactory;
import teetime.stage.trace.traceReconstruction.TraceReconstructionFilter;
import teetime.util.ConcurrentHashMapWithDefault;
-import kieker.common.record.IMonitoringRecord;
-import kieker.common.record.flow.IFlowRecord;
-
public class TcpTraceReconstructionConf extends Configuration {
private static final int MIO = 1000000;
@@ -54,13 +54,8 @@ public class TcpTraceReconstructionConf extends Configuration {
private void init() {
Pipeline<Distributor<Long>> clockStage = this.buildClockPipeline(1000);
- declareActive(clockStage.getFirstStage());
-
Pipeline<Distributor<Long>> clock2Stage = this.buildClockPipeline(2000);
- declareActive(clock2Stage.getFirstStage());
-
- Stage pipeline = this.buildPipeline(clockStage.getLastStage(), clock2Stage.getLastStage());
- declareActive(pipeline);
+ this.buildPipeline(clockStage.getLastStage(), clock2Stage.getLastStage());
}
private Pipeline<Distributor<Long>> buildClockPipeline(final long intervalDelayInMs) {
@@ -73,7 +68,7 @@ public class TcpTraceReconstructionConf extends Configuration {
return new Pipeline<Distributor<Long>>(clock, distributor);
}
- private Stage buildPipeline(final Distributor<Long> clockStage, final Distributor<Long> clock2Stage) {
+ private AbstractStage buildPipeline(final Distributor<Long> clockStage, final Distributor<Long> clock2Stage) {
// create stages
TcpReaderStage tcpReader = new TcpReaderStage();
this.recordCounter = new Counter<IMonitoringRecord>();
diff --git a/src/performancetest/java/teetime/examples/traceReconstruction/TraceReconstructionConf.java b/src/performancetest/java/teetime/examples/traceReconstruction/TraceReconstructionConf.java
index 0afcb0504113fdb7cd703cb0428844697b55890c..25cc3a92e68cb889c2a51a90a1f3970b6c9f888f 100644
--- a/src/performancetest/java/teetime/examples/traceReconstruction/TraceReconstructionConf.java
+++ b/src/performancetest/java/teetime/examples/traceReconstruction/TraceReconstructionConf.java
@@ -19,8 +19,11 @@ import java.io.File;
import java.util.LinkedList;
import java.util.List;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.record.flow.IFlowRecord;
+
+import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
-import teetime.framework.Stage;
import teetime.stage.Cache;
import teetime.stage.Clock;
import teetime.stage.CollectorSink;
@@ -39,9 +42,6 @@ import teetime.stage.trace.traceReconstruction.EventBasedTraceFactory;
import teetime.stage.trace.traceReconstruction.TraceReconstructionFilter;
import teetime.util.ConcurrentHashMapWithDefault;
-import kieker.common.record.IMonitoringRecord;
-import kieker.common.record.flow.IFlowRecord;
-
public class TraceReconstructionConf extends Configuration {
private final List<EventBasedTrace> elementCollection = new LinkedList<EventBasedTrace>();
@@ -62,10 +62,7 @@ public class TraceReconstructionConf extends Configuration {
private void init() {
Clock clockStage = this.buildClockPipeline();
- declareActive(clockStage);
-
- Stage pipeline = this.buildPipeline(clockStage);
- declareActive(pipeline);
+ this.buildPipeline(clockStage);
}
private Clock buildClockPipeline() {
@@ -75,7 +72,7 @@ public class TraceReconstructionConf extends Configuration {
return clock;
}
- private Stage buildPipeline(final Clock clockStage) {
+ private AbstractStage buildPipeline(final Clock clockStage) {
this.classNameRegistryRepository = new ClassNameRegistryRepository();
// create stages
diff --git a/src/performancetest/java/teetime/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreadsConfiguration.java b/src/performancetest/java/teetime/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreadsConfiguration.java
index 6484124baa62987c0d606cdca3ad0510fb571a2f..759131901f53addc332828d71bfc751c8eccbefe 100644
--- a/src/performancetest/java/teetime/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreadsConfiguration.java
+++ b/src/performancetest/java/teetime/examples/traceReconstructionWithThreads/TcpTraceReconstructionAnalysisWithThreadsConfiguration.java
@@ -21,10 +21,13 @@ import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.record.flow.IFlowRecord;
+import kieker.common.record.flow.trace.TraceMetadata;
+
import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
import teetime.framework.Pipeline;
-import teetime.framework.Stage;
import teetime.framework.pipe.IMonitorablePipe;
import teetime.stage.Clock;
import teetime.stage.Counter;
@@ -41,10 +44,6 @@ import teetime.stage.trace.traceReconstruction.EventBasedTraceFactory;
import teetime.stage.trace.traceReconstruction.TraceReconstructionFilter;
import teetime.util.ConcurrentHashMapWithDefault;
-import kieker.common.record.IMonitoringRecord;
-import kieker.common.record.flow.IFlowRecord;
-import kieker.common.record.flow.trace.TraceMetadata;
-
public class TcpTraceReconstructionAnalysisWithThreadsConfiguration extends Configuration {
private static final int NUM_VIRTUAL_CORES = Runtime.getRuntime().availableProcessors();
@@ -92,17 +91,12 @@ public class TcpTraceReconstructionAnalysisWithThreadsConfiguration extends Conf
private void init() {
Pipeline<Distributor<IMonitoringRecord>> tcpPipeline = this.buildTcpPipeline();
- declareActive(tcpPipeline.getFirstStage());
-
Pipeline<Distributor<Long>> clockStage = this.buildClockPipeline(1000);
- declareActive(clockStage.getFirstStage());
-
Pipeline<Distributor<Long>> clock2Stage = this.buildClockPipeline(2000);
- declareActive(clock2Stage.getFirstStage());
for (int i = 0; i < this.numWorkerThreads; i++) {
- Stage pipeline = this.buildPipeline(tcpPipeline.getLastStage(), clockStage.getLastStage(), clock2Stage.getLastStage());
- declareActive(pipeline);
+ AbstractStage pipeline = this.buildPipeline(tcpPipeline.getLastStage(), clockStage.getLastStage(), clock2Stage.getLastStage());
+ pipeline.declareActive();
}
}
@@ -156,7 +150,7 @@ public class TcpTraceReconstructionAnalysisWithThreadsConfiguration extends Conf
}
}
- private Stage buildPipeline(final Distributor<IMonitoringRecord> tcpReaderPipeline, final Distributor<Long> clockStage,
+ private AbstractStage buildPipeline(final Distributor<IMonitoringRecord> tcpReaderPipeline, final Distributor<Long> clockStage,
final Distributor<Long> clock2Stage) {
// create stages
Relay<IMonitoringRecord> relay = new Relay<IMonitoringRecord>();
@@ -191,6 +185,9 @@ public class TcpTraceReconstructionAnalysisWithThreadsConfiguration extends Conf
// connectPorts(traceThroughputFilter.getOutputPort(), traceCounter.getInputPort());
connectPorts(traceCounter.getOutputPort(), endStage.getInputPort());
+ recordThroughputFilter.declareActive();
+ traceThroughputFilter.declareActive();
+
return relay;
}
diff --git a/src/performancetest/java/teetime/examples/traceReductionWithThreads/TcpTraceReductionAnalysisWithThreadsConfiguration.java b/src/performancetest/java/teetime/examples/traceReductionWithThreads/TcpTraceReductionAnalysisWithThreadsConfiguration.java
index ebea3fcdabeb5be84c6c2dc468a4064e639663a4..876d38e92ad8593919791e73ff7010efb2ab702d 100644
--- a/src/performancetest/java/teetime/examples/traceReductionWithThreads/TcpTraceReductionAnalysisWithThreadsConfiguration.java
+++ b/src/performancetest/java/teetime/examples/traceReductionWithThreads/TcpTraceReductionAnalysisWithThreadsConfiguration.java
@@ -23,10 +23,13 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.record.flow.IFlowRecord;
+import kieker.common.record.flow.trace.TraceMetadata;
+
import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
import teetime.framework.Pipeline;
-import teetime.framework.Stage;
import teetime.framework.pipe.IMonitorablePipe;
import teetime.stage.Clock;
import teetime.stage.Counter;
@@ -46,10 +49,6 @@ import teetime.stage.trace.traceReduction.TraceAggregationBuffer;
import teetime.stage.trace.traceReduction.TraceReductionFilter;
import teetime.util.ConcurrentHashMapWithDefault;
-import kieker.common.record.IMonitoringRecord;
-import kieker.common.record.flow.IFlowRecord;
-import kieker.common.record.flow.trace.TraceMetadata;
-
public class TcpTraceReductionAnalysisWithThreadsConfiguration extends Configuration {
private static final int NUM_VIRTUAL_CORES = Runtime.getRuntime().availableProcessors();
@@ -88,17 +87,11 @@ public class TcpTraceReductionAnalysisWithThreadsConfiguration extends Configura
private void init() {
final Pipeline<Distributor<IMonitoringRecord>> tcpPipeline = this.buildTcpPipeline();
- declareActive(tcpPipeline.getFirstStage());
-
final Pipeline<Distributor<Long>> clockStage = this.buildClockPipeline(1000);
- declareActive(clockStage.getFirstStage());
-
final Pipeline<Distributor<Long>> clock2Stage = this.buildClockPipeline(5000);
- declareActive(clock2Stage.getFirstStage());
-
for (int i = 0; i < this.numWorkerThreads; i++) {
- final Stage pipeline = this.buildPipeline(tcpPipeline, clockStage, clock2Stage);
- declareActive(pipeline);
+ final AbstractStage pipeline = this.buildPipeline(tcpPipeline, clockStage, clock2Stage);
+ pipeline.declareActive();
}
}
@@ -158,7 +151,7 @@ public class TcpTraceReductionAnalysisWithThreadsConfiguration extends Configura
private final StageFactory<Counter<TraceAggregationBuffer>> traceCounterFactory;
private final StageFactory<ElementThroughputMeasuringStage<TraceAggregationBuffer>> traceThroughputFilterFactory;
- private Stage buildPipeline(final Pipeline<Distributor<IMonitoringRecord>> tcpPipeline, final Pipeline<Distributor<Long>> clockStage,
+ private AbstractStage buildPipeline(final Pipeline<Distributor<IMonitoringRecord>> tcpPipeline, final Pipeline<Distributor<Long>> clockStage,
final Pipeline<Distributor<Long>> clock2Stage) {
// create stages
Relay<IMonitoringRecord> relay = new Relay<IMonitoringRecord>();
@@ -192,6 +185,9 @@ public class TcpTraceReductionAnalysisWithThreadsConfiguration extends Configura
connectPorts(clock2Stage.getLastStage().getNewOutputPort(), traceReductionFilter.getTriggerInputPort(), 10);
connectPorts(clockStage.getLastStage().getNewOutputPort(), traceThroughputFilter.getTriggerInputPort(), 10);
+ traceReductionFilter.declareActive();
+ traceThroughputFilter.declareActive();
+
return relay;
}
diff --git a/src/test/java/teetime/stage/io/filesystem/Dir2RecordsFilterTest.java b/src/test/java/teetime/stage/io/filesystem/Dir2RecordsFilterTest.java
index dc3f1d3242f1b3dd6159d5348f5b11172dc2f409..8839df2177cdd3bf3b8be23e8d35dd370960f9ed 100644
--- a/src/test/java/teetime/stage/io/filesystem/Dir2RecordsFilterTest.java
+++ b/src/test/java/teetime/stage/io/filesystem/Dir2RecordsFilterTest.java
@@ -19,17 +19,17 @@ import java.io.File;
import org.junit.Test;
+import kieker.common.record.IMonitoringRecord;
+
import teetime.framework.AbstractCompositeStage;
+import teetime.framework.AbstractStage;
import teetime.framework.Configuration;
import teetime.framework.Execution;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
import teetime.stage.InitialElementProducer;
import teetime.stage.className.ClassNameRegistryRepository;
import teetime.stage.io.Printer;
-import kieker.common.record.IMonitoringRecord;
-
public class Dir2RecordsFilterTest {
class TestConfiguration extends Configuration {
@@ -52,14 +52,13 @@ public class Dir2RecordsFilterTest {
this.reader = new Dir2RecordsFilter(new ClassNameRegistryRepository());
connectPorts(this.producer.getOutputPort(), this.reader.getInputPort());
- declareActive(producer);
}
public OutputPort<IMonitoringRecord> getOutputPort() {
return this.reader.getOutputPort();
}
- public Stage getFirstStage() {
+ public AbstractStage getFirstStage() {
return this.producer;
}