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..c22fa6ac1e7e2fd4359e97a7c71b67c2734feded 100644
--- a/src/main/java/teetime/stage/io/filesystem/Dir2RecordsFilter.java
+++ b/src/main/java/teetime/stage/io/filesystem/Dir2RecordsFilter.java
@@ -17,10 +17,14 @@ package teetime.stage.io.filesystem;
import java.io.File;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.util.filesystem.BinaryCompressionMethod;
+import kieker.common.util.filesystem.FSUtil;
+
import teetime.framework.AbstractCompositeStage;
+import teetime.framework.AbstractStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
import teetime.stage.FileExtensionSwitch;
import teetime.stage.basic.merger.Merger;
import teetime.stage.className.ClassNameRegistryCreationFilter;
@@ -29,10 +33,6 @@ import teetime.stage.io.Directory2FilesFilter;
import teetime.stage.io.filesystem.format.binary.file.BinaryFile2RecordFilter;
import teetime.stage.io.filesystem.format.text.file.DatFile2RecordFilter;
-import kieker.common.record.IMonitoringRecord;
-import kieker.common.util.filesystem.BinaryCompressionMethod;
-import kieker.common.util.filesystem.FSUtil;
-
/**
* @author Christian Wulf
*
@@ -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..9a405c589f77c504546e55626d8aa205ba7dd3df 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
@@ -17,16 +17,16 @@ package teetime.stage.io.filesystem.format.binary;
import java.io.File;
+import kieker.common.record.IMonitoringRecord;
+
import teetime.framework.AbstractCompositeStage;
+import teetime.framework.AbstractStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
import teetime.stage.className.ClassNameRegistryCreationFilter;
import teetime.stage.className.ClassNameRegistryRepository;
import teetime.stage.io.filesystem.format.binary.file.BinaryFile2RecordFilter;
-import kieker.common.record.IMonitoringRecord;
-
public class DirWithBin2RecordFilter extends AbstractCompositeStage {
private final ClassNameRegistryCreationFilter classNameRegistryCreationFilter;
@@ -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..a5e1f8d4b082b9ff0fa40cc04376f4417ee1a4fc 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
@@ -17,15 +17,15 @@ package teetime.stage.io.filesystem.format.text.file;
import java.io.File;
+import kieker.common.record.IMonitoringRecord;
+
import teetime.framework.AbstractCompositeStage;
+import teetime.framework.AbstractStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
-import teetime.framework.Stage;
import teetime.stage.className.ClassNameRegistryRepository;
import teetime.stage.io.File2TextLinesFilter;
-import kieker.common.record.IMonitoringRecord;
-
/**
* @author Christian Wulf
*
@@ -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/RecordReaderConfiguration.java b/src/performancetest/java/teetime/examples/recordReader/RecordReaderConfiguration.java
index f50b52208325806eafa34f92f4d230ebd6f1e173..6c9850580a9076cfae51cd69b018a2b2fc8f4a3b 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
*
@@ -42,11 +42,11 @@ public class RecordReaderConfiguration extends Configuration {
}
private void buildConfiguration() {
- final Stage producerPipeline = this.buildProducerPipeline();
+ final AbstractStage producerPipeline = this.buildProducerPipeline();
declareActive(producerPipeline);
}
- private Stage buildProducerPipeline() {
+ private AbstractStage buildProducerPipeline() {
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..d624114a3720e59a7c1bc9f788eddf703767acea 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;
@@ -41,7 +41,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());
+ final AbstractStage tcpPipeline = this.buildTcpPipeline(clockPipeline.getLastStage());
declareActive(tcpPipeline);
}
@@ -56,7 +56,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>();
diff --git a/src/performancetest/java/teetime/examples/traceReconstruction/TcpTraceReconstructionConf.java b/src/performancetest/java/teetime/examples/traceReconstruction/TcpTraceReconstructionConf.java
index c05eb5a65729e20d4d07c2612b7ea6ae8d0376f7..c84dd0a1810290ffa47894eafff056af3690bc25 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;
@@ -59,7 +59,7 @@ public class TcpTraceReconstructionConf extends Configuration {
Pipeline<Distributor<Long>> clock2Stage = this.buildClockPipeline(2000);
declareActive(clock2Stage.getFirstStage());
- Stage pipeline = this.buildPipeline(clockStage.getLastStage(), clock2Stage.getLastStage());
+ AbstractStage pipeline = this.buildPipeline(clockStage.getLastStage(), clock2Stage.getLastStage());
declareActive(pipeline);
}
@@ -73,7 +73,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..717ff484550d90ae56e94f739f07721e899f4bef 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>();
@@ -64,7 +64,7 @@ public class TraceReconstructionConf extends Configuration {
Clock clockStage = this.buildClockPipeline();
declareActive(clockStage);
- Stage pipeline = this.buildPipeline(clockStage);
+ AbstractStage pipeline = this.buildPipeline(clockStage);
declareActive(pipeline);
}
@@ -75,7 +75,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..47a284c7c76b36410b6db8fe0ccfd5a99d05855c 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();
@@ -101,7 +100,7 @@ public class TcpTraceReconstructionAnalysisWithThreadsConfiguration extends Conf
declareActive(clock2Stage.getFirstStage());
for (int i = 0; i < this.numWorkerThreads; i++) {
- Stage pipeline = this.buildPipeline(tcpPipeline.getLastStage(), clockStage.getLastStage(), clock2Stage.getLastStage());
+ AbstractStage pipeline = this.buildPipeline(tcpPipeline.getLastStage(), clockStage.getLastStage(), clock2Stage.getLastStage());
declareActive(pipeline);
}
}
@@ -156,7 +155,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>();
diff --git a/src/performancetest/java/teetime/examples/traceReductionWithThreads/TcpTraceReductionAnalysisWithThreadsConfiguration.java b/src/performancetest/java/teetime/examples/traceReductionWithThreads/TcpTraceReductionAnalysisWithThreadsConfiguration.java
index ebea3fcdabeb5be84c6c2dc468a4064e639663a4..606724385bb1164774ed4df6e28802a8f52371ce 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();
@@ -97,7 +96,7 @@ public class TcpTraceReductionAnalysisWithThreadsConfiguration extends Configura
declareActive(clock2Stage.getFirstStage());
for (int i = 0; i < this.numWorkerThreads; i++) {
- final Stage pipeline = this.buildPipeline(tcpPipeline, clockStage, clock2Stage);
+ final AbstractStage pipeline = this.buildPipeline(tcpPipeline, clockStage, clock2Stage);
declareActive(pipeline);
}
}
@@ -158,7 +157,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>();
diff --git a/src/test/java/dummy b/src/test/java/dummy
deleted file mode 100644
index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000
diff --git a/src/test/java/teetime/stage/io/filesystem/Dir2RecordsFilterTest.java b/src/test/java/teetime/stage/io/filesystem/Dir2RecordsFilterTest.java
index dc3f1d3242f1b3dd6159d5348f5b11172dc2f409..486bba06441487ef96a52b9c0a4e68c8be0abfb3 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 {
@@ -59,7 +59,7 @@ public class Dir2RecordsFilterTest {
return this.reader.getOutputPort();
}
- public Stage getFirstStage() {
+ public AbstractStage getFirstStage() {
return this.producer;
}