added super call to classes which override onTerminating

renamed Pipeline to OldPipeline and marked as deprecated #19

src/main/java/teetime/framework/HeadPipeline.java

 package teetime.framework;
 
-public class HeadPipeline<FirstStage extends HeadStage, LastStage extends StageWithPort> extends Pipeline<FirstStage, LastStage> implements HeadStage {
+public class HeadPipeline<FirstStage extends HeadStage, LastStage extends StageWithPort> extends OldPipeline<FirstStage, LastStage> implements HeadStage {
 
 	public HeadPipeline() {}
 

src/main/java/teetime/framework/Pipeline.java

-package teetime.framework;
-
-import java.util.List;
-
-import teetime.framework.signal.ISignal;
-import teetime.framework.validation.InvalidPortConnection;
-
-public class Pipeline<FirstStage extends StageWithPort, LastStage extends StageWithPort> implements StageWithPort {
-
-	protected FirstStage firstStage;
-	protected LastStage lastStage;
-
-	public FirstStage getFirstStage() {
-		return this.firstStage;
-	}
-
-	public void setFirstStage(final FirstStage firstStage) {
-		this.firstStage = firstStage;
-	}
-
-	public LastStage getLastStage() {
-		return this.lastStage;
-	}
-
-	public void setLastStage(final LastStage lastStage) {
-		this.lastStage = lastStage;
-	}
-
-	@Override
-	public String getId() {
-		return this.firstStage.getId();
-	}
-
-	@Override
-	public void executeWithPorts() {
-		this.firstStage.executeWithPorts();
-	}
-
-	@Override
-	public StageWithPort getParentStage() {
-		return this.firstStage.getParentStage();
-	}
-
-	@Override
-	public void setParentStage(final StageWithPort parentStage, final int index) {
-		this.firstStage.setParentStage(parentStage, index);
-	}
-
-	@Override
-	public void onSignal(final ISignal signal, final InputPort<?> inputPort) {
-		this.firstStage.onSignal(signal, inputPort);
-	}
-
-	@Override
-	public void validateOutputPorts(final List<InvalidPortConnection> invalidPortConnections) {
-		this.lastStage.validateOutputPorts(invalidPortConnections);
-	}
-
-}

src/main/java/teetime/stage/Cache.java

@@ -21,6 +21,7 @@ public class Cache<T> extends ConsumerStage<T> {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		this.logger.debug("Emitting " + this.cachedObjects.size() + " cached elements...");
 		StopWatch stopWatch = new StopWatch();
 		stopWatch.start();

src/main/java/teetime/stage/CollectorSink.java

@@ -46,6 +46,7 @@ public class CollectorSink<T> extends ConsumerStage<T> {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		System.out.println("size: " + this.elements.size());
 	}
 

src/main/java/teetime/stage/basic/Delay.java

@@ -28,6 +28,7 @@ public class Delay<T> extends AbstractStage {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		while (!this.inputPort.getPipe().isEmpty()) {
 			this.executeWithPorts();
 		}

src/main/java/teetime/stage/basic/distributor/Distributor.java

@@ -39,6 +39,7 @@ public class Distributor<T> extends ConsumerStage<T> {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		// for (OutputPort<T> op : this.outputPortList) {
 		// op.getPipe().close();
 		// System.out.println("End signal sent, size: " + op.getPipe().size());

src/main/java/teetime/stage/basic/merger/Merger.java

@@ -63,6 +63,7 @@ public class Merger<T> extends AbstractStage {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		this.finishedInputPorts++;
 	}
 

src/main/java/teetime/stage/kieker/Dir2RecordsFilter.java

@@ -19,7 +19,7 @@ import java.io.File;
 
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
-import teetime.framework.Pipeline;
+import teetime.framework.OldPipeline;
 import teetime.framework.pipe.PipeFactory;
 import teetime.framework.pipe.SingleElementPipe;
 import teetime.framework.pipe.PipeFactory.PipeOrdering;
@@ -42,7 +42,7 @@ import kieker.common.util.filesystem.FSUtil;
  *
  * @since 1.10
  */
-public class Dir2RecordsFilter extends Pipeline<ClassNameRegistryCreationFilter, Merger<IMonitoringRecord>> {
+public class Dir2RecordsFilter extends OldPipeline<ClassNameRegistryCreationFilter, Merger<IMonitoringRecord>> {
 
 	private final PipeFactory pipeFactory = PipeFactory.INSTANCE;
 	private ClassNameRegistryRepository classNameRegistryRepository;

src/main/java/teetime/stage/kieker/DirWithBin2RecordFilter.java

@@ -4,7 +4,7 @@ import java.io.File;
 
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
-import teetime.framework.Pipeline;
+import teetime.framework.OldPipeline;
 import teetime.stage.io.Directory2FilesFilter;
 import teetime.stage.kieker.className.ClassNameRegistryCreationFilter;
 import teetime.stage.kieker.className.ClassNameRegistryRepository;
@@ -12,7 +12,7 @@ import teetime.stage.kieker.fileToRecord.BinaryFile2RecordFilter;
 
 import kieker.common.record.IMonitoringRecord;
 
-public class DirWithBin2RecordFilter extends Pipeline<ClassNameRegistryCreationFilter, BinaryFile2RecordFilter> {
+public class DirWithBin2RecordFilter extends OldPipeline<ClassNameRegistryCreationFilter, BinaryFile2RecordFilter> {
 
 	private ClassNameRegistryRepository classNameRegistryRepository;
 

src/main/java/teetime/stage/kieker/DirWithDat2RecordFilter.java

@@ -4,7 +4,7 @@ import java.io.File;
 
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
-import teetime.framework.Pipeline;
+import teetime.framework.OldPipeline;
 import teetime.stage.io.Directory2FilesFilter;
 import teetime.stage.kieker.className.ClassNameRegistryCreationFilter;
 import teetime.stage.kieker.className.ClassNameRegistryRepository;
@@ -12,7 +12,7 @@ import teetime.stage.kieker.fileToRecord.DatFile2RecordFilter;
 
 import kieker.common.record.IMonitoringRecord;
 
-public class DirWithDat2RecordFilter extends Pipeline<ClassNameRegistryCreationFilter, DatFile2RecordFilter> {
+public class DirWithDat2RecordFilter extends OldPipeline<ClassNameRegistryCreationFilter, DatFile2RecordFilter> {
 
 	private ClassNameRegistryRepository classNameRegistryRepository;
 

src/main/java/teetime/stage/kieker/TCPReaderSink.java

@@ -150,6 +150,7 @@ public class TCPReaderSink extends ProducerStage<IMonitoringRecord> {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		this.executorService.shutdown();
 		this.tcpStringReader.interrupt();
 		super.onTerminating();

src/main/java/teetime/stage/kieker/fileToRecord/DatFile2RecordFilter.java

@@ -19,7 +19,7 @@ import java.io.File;
 
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
-import teetime.framework.Pipeline;
+import teetime.framework.OldPipeline;
 import teetime.framework.pipe.SingleElementPipe;
 import teetime.stage.io.File2TextLinesFilter;
 import teetime.stage.kieker.className.ClassNameRegistryRepository;
@@ -32,7 +32,7 @@ import kieker.common.record.IMonitoringRecord;
  *
  * @since 1.10
  */
-public class DatFile2RecordFilter extends Pipeline<File2TextLinesFilter, TextLine2RecordFilter> {
+public class DatFile2RecordFilter extends OldPipeline<File2TextLinesFilter, TextLine2RecordFilter> {
 
 	public DatFile2RecordFilter(final ClassNameRegistryRepository classNameRegistryRepository) {
 		File2TextLinesFilter file2TextLinesFilter = new File2TextLinesFilter();

src/main/java/teetime/stage/kieker/traceReconstruction/TraceReconstructionFilter.java

@@ -95,6 +95,7 @@ public class TraceReconstructionFilter extends ConsumerStage<IFlowRecord> {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		for (Long traceId : this.traceId2trace.keySet()) {
 			this.put(traceId, false);
 		}

src/main/java/teetime/stage/kieker/traceReduction/TraceReductionFilter.java

@@ -74,6 +74,7 @@ public class TraceReductionFilter extends ConsumerStage<TraceEventRecords> {
 
 	@Override
 	public void onTerminating() {
+		super.onTerminating();
 		synchronized (this.trace2buffer) { // BETTER hide and improve synchronization in the buffer
 			for (final Entry<TraceEventRecords, TraceAggregationBuffer> entry : this.trace2buffer.entrySet()) {
 				final TraceAggregationBuffer buffer = entry.getValue();