Merge remote-tracking branch 'origin/removed-onIsPipelineHead'

Conflicts:
	src/main/java/teetime/framework/HeadPipeline.java
	src/main/java/teetime/framework/Pipeline.java
	src/main/java/teetime/framework/StageWithPort.java

src/main/java/teetime/framework/AbstractStage.java

@@ -126,7 +126,6 @@ public abstract class AbstractStage implements Stage {
 
 	public void onTerminating() {
 		// empty default implementation
-		this.onIsPipelineHead();
 	}
 
 	protected <T> InputPort<T> createInputPort() {

src/main/java/teetime/framework/ConsumerStage.java

@@ -15,11 +15,6 @@ public abstract class ConsumerStage<I> extends AbstractStage {
 		this.execute(element);
 	}
 
-	@Override
-	public void onIsPipelineHead() {
-		// do nothing
-	}
-
 	protected abstract void execute(I element);
 
 }

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

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

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

@@ -5,7 +5,7 @@ import java.util.List;
 import teetime.framework.signal.ISignal;
 import teetime.framework.validation.InvalidPortConnection;
 
-public class Pipeline<FirstStage extends Stage, LastStage extends Stage> implements Stage {
+public class OldPipeline<FirstStage extends Stage, LastStage extends Stage> implements Stage {
 
 	protected FirstStage firstStage;
 	protected LastStage lastStage;
@@ -46,11 +46,6 @@ public class Pipeline<FirstStage extends Stage, LastStage extends Stage> impleme
 		this.firstStage.setParentStage(parentStage, index);
 	}
 
-	@Override
-	public void onIsPipelineHead() {
-		this.firstStage.onIsPipelineHead();
-	}
-
 	@Override
 	public void onSignal(final ISignal signal, final InputPort<?> inputPort) {
 		this.firstStage.onSignal(signal, inputPort);

src/main/java/teetime/framework/ProducerStage.java

@@ -23,11 +23,6 @@ public abstract class ProducerStage<O> extends AbstractStage implements HeadStag
 		this.execute();
 	}
 
-	@Override
-	public void onIsPipelineHead() {
-		// do nothing
-	}
-
 	@Override
 	public void terminate() {
 		this.shouldTerminate = true;

src/main/java/teetime/framework/Stage.java

@@ -15,9 +15,6 @@ public interface Stage {
 
 	void setParentStage(Stage parentStage, int index);
 
-	// BETTER remove this method since it will be replaced by onTerminating()
-	void onIsPipelineHead();
-
 	void onSignal(ISignal signal, InputPort<?> inputPort);
 
 	/**

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

@@ -20,7 +20,8 @@ public class Cache<T> extends ConsumerStage<T> {
 	}
 
 	@Override
-	public void onIsPipelineHead() {
+	public void onTerminating() {
+		super.onTerminating();
 		this.logger.debug("Emitting " + this.cachedObjects.size() + " cached elements...");
 		StopWatch stopWatch = new StopWatch();
 		stopWatch.start();
@@ -29,7 +30,7 @@ public class Cache<T> extends ConsumerStage<T> {
 		}
 		stopWatch.end();
 		this.logger.debug("Emitting took " + TimeUnit.NANOSECONDS.toMillis(stopWatch.getDurationInNs()) + " ms");
-		super.onIsPipelineHead();
+		super.onTerminating();
 	}
 
 	public OutputPort<T> getOutputPort() {

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

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

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

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

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

@@ -21,9 +21,9 @@ import teetime.framework.OutputPort;
 
 /**
  * @author Christian Wulf
- * 
+ *
  * @since 1.10
- * 
+ *
  * @param T
  *            the type of the input port and the output ports
  */
@@ -38,7 +38,8 @@ public class Distributor<T> extends ConsumerStage<T> {
 	}
 
 	@Override
-	public void onIsPipelineHead() {
+	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

@@ -62,7 +62,8 @@ public class Merger<T> extends AbstractStage {
 	}
 
 	@Override
-	public void onIsPipelineHead() {
+	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

@@ -149,10 +149,11 @@ public class TCPReaderSink extends ProducerStage<IMonitoringRecord> {
 	}
 
 	@Override
-	public void onIsPipelineHead() {
+	public void onTerminating() {
+		super.onTerminating();
 		this.executorService.shutdown();
 		this.tcpStringReader.interrupt();
-		super.onIsPipelineHead();
+		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

@@ -29,7 +29,7 @@ import kieker.common.record.flow.trace.TraceMetadata;
 
 /**
  * @author Christian Wulf
- * 
+ *
  * @since 1.10
  */
 public class TraceReconstructionFilter extends ConsumerStage<IFlowRecord> {
@@ -94,12 +94,13 @@ public class TraceReconstructionFilter extends ConsumerStage<IFlowRecord> {
 	}
 
 	@Override
-	public void onIsPipelineHead() {
+	public void onTerminating() {
+		super.onTerminating();
 		for (Long traceId : this.traceId2trace.keySet()) {
 			this.put(traceId, false);
 		}
 
-		super.onIsPipelineHead();
+		super.onTerminating();
 	}
 
 	private void sendTraceBuffer(final TraceBuffer traceBuffer) {

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

@@ -30,11 +30,11 @@ import kieker.analysis.plugin.filter.flow.TraceEventRecords;
  * This filter collects incoming traces for a specified amount of time.
  * Any traces representing the same series of events will be used to calculate statistical informations like the average runtime of this kind of trace.
  * Only one specimen of these traces containing this information will be forwarded from this filter.
- * 
+ *
  * Statistical outliers regarding the runtime of the trace will be treated special and therefore send out as they are and will not be mixed with others.
- * 
+ *
  * @author Jan Waller, Florian Biss
- * 
+ *
  * @since
  */
 public class TraceReductionFilter extends ConsumerStage<TraceEventRecords> {
@@ -73,7 +73,8 @@ public class TraceReductionFilter extends ConsumerStage<TraceEventRecords> {
 	}
 
 	@Override
-	public void onIsPipelineHead() {
+	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();
@@ -84,7 +85,7 @@ public class TraceReductionFilter extends ConsumerStage<TraceEventRecords> {
 			this.trace2buffer.clear();
 		}
 
-		super.onIsPipelineHead();
+		super.onTerminating();
 	}
 
 	private void processTimeoutQueue(final long timestampInNs) {