init

.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src/main/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.6">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="output" path="target/classes"/>
+</classpath>

.gitignore

+/bin
+/target

.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>teetime</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+	</natures>
+</projectDescription>

.settings/org.eclipse.core.resources.prefs

+eclipse.preferences.version=1
+encoding//src/main/java=UTF-8
+encoding//src/test/java=UTF-8
+encoding/<project>=UTF-8

.settings/org.eclipse.jdt.core.prefs

+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.6
+org.eclipse.jdt.core.compiler.compliance=1.6
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
+org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.source=1.6

.settings/org.eclipse.m2e.core.prefs

+activeProfiles=
+eclipse.preferences.version=1
+resolveWorkspaceProjects=true
+version=1

teetime/pom.xml → pom.xml

@@ -18,8 +18,28 @@
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
-      <version>3.8.1</version>
+      <version>4.11</version>
       <scope>test</scope>
     </dependency>
+	<dependency>
+		<groupId>org.hamcrest</groupId>
+		<artifactId>hamcrest-core</artifactId>
+		<version>1.3</version>
+	</dependency>
+	<dependency>
+		<groupId>org.hamcrest</groupId>
+		<artifactId>hamcrest-library</artifactId>
+		<version>1.3</version>
+	</dependency>
+	<dependency>
+      <groupId>net.kieker-monitoring</groupId>
+      <artifactId>kieker</artifactId>
+      <version>1.9</version>
+    </dependency>
+	<dependency>
+		<groupId>com.google.guava</groupId>
+		<artifactId>guava</artifactId>
+		<version>17.0</version>
+	</dependency>
   </dependencies>
 </project>

results/chw-work/Kieker.csv

+50;222974790;249344303;255853039;259406468;302300903;256949018;3867545
+100;479268312;501520758;514242657;520879474;547129527;512607537;3861798
+150;769605950;814607155;827869293;840828881;997575901;832140772;10406475
+200;1053294865;1114439212;1133247837;1174248560;1374436002;1150302103;17160918
+250;1347713034;1403914765;1423028403;1445675354;1514311623;1429062984;10029167
+300;1634709890;1692022888;1716632518;1733769931;1845717310;1714584838;10851185
+350;1898198121;1979121553;2005629409;2021975347;2104626641;2002653027;12416337
+400;2184222053;2264418050;2297249500;2347125585;2571002690;2312469267;20003529
+450;2114346436;2474942703;2513493875;2554615702;2821436913;2484613004;39023338
+500;2714958023;2807430509;2834917858;2855149332;2908789030;2833729060;9892356
+550;2953534484;3129759679;3152678393;3207891190;3355362594;3171670501;21159351
+600;3375834220;3456305673;3507196145;3564639277;4045959350;3525562922;30257420
+650;3733499812;3793166710;3842865043;3931548874;4501397192;3901445690;47083708
+700;3902291379;4145148703;4226156701;4324776258;5536500029;4373255110;108796825
+750;4488138338;4536918344;4577786471;4664000637;4766312788;4598726882;21193905
+800;4787523343;4899780250;4954223740;5032639329;5227891132;4974727780;27106090
+850;5330114612;5428669718;5497654926;5559696251;5744444585;5503212946;26634323
+900;4915166811;5759447724;5940546980;6032643433;7847659872;6004221609;120237698
+950;6278789404;6558588268;6770077633;7249897679;10036245303;7069647198;217586495
+1000;6998942511;7251777843;7319524934;7434979353;8077187734;7366103358;56194326

results/chw-work/TeeTime.csv

+50;81005535;92166652;93302961;95672050;103493206;93511502;1038983
+100;174480091;183187963;185029128;188204061;194225513;185294239;1154436
+150;239712770;274010174;277981097;280009048;295593477;276621639;2183336
+200;337821234;367758134;371051706;377328500;478632833;374718182;5375113
+250;385948931;442055422;457899298;465820620;524613120;452422321;7162054
+300;514072703;542246436;547270333;553017151;564887640;546601274;2986762
+350;615707092;645556962;650576755;657115458;671433855;651241364;2766588
+400;700708593;734629552;740820958;745605936;754731302;739358059;3082558
+450;803342998;831226084;835117780;840909342;850780207;835062277;2948398
+500;876183364;920951807;930805839;937619999;963086376;929382628;4780910
+550;1003258678;1028639258;1035441512;1042563969;1072049712;1035805640;3409157
+600;1093110018;1131170625;1137555794;1147802691;1268041677;1142788571;7089446
+650;1181869795;1215763661;1224030638;1233324726;1278020097;1223997024;4308705
+700;1277366144;1316095891;1326646981;1336546582;1378479587;1327486503;4854941
+750;1372059934;1404826111;1418028315;1430832318;1457884930;1416883852;5369178
+800;1474965281;1493768979;1500705882;1512160518;1543885627;1503343105;3938081
+850;1546340924;1592571626;1605533267;1613038736;1629217593;1601614338;4648053
+900;1604224130;1687782667;1702500907;1711553611;1828215358;1701607220;9538072
+950;1737332802;1772582603;1787626793;1810397720;1878440793;1792495376;8716251
+1000;1835066054;1875513813;1889568248;1905512700;1982777611;1896242150;8591685

src/main/java/experiment/Experiment1.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package experiment;
+
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import javax.security.auth.login.Configuration;
+
+import kieker.analysis.AnalysisController;
+import kieker.analysis.IAnalysisController;
+import kieker.analysis.stage.EmptyPassOnFilter;
+import kieker.analysis.stage.ObjectProducer;
+import teetime.framework.concurrent.StageTerminationPolicy;
+import teetime.framework.concurrent.WorkerThread;
+import teetime.framework.core.Analysis;
+import teetime.framework.core.IStage;
+import teetime.framework.core.Pipeline;
+import teetime.framework.sequential.QueuePipe;
+import teetime.stage.NoopFilter;
+import teetime.util.StatisticsUtil;
+
+/**
+ * @author Nils Christian Ehmke
+ * 
+ * @since 1.10
+ */
+public class Experiment1 {
+
+	private static final int NUMBER_OF_WARMUP_RUNS_PER_EXPERIMENT = 5;
+	private static final int NUMBER_OF_MEASURED_RUNS_PER_EXPERIMENT = 50;
+
+	private static final int NUMBER_OF_OBJECTS_TO_SEND = 10000;
+
+	private static final int NUMBER_OF_MINIMAL_FILTERS = 50;
+	private static final int NUMBER_OF_MAXIMAL_FILTERS = 1000;
+	private static final int NUMBER_OF_FILTERS_PER_STEP = 50;
+
+	private static final IAnalysis[] analyses = { new TeeTimeAnalysis(), new KiekerAnalysis() };
+
+	private static final List<Long> measuredTimes = new ArrayList<Long>();
+
+	public static void main(final String[] args) throws Exception {
+		System.setProperty("kieker.common.logging.Log", "NONE");
+
+		for (final IAnalysis analysis : analyses) {
+			for (int numberOfFilters = NUMBER_OF_MINIMAL_FILTERS; numberOfFilters <= NUMBER_OF_MAXIMAL_FILTERS; numberOfFilters += NUMBER_OF_FILTERS_PER_STEP) {
+				// Warmup
+				for (int run = 0; run < NUMBER_OF_WARMUP_RUNS_PER_EXPERIMENT; run++) {
+					analysis.initialize(numberOfFilters, NUMBER_OF_OBJECTS_TO_SEND);
+					analysis.execute();
+				}
+
+				// Actual measurement
+				for (int run = 0; run < NUMBER_OF_MEASURED_RUNS_PER_EXPERIMENT; run++) {
+					final long tin = System.nanoTime();
+
+					analysis.initialize(numberOfFilters, NUMBER_OF_OBJECTS_TO_SEND);
+					analysis.execute();
+
+					final long tout = System.nanoTime();
+					Experiment1.addMeasuredTime((tout - tin));
+				}
+
+				Experiment1.writeAndClearMeasuredTime(analysis.getName(), numberOfFilters);
+			}
+		}
+	}
+
+	private static void addMeasuredTime(final long time) {
+		measuredTimes.add(new Long(time));
+	}
+
+	private static void writeAndClearMeasuredTime(final String analysisName, final int numberOfFilters) throws IOException {
+		final FileWriter fileWriter = new FileWriter(analysisName + ".csv", true);
+		fileWriter.write(Integer.toString(numberOfFilters));
+		fileWriter.write(";");
+
+		final Map<Double, Long> quintiles = StatisticsUtil.calculateQuintiles(measuredTimes);
+		for (final Long value : quintiles.values()) {
+			fileWriter.write(Long.toString(value));
+			fileWriter.write(";");
+		}
+
+		fileWriter.write(Long.toString(StatisticsUtil.calculateAverage(measuredTimes)));
+		fileWriter.write(";");
+
+		fileWriter.write(Long.toString(StatisticsUtil.calculateConfidenceWidth(measuredTimes)));
+
+		fileWriter.write("\n");
+		fileWriter.close();
+
+		measuredTimes.clear();
+	}
+
+	private static interface IAnalysis {
+
+		public void initialize(int numberOfFilters, int numberOfObjectsToSend) throws Exception;
+
+		public String getName();
+
+		public void execute() throws Exception;
+
+	}
+
+	private static final class TeeTimeAnalysis extends Analysis implements IAnalysis {
+
+		private static final int SECONDS = 1000;
+
+		private Pipeline pipeline;
+		private WorkerThread workerThread;
+
+		public TeeTimeAnalysis() {}
+
+		@Override
+		public void initialize(final int numberOfFilters, final int numberOfObjectsToSend) {
+
+			@SuppressWarnings("unchecked")
+			final NoopFilter<Object>[] noopFilters = new NoopFilter[numberOfFilters];
+			// create stages
+			final teetime.stage.basic.ObjectProducer<Object> objectProducer = new teetime.stage.basic.ObjectProducer<Object>(
+					numberOfObjectsToSend, new Callable<Object>() {
+						@Override
+						public Object call() throws Exception {
+							return new Object();
+						}
+					});
+			for (int i = 0; i < noopFilters.length; i++) {
+				noopFilters[i] = new NoopFilter<Object>();
+			}
+
+			// add each stage to a stage list
+			final List<IStage> startStages = new LinkedList<IStage>();
+			startStages.add(objectProducer);
+
+			final List<IStage> stages = new LinkedList<IStage>();
+			stages.add(objectProducer);
+			stages.addAll(Arrays.asList(noopFilters));
+
+			// connect stages by pipes
+			QueuePipe.connect(objectProducer.outputPort, noopFilters[0].inputPort);
+			for (int i = 1; i < noopFilters.length; i++) {
+				QueuePipe.connect(noopFilters[i - 1].outputPort, noopFilters[i].inputPort);
+			}
+
+			this.pipeline = new Pipeline();
+			this.pipeline.setStartStages(startStages);
+			this.pipeline.setStages(stages);
+
+			this.workerThread = new WorkerThread(this.pipeline, 0);
+			this.workerThread.terminate(StageTerminationPolicy.TERMINATE_STAGE_AFTER_UNSUCCESSFUL_EXECUTION);
+		}
+
+		@Override
+		public String getName() {
+			return "TeeTime";
+		}
+
+		@Override
+		public void execute() {
+			super.start();
+
+			this.workerThread.start();
+			try {
+				this.workerThread.join(60 * SECONDS);
+			} catch (final InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+
+	}
+
+	private static final class KiekerAnalysis implements IAnalysis {
+
+		private IAnalysisController ac;
+
+		public KiekerAnalysis() {}
+
+		@Override
+		public void initialize(final int numberOfFilters, final int numberOfObjectsToSend) throws Exception {
+			this.ac = new AnalysisController();
+
+			final Configuration producerConfig = new Configuration();
+			producerConfig.setProperty(ObjectProducer.CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE, Long.toString(numberOfObjectsToSend));
+			final ObjectProducer<Object> producer = new ObjectProducer<Object>(producerConfig, this.ac, new Callable<Object>() {
+				@Override
+				public Object call() throws Exception {
+					return new Object();
+				}
+			});
+
+			EmptyPassOnFilter predecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+			this.ac.connect(producer, ObjectProducer.OUTPUT_PORT_NAME, predecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+			for (int idx = 0; idx < (numberOfFilters - 1); idx++) {
+				final EmptyPassOnFilter newPredecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+				this.ac.connect(predecessor, EmptyPassOnFilter.OUTPUT_PORT_NAME, newPredecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+				predecessor = newPredecessor;
+			}
+		}
+
+		@Override
+		public String getName() {
+			return "Kieker";
+		}
+
+		@Override
+		public void execute() throws Exception {
+			this.ac.run();
+		}
+
+	}
+
+}

src/main/java/experiment/Experiment2.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package experiment;
+
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import javax.security.auth.login.Configuration;
+
+import kieker.analysis.AnalysisController;
+import kieker.analysis.IAnalysisController;
+import kieker.analysis.stage.CollectorSink;
+import kieker.analysis.stage.EmptyPassOnFilter;
+import kieker.analysis.stage.ObjectProducer;
+import kieker.analysis.stage.StartTimestampFilter;
+import kieker.analysis.stage.StopTimestampFilter;
+import teetime.examples.throughput.TimestampObject;
+import teetime.framework.concurrent.WorkerThread;
+import teetime.framework.core.Analysis;
+import teetime.framework.core.IStage;
+import teetime.framework.core.Pipeline;
+import teetime.framework.sequential.MethodCallPipe;
+import teetime.framework.sequential.QueuePipe;
+import teetime.stage.NoopFilter;
+import teetime.util.StatisticsUtil;
+
+/**
+ * @author Nils Christian Ehmke
+ * 
+ * @since 1.10
+ */
+public class Experiment2 {
+
+	private static final int NUMBER_OF_WARMUP_RUNS_PER_EXPERIMENT = 5;
+	private static final int NUMBER_OF_MEASURED_RUNS_PER_EXPERIMENT = 50;
+
+	private static final int NUMBER_OF_OBJECTS_TO_SEND = 10000;
+
+	private static final int NUMBER_OF_MINIMAL_FILTERS = 50;
+	private static final int NUMBER_OF_MAXIMAL_FILTERS = 1000;
+	private static final int NUMBER_OF_FILTERS_PER_STEP = 50;
+
+	private static final IAnalysis[] analyses = { new TeeTimeAnalysis(true), new TeeTimeAnalysis(false), new KiekerAnalysis() };
+
+	private static final List<Long> measuredTimes = new ArrayList<Long>();
+
+	public static void main(final String[] args) throws Exception {
+		System.setProperty("kieker.common.logging.Log", "NONE");
+
+		for (final IAnalysis analysis : analyses) {
+			for (int numberOfFilters = NUMBER_OF_MINIMAL_FILTERS; numberOfFilters <= NUMBER_OF_MAXIMAL_FILTERS; numberOfFilters += NUMBER_OF_FILTERS_PER_STEP) {
+				// Warmup
+				for (int run = 0; run < NUMBER_OF_WARMUP_RUNS_PER_EXPERIMENT; run++) {
+					analysis.initialize(numberOfFilters, NUMBER_OF_OBJECTS_TO_SEND);
+					analysis.execute();
+				}
+
+				// Actual measurement
+				for (int run = 0; run < NUMBER_OF_MEASURED_RUNS_PER_EXPERIMENT; run++) {
+					final long tin = System.nanoTime();
+
+					analysis.initialize(numberOfFilters, NUMBER_OF_OBJECTS_TO_SEND);
+					analysis.execute();
+
+					final long tout = System.nanoTime();
+					Experiment2.addMeasuredTime((tout - tin));
+				}
+
+				Experiment2.writeAndClearMeasuredTime(analysis.getName(), numberOfFilters);
+			}
+		}
+	}
+
+	private static void addMeasuredTime(final long time) {
+		measuredTimes.add(new Long(time));
+	}
+
+	private static void writeAndClearMeasuredTime(final String analysisName, final int numberOfFilters) throws IOException {
+		final FileWriter fileWriter = new FileWriter(analysisName + ".csv", true);
+		fileWriter.write(Integer.toString(numberOfFilters));
+		fileWriter.write(";");
+
+		final Map<Double, Long> quintiles = StatisticsUtil.calculateQuintiles(measuredTimes);
+		for (final Long value : quintiles.values()) {
+			fileWriter.write(Long.toString(value));
+			fileWriter.write(";");
+		}
+
+		fileWriter.write(Long.toString(StatisticsUtil.calculateAverage(measuredTimes)));
+		fileWriter.write(";");
+
+		fileWriter.write(Long.toString(StatisticsUtil.calculateConfidenceWidth(measuredTimes)));
+
+		fileWriter.write("\n");
+		fileWriter.close();
+
+		measuredTimes.clear();
+	}
+
+	private static interface IAnalysis {
+
+		public String getName();
+
+		public void execute() throws Exception;
+
+		public void initialize(int numberOfFilters, int numberOfObjectsToSend) throws Exception;
+
+	}
+
+	private static final class TeeTimeAnalysis extends Analysis implements IAnalysis {
+
+		private static final int SECONDS = 1000;
+
+		private Pipeline pipeline;
+		private WorkerThread workerThread;
+		private final boolean shouldUseQueue;
+
+		public TeeTimeAnalysis(final boolean shouldUseQueue) {
+			this.shouldUseQueue = shouldUseQueue;
+		}
+
+		@Override
+		public void initialize(final int numberOfFilters, final int numberOfObjectsToSend) throws Exception {
+			@SuppressWarnings("unchecked")
+			final NoopFilter<TimestampObject>[] noopFilters = new NoopFilter[numberOfFilters];
+			// create stages
+			final teetime.stage.basic.ObjectProducer<TimestampObject> objectProducer = new teetime.stage.basic.ObjectProducer<TimestampObject>(
+					numberOfObjectsToSend, new Callable<TimestampObject>() {
+						@Override
+						public TimestampObject call() throws Exception {
+							return new TimestampObject();
+						}
+					});
+			final teetime.stage.StartTimestampFilter startTimestampFilter = new teetime.stage.StartTimestampFilter();
+			for (int i = 0; i < noopFilters.length; i++) {
+				noopFilters[i] = new NoopFilter<TimestampObject>();
+			}
+			final teetime.stage.StopTimestampFilter stopTimestampFilter = new teetime.stage.StopTimestampFilter();
+			final teetime.stage.CollectorSink<TimestampObject> collectorSink = new teetime.stage.CollectorSink<TimestampObject>(
+					this.timestampObjects);
+
+			// add each stage to a stage list
+			final List<IStage> startStages = new LinkedList<IStage>();
+			startStages.add(objectProducer);
+
+			final List<IStage> stages = new LinkedList<IStage>();
+			stages.add(objectProducer);
+			if (this.shouldUseQueue) {
+				stages.add(startTimestampFilter);
+				stages.addAll(Arrays.asList(noopFilters));
+				stages.add(stopTimestampFilter);
+				stages.add(collectorSink);
+
+				// connect stages by pipes
+				QueuePipe.connect(objectProducer.outputPort, startTimestampFilter.inputPort);
+				QueuePipe.connect(startTimestampFilter.outputPort, noopFilters[0].inputPort);
+				for (int i = 1; i < noopFilters.length; i++) {
+					QueuePipe.connect(noopFilters[i - 1].outputPort, noopFilters[i].inputPort);
+				}
+				QueuePipe.connect(noopFilters[noopFilters.length - 1].outputPort, stopTimestampFilter.inputPort);
+				QueuePipe.connect(stopTimestampFilter.outputPort, collectorSink.objectInputPort);
+			} else {
+				// connect stages by pipes
+				MethodCallPipe.connect(objectProducer.outputPort, startTimestampFilter.inputPort);
+				MethodCallPipe.connect(startTimestampFilter.outputPort, noopFilters[0].inputPort);
+				for (int i = 1; i < noopFilters.length; i++) {
+					MethodCallPipe.connect(noopFilters[i - 1].outputPort, noopFilters[i].inputPort);
+				}
+				MethodCallPipe.connect(noopFilters[noopFilters.length - 1].outputPort, stopTimestampFilter.inputPort);
+				MethodCallPipe.connect(stopTimestampFilter.outputPort, collectorSink.objectInputPort);
+			}
+
+			this.pipeline = new Pipeline();
+			this.pipeline.setStartStages(startStages);
+			this.pipeline.setStages(stages);
+		}
+
+		@Override
+		public String getName() {
+			return "TeeTime" + (this.shouldUseQueue ? "-Queues" : "-NoQueues");
+		}
+
+		@Override
+		public void execute() {
+			super.start();
+
+			this.workerThread.start();
+			try {
+				this.workerThread.join(60 * SECONDS);
+			} catch (final InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+
+	}
+
+	private static final class KiekerAnalysis implements IAnalysis {
+
+		private IAnalysisController ac;
+
+		public KiekerAnalysis() {}
+
+		@Override
+		public void initialize(final int numberOfFilters, final int numberOfObjectsToSend) throws Exception {
+			this.ac = new AnalysisController();
+
+			final Configuration producerConfig = new Configuration();
+			producerConfig.setProperty(ObjectProducer.CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE, Long.toString(numberOfObjectsToSend));
+			final ObjectProducer<TimestampObject> producer = new ObjectProducer<TimestampObject>(producerConfig, this.ac, new Callable<TimestampObject>() {
+				@Override
+				public TimestampObject call() throws Exception {
+					return new TimestampObject();
+				}
+			});
+
+			final StartTimestampFilter startTimestampFilter = new StartTimestampFilter(new Configuration(), this.ac);
+			EmptyPassOnFilter predecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+			this.ac.connect(producer, ObjectProducer.OUTPUT_PORT_NAME, startTimestampFilter, StartTimestampFilter.INPUT_PORT_NAME);
+			this.ac.connect(startTimestampFilter, StartTimestampFilter.OUTPUT_PORT_NAME, predecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+			for (int idx = 0; idx < (numberOfFilters - 1); idx++) {
+				final EmptyPassOnFilter newPredecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+				this.ac.connect(predecessor, EmptyPassOnFilter.OUTPUT_PORT_NAME, newPredecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+				predecessor = newPredecessor;
+			}
+			final StopTimestampFilter stopTimestampFilter = new StopTimestampFilter(new Configuration(), this.ac);
+			final CollectorSink<TimestampObject> collectorSink = new CollectorSink<TimestampObject>(new Configuration(), this.ac, this.timestampObjects);
+
+			this.ac.connect(predecessor, EmptyPassOnFilter.OUTPUT_PORT_NAME, stopTimestampFilter, StopTimestampFilter.INPUT_PORT_NAME);
+			this.ac.connect(stopTimestampFilter, StopTimestampFilter.OUTPUT_PORT_NAME, collectorSink, CollectorSink.INPUT_PORT_NAME);
+		}
+
+		@Override
+		public String getName() {
+			return "Kieker";
+		}
+
+		@Override
+		public void execute() throws Exception {
+			this.ac.run();
+		}
+
+	}
+
+}

src/main/java/kieker/analysis/examples/ThroughputAnalysis.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.examples;
+
+import java.util.concurrent.Callable;
+
+import kieker.analysis.AnalysisController;
+import kieker.analysis.IAnalysisController;
+import kieker.analysis.exception.AnalysisConfigurationException;
+import kieker.analysis.stage.EmptyPassOnFilter;
+import kieker.analysis.stage.ObjectProducer;
+import kieker.common.configuration.Configuration;
+
+/**
+ * @author Nils Christian Ehmke
+ * 
+ * @since 1.10
+ */
+public class ThroughputAnalysis<T> {
+
+	private final IAnalysisController ac = new AnalysisController();
+	private int numNoopFilters;
+	private int numInputObjects;
+	private Callable<T> inputObjectCreator;
+
+	public void setNumNoopFilters(final int numNoopFilters) {
+		this.numNoopFilters = numNoopFilters;
+	}
+
+	public void setInput(final int numInputObjects, final Callable<T> inputObjectCreator) {
+		this.numInputObjects = numInputObjects;
+		this.inputObjectCreator = inputObjectCreator;
+	}
+
+	public void start() throws IllegalStateException, AnalysisConfigurationException {
+		this.ac.run();
+	}
+
+	public void init() throws IllegalStateException, AnalysisConfigurationException {
+		final Configuration producerConfig = new Configuration();
+		producerConfig.setProperty(ObjectProducer.CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE, Long.toString(this.numInputObjects));
+		final ObjectProducer<T> producer = new ObjectProducer<T>(producerConfig, this.ac, this.inputObjectCreator);
+
+		EmptyPassOnFilter predecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+		this.ac.connect(producer, ObjectProducer.OUTPUT_PORT_NAME, predecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+		for (int idx = 0; idx < (this.numNoopFilters - 1); idx++) {
+			final EmptyPassOnFilter newPredecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+			this.ac.connect(predecessor, EmptyPassOnFilter.OUTPUT_PORT_NAME, newPredecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+			predecessor = newPredecessor;
+		}
+	}
+
+}

src/main/java/kieker/analysis/examples/ThroughputTimestampAnalysis.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.examples;
+
+import java.util.Collection;
+import java.util.concurrent.Callable;
+
+import kieker.analysis.AnalysisController;
+import kieker.analysis.IAnalysisController;
+import kieker.analysis.exception.AnalysisConfigurationException;
+import kieker.analysis.stage.CollectorSink;
+import kieker.analysis.stage.EmptyPassOnFilter;
+import kieker.analysis.stage.ObjectProducer;
+import kieker.analysis.stage.StartTimestampFilter;
+import kieker.analysis.stage.StopTimestampFilter;
+import kieker.common.configuration.Configuration;
+import teetime.examples.throughput.TimestampObject;
+
+/**
+ * @author Nils Christian Ehmke
+ * 
+ * @since 1.10
+ */
+public class ThroughputTimestampAnalysis {
+
+	private final IAnalysisController ac = new AnalysisController();
+	private int numNoopFilters;
+	private int numInputObjects;
+	private Callable<TimestampObject> inputObjectCreator;
+	private Collection<TimestampObject> timestampObjects;
+
+	public void setNumNoopFilters(final int numNoopFilters) {
+		this.numNoopFilters = numNoopFilters;
+	}
+
+	public void setInput(final int numInputObjects, final Callable<TimestampObject> inputObjectCreator) {
+		this.numInputObjects = numInputObjects;
+		this.inputObjectCreator = inputObjectCreator;
+	}
+
+	public void setTimestampObjects(final Collection<TimestampObject> timestampObjects) {
+		this.timestampObjects = timestampObjects;
+	}
+
+	public void start() throws IllegalStateException, AnalysisConfigurationException {
+		this.ac.run();
+	}
+
+	public void init() throws IllegalStateException, AnalysisConfigurationException {
+		final Configuration producerConfig = new Configuration();
+		producerConfig.setProperty(ObjectProducer.CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE, Long.toString(this.numInputObjects));
+		final ObjectProducer<TimestampObject> producer = new ObjectProducer<TimestampObject>(producerConfig, this.ac, this.inputObjectCreator);
+
+		final StartTimestampFilter startTimestampFilter = new StartTimestampFilter(new Configuration(), this.ac);
+		EmptyPassOnFilter predecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+		this.ac.connect(producer, ObjectProducer.OUTPUT_PORT_NAME, startTimestampFilter, StartTimestampFilter.INPUT_PORT_NAME);
+		this.ac.connect(startTimestampFilter, StartTimestampFilter.OUTPUT_PORT_NAME, predecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+		for (int idx = 0; idx < (this.numNoopFilters - 1); idx++) {
+			final EmptyPassOnFilter newPredecessor = new EmptyPassOnFilter(new Configuration(), this.ac);
+			this.ac.connect(predecessor, EmptyPassOnFilter.OUTPUT_PORT_NAME, newPredecessor, EmptyPassOnFilter.INPUT_PORT_NAME);
+			predecessor = newPredecessor;
+		}
+		final StopTimestampFilter stopTimestampFilter = new StopTimestampFilter(new Configuration(), this.ac);
+		final CollectorSink<TimestampObject> collectorSink = new CollectorSink<TimestampObject>(new Configuration(), this.ac, this.timestampObjects);
+
+		this.ac.connect(predecessor, EmptyPassOnFilter.OUTPUT_PORT_NAME, stopTimestampFilter, StopTimestampFilter.INPUT_PORT_NAME);
+		this.ac.connect(stopTimestampFilter, StopTimestampFilter.OUTPUT_PORT_NAME, collectorSink, CollectorSink.INPUT_PORT_NAME);
+	}
+
+}

src/main/java/kieker/analysis/stage/CacheFilter.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.stage;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import kieker.analysis.IProjectContext;
+import kieker.analysis.plugin.annotation.InputPort;
+import kieker.analysis.plugin.annotation.OutputPort;
+import kieker.analysis.plugin.annotation.Plugin;
+import kieker.analysis.plugin.filter.AbstractFilterPlugin;
+import kieker.common.configuration.Configuration;
+
+/**
+ * @author Nils Christian Ehmke
+ * 
+ * @since 1.10
+ */
+@Plugin(outputPorts = @OutputPort(name = CacheFilter.OUTPUT_PORT_NAME))
+public class CacheFilter extends AbstractFilterPlugin {
+
+	public static final String INPUT_PORT_NAME = "input";
+	public static final String OUTPUT_PORT_NAME = "output";
+
+	private final List<Object> cache = new ArrayList<Object>();
+
+	public CacheFilter(final Configuration configuration,
+			final IProjectContext projectContext) {
+		super(configuration, projectContext);
+	}
+
+	@Override
+	public void terminate(final boolean error) {
+		for (final Object data : this.cache) {
+			super.deliver(EmptyPassOnFilter.OUTPUT_PORT_NAME, data);
+		}
+		super.terminate(error);
+	}
+
+	@Override
+	public Configuration getCurrentConfiguration() {
+		return new Configuration();
+	}
+
+	@InputPort(name = CacheFilter.INPUT_PORT_NAME)
+	public void input(final Object data) {
+		this.cache.add(data);
+	}
+
+}

src/main/java/kieker/analysis/stage/CollectorSink.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.stage;
+
+import java.util.Collection;
+
+import kieker.analysis.IProjectContext;
+import kieker.analysis.plugin.annotation.InputPort;
+import kieker.analysis.plugin.filter.AbstractFilterPlugin;
+import kieker.common.configuration.Configuration;
+
+public class CollectorSink<T> extends AbstractFilterPlugin {
+
+	public static final String INPUT_PORT_NAME = "input";
+
+	private static final int THRESHOLD = 10000;
+
+	private Collection<T> objects;
+
+	public CollectorSink(final Configuration configuration, final IProjectContext projectContext, final Collection<T> collection) {
+		super(configuration, projectContext);
+		this.objects = collection;
+	}
+
+	@InputPort(name = CollectorSink.INPUT_PORT_NAME)
+	public void execute(final T object) {
+		this.objects.add(object);
+		if ((this.objects.size() % THRESHOLD) == 0) {
+			System.out.println("size: " + this.objects.size());
+		}
+	}
+
+	public Collection<T> getObjects() {
+		return this.objects;
+	}
+
+	public void setObjects(final Collection<T> objects) {
+		this.objects = objects;
+	}
+
+	@Override
+	public Configuration getCurrentConfiguration() {
+		return new Configuration();
+	}
+
+}

src/main/java/kieker/analysis/stage/EmptyPassOnFilter.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.stage;
+
+import kieker.analysis.IProjectContext;
+import kieker.analysis.plugin.annotation.InputPort;
+import kieker.analysis.plugin.annotation.OutputPort;
+import kieker.analysis.plugin.annotation.Plugin;
+import kieker.analysis.plugin.filter.AbstractFilterPlugin;
+import kieker.common.configuration.Configuration;
+
+
+/**
+ * @author Nils Christian Ehmke
+ * 
+ * @since 1.10
+ */
+@Plugin(outputPorts = @OutputPort(name = EmptyPassOnFilter.OUTPUT_PORT_NAME))
+public class EmptyPassOnFilter extends AbstractFilterPlugin {
+
+	public static final String INPUT_PORT_NAME = "input";
+	public static final String OUTPUT_PORT_NAME = "output";
+
+	public EmptyPassOnFilter(final Configuration configuration, final IProjectContext projectContext) {
+		super(configuration, projectContext);
+	}
+
+	@Override
+	public Configuration getCurrentConfiguration() {
+		return new Configuration();
+	}
+
+	@InputPort(name = EmptyPassOnFilter.INPUT_PORT_NAME)
+	public void input(final Object data) {
+		super.deliver(EmptyPassOnFilter.OUTPUT_PORT_NAME, data);
+	}
+
+}

src/main/java/kieker/analysis/stage/MappingFileParser.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.stage;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.Map;
+
+import kieker.common.logging.Log;
+import kieker.common.util.filesystem.FSUtil;
+import teetime.stage.kieker.className.ClassNameRegistry;
+
+
+/**
+ * @author Christian Wulf
+ * 
+ * @since 1.10
+ */
+public class MappingFileParser {
+
+	protected Log logger;
+
+	private static final Map<String, String> filePrefixRegistry = new HashMap<String, String>();
+
+	static {
+		filePrefixRegistry.put(FSUtil.MAP_FILENAME, FSUtil.FILE_PREFIX);
+		filePrefixRegistry.put(FSUtil.LEGACY_MAP_FILENAME, FSUtil.LEGACY_FILE_PREFIX);
+	}
+
+	public MappingFileParser(final Log logger) {
+		this.logger = logger;
+	}
+
+	public ClassNameRegistry parseFromStream(final InputStream inputStream) {
+		final ClassNameRegistry classNameRegistry = new ClassNameRegistry();
+
+		BufferedReader in = null;
+		try {
+			in = new BufferedReader(new InputStreamReader(inputStream, FSUtil.ENCODING));
+			String line;
+			while ((line = in.readLine()) != null) { // NOPMD (assign)
+				this.parseTextLine(line, classNameRegistry);
+			}
+		} catch (final IOException ex) {
+			this.logger.error("Error reading mapping file", ex);
+		} finally {
+			if (in != null) {
+				try {
+					in.close();
+				} catch (final IOException ex) {
+					this.logger.error("Exception while closing input stream for mapping file", ex);
+				}
+			}
+		}
+
+		return classNameRegistry;
+	}
+
+	private void parseTextLine(final String line, final Map<Integer, String> stringRegistry) {
+		if (line.length() == 0) {
+			return; // ignore empty lines
+		}
+		final int split = line.indexOf('=');
+		if (split == -1) {
+			this.logger.error("Failed to find character '=' in line: {" + line + "}. It must consist of a ID=VALUE pair.");
+			return; // continue on errors
+		}
+		final String key = line.substring(0, split);
+		final String value = FSUtil.decodeNewline(line.substring(split + 1));
+		// the leading $ is optional
+		final Integer id;
+		try {
+			id = Integer.valueOf((key.charAt(0) == '$') ? key.substring(1) : key); // NOCS
+		} catch (final NumberFormatException ex) {
+			this.logger.error("Error reading mapping file, id must be integer", ex);
+			return; // continue on errors
+		}
+		final String prevVal = stringRegistry.put(id, value);
+		if (prevVal != null) {
+			this.logger.error("Found addional entry for id='" + id + "', old value was '" + prevVal + "' new value is '" + value + "'");
+		}
+	}
+
+	/**
+	 * @since 1.10
+	 */
+	public File findMappingFile(final File dirPath) {
+		File mappingFile = new File(dirPath, FSUtil.MAP_FILENAME);
+		if (!mappingFile.exists()) {
+			// No mapping file found. Check whether we find a legacy tpmon.map file!
+			mappingFile = new File(dirPath, FSUtil.LEGACY_MAP_FILENAME);
+			if (mappingFile.exists()) {
+				this.logger.info("Directory '" + dirPath + "' contains no file '" + FSUtil.MAP_FILENAME + "'. Found '" + FSUtil.LEGACY_MAP_FILENAME
+						+ "' ... switching to legacy mode");
+			} else {
+				// no {kieker|tpmon}.map exists. This is valid for very old monitoring logs. Hence, only dump a log.warn
+				this.logger.warn("No mapping file in directory '" + dirPath.getAbsolutePath() + "'");
+				return null;
+			}
+		}
+
+		return mappingFile;
+	}
+
+	/**
+	 * @return <code>null</code> if a file prefix for the given <code>mappingFile</code> is not registered.
+	 * @since 1.10
+	 */
+	public String getFilePrefixFromMappingFile(final File mappingFile) {
+		return MappingFileParser.filePrefixRegistry.get(mappingFile.getName());
+	}
+}

src/main/java/kieker/analysis/stage/ObjectProducer.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.stage;
+
+import java.util.concurrent.Callable;
+
+import kieker.analysis.IProjectContext;
+import kieker.analysis.plugin.annotation.OutputPort;
+import kieker.analysis.plugin.annotation.Plugin;
+import kieker.analysis.plugin.annotation.Property;
+import kieker.analysis.plugin.reader.AbstractReaderPlugin;
+import kieker.common.configuration.Configuration;
+
+
+@Plugin(
+		outputPorts = @OutputPort(name = ObjectProducer.OUTPUT_PORT_NAME),
+		configuration = @Property(name = ObjectProducer.CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE, defaultValue = "100"))
+public class ObjectProducer<T> extends AbstractReaderPlugin {
+
+	public static final String CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE = "numObjectsToCreate";
+	public static final String OUTPUT_PORT_NAME = "output";
+
+	private Callable<T> objectCreator;
+	private long numObjectsToCreate;
+
+	public ObjectProducer(final Configuration configuration, final IProjectContext projectContext, final Callable<T> objectCreator) {
+		super(configuration, projectContext);
+
+		this.numObjectsToCreate = configuration.getLongProperty(CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE);
+		this.objectCreator = objectCreator;
+	}
+
+	public Callable<T> getObjectCreator() {
+		return this.objectCreator;
+	}
+
+	public void setObjectCreator(final Callable<T> objectCreator) {
+		this.objectCreator = objectCreator;
+	}
+
+	public long getNumObjectsToCreate() {
+		return this.numObjectsToCreate;
+	}
+
+	public void setNumObjectsToCreate(final long numObjectsToCreate) {
+		this.numObjectsToCreate = numObjectsToCreate;
+	}
+
+	@Override
+	public boolean read() {
+		for (int idx = 0; idx < this.numObjectsToCreate; idx++) {
+			try {
+				final T newObject = this.objectCreator.call();
+				super.deliver(OUTPUT_PORT_NAME, newObject);
+			} catch (final Exception e) {
+				throw new IllegalStateException();
+			}
+		}
+
+		return true;
+	}
+
+	@Override
+	public void terminate(final boolean error) {
+		// Nothing to do
+	}
+
+	@Override
+	public Configuration getCurrentConfiguration() {
+		final Configuration configuration = new Configuration();
+		configuration.setProperty(CONFIG_PROPERTY_NAME_OBJECTS_TO_CREATE, Long.toString(this.numObjectsToCreate));
+		return configuration;
+	}
+
+}

src/main/java/kieker/analysis/stage/RecordFromBinaryFileCreator.java

+/***************************************************************************
+ * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ***************************************************************************/
+package kieker.analysis.stage;
+
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.IOException;
+
+import kieker.common.exception.MonitoringRecordException;
+import kieker.common.logging.Log;
+import kieker.common.record.AbstractMonitoringRecord;
+import kieker.common.record.IMonitoringRecord;
+import teetime.stage.kieker.className.ClassNameRegistry;
+import teetime.stage.kieker.className.ClassNameRegistryRepository;
+
+/**
+ * @author Christian Wulf
+ * 
+ * @since 1.10
+ */
+public class RecordFromBinaryFileCreator {
+
+	private final Log logger;
+	private final ClassNameRegistryRepository classNameRegistryRepository;
+
+	public RecordFromBinaryFileCreator(final Log logger, final ClassNameRegistryRepository classNameRegistryRepository) {
+		this.logger = logger;
+		this.classNameRegistryRepository = classNameRegistryRepository;
+	}
+
+	public IMonitoringRecord createRecordFromBinaryFile(final File binaryFile, final DataInputStream inputStream) throws IOException, MonitoringRecordException {
+		final ClassNameRegistry classNameRegistry = this.classNameRegistryRepository.get(binaryFile.getParentFile());
+
+		final Integer id;
+		try {
+			id = inputStream.readInt();
+		} catch (final EOFException eof) {
+			return null; // we are finished
+		}
+		final String classname = classNameRegistry.get(id);
+		if (classname == null) {
+			this.logger.error("Missing classname mapping for record type id " + "'" + id + "'");
+			return null; // we can't easily recover on errors
+		}
+
+		final Class<? extends IMonitoringRecord> clazz = AbstractMonitoringRecord.classForName(classname);
+		final Class<?>[] typeArray = AbstractMonitoringRecord.typesForClass(clazz);
+
+		// read record
+		final long loggingTimestamp = inputStream.readLong(); // NOPMD (must be read here!)
+		final Object[] objectArray = new Object[typeArray.length];
+		int idx = -1;
+		for (final Class<?> type : typeArray) {
+			idx++;
+			if (type == String.class) {
+				final Integer strId = inputStream.readInt();
+				final String str = classNameRegistry.get(strId);
+				if (str == null) {
+					this.logger.error("No String mapping found for id " + strId.toString());
+					objectArray[idx] = "";
+				} else {
+					objectArray[idx] = str;
+				}
+			} else if ((type == int.class) || (type == Integer.class)) {
+				objectArray[idx] = inputStream.readInt();
+			} else if ((type == long.class) || (type == Long.class)) {
+				objectArray[idx] = inputStream.readLong();
+			} else if ((type == float.class) || (type == Float.class)) {
+				objectArray[idx] = inputStream.readFloat();
+			} else if ((type == double.class) || (type == Double.class)) {
+				objectArray[idx] = inputStream.readDouble();
+			} else if ((type == byte.class) || (type == Byte.class)) {
+				objectArray[idx] = inputStream.readByte();
+			} else if ((type == short.class) || (type == Short.class)) { // NOPMD (short)
+				objectArray[idx] = inputStream.readShort();
+			} else if ((type == boolean.class) || (type == Boolean.class)) {
+				objectArray[idx] = inputStream.readBoolean();
+			} else {
+				if (inputStream.readByte() != 0) {
+					this.logger.error("Unexpected value for unsupported type: " + clazz.getName());
+					return null; // breaking error (break would not terminate the correct loop)
+				}
+				this.logger.warn("Unsupported type: " + clazz.getName());
+				objectArray[idx] = null;
+			}
+		}
+		final IMonitoringRecord record = AbstractMonitoringRecord.createFromArray(clazz, objectArray);
+		record.setLoggingTimestamp(loggingTimestamp);
+
+		return record;
+	}
+}