first commit

.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src">
+		<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.7">
+		<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
\ No newline at end of file

.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>kieker-stages</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.jdt.core.prefs

+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.annotation.inheritNullAnnotations=disabled
+org.eclipse.jdt.core.compiler.annotation.missingNonNullByDefaultAnnotation=ignore
+org.eclipse.jdt.core.compiler.annotation.nonnull=org.eclipse.jdt.annotation.NonNull
+org.eclipse.jdt.core.compiler.annotation.nonnullbydefault=org.eclipse.jdt.annotation.NonNullByDefault
+org.eclipse.jdt.core.compiler.annotation.nullable=org.eclipse.jdt.annotation.Nullable
+org.eclipse.jdt.core.compiler.annotation.nullanalysis=disabled
+org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
+org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.7
+org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
+org.eclipse.jdt.core.compiler.compliance=1.7
+org.eclipse.jdt.core.compiler.debug.lineNumber=generate
+org.eclipse.jdt.core.compiler.debug.localVariable=generate
+org.eclipse.jdt.core.compiler.debug.sourceFile=generate
+org.eclipse.jdt.core.compiler.problem.annotationSuperInterface=warning
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.autoboxing=ignore
+org.eclipse.jdt.core.compiler.problem.comparingIdentical=warning
+org.eclipse.jdt.core.compiler.problem.deadCode=warning
+org.eclipse.jdt.core.compiler.problem.deprecation=warning
+org.eclipse.jdt.core.compiler.problem.deprecationInDeprecatedCode=disabled
+org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=disabled
+org.eclipse.jdt.core.compiler.problem.discouragedReference=warning
+org.eclipse.jdt.core.compiler.problem.emptyStatement=ignore
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
+org.eclipse.jdt.core.compiler.problem.explicitlyClosedAutoCloseable=ignore
+org.eclipse.jdt.core.compiler.problem.fallthroughCase=ignore
+org.eclipse.jdt.core.compiler.problem.fatalOptionalError=disabled
+org.eclipse.jdt.core.compiler.problem.fieldHiding=ignore
+org.eclipse.jdt.core.compiler.problem.finalParameterBound=warning
+org.eclipse.jdt.core.compiler.problem.finallyBlockNotCompletingNormally=warning
+org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.problem.hiddenCatchBlock=warning
+org.eclipse.jdt.core.compiler.problem.includeNullInfoFromAsserts=disabled
+org.eclipse.jdt.core.compiler.problem.incompatibleNonInheritedInterfaceMethod=warning
+org.eclipse.jdt.core.compiler.problem.incompleteEnumSwitch=warning
+org.eclipse.jdt.core.compiler.problem.indirectStaticAccess=ignore
+org.eclipse.jdt.core.compiler.problem.localVariableHiding=ignore
+org.eclipse.jdt.core.compiler.problem.methodWithConstructorName=warning
+org.eclipse.jdt.core.compiler.problem.missingDefaultCase=ignore
+org.eclipse.jdt.core.compiler.problem.missingDeprecatedAnnotation=ignore
+org.eclipse.jdt.core.compiler.problem.missingEnumCaseDespiteDefault=disabled
+org.eclipse.jdt.core.compiler.problem.missingHashCodeMethod=ignore
+org.eclipse.jdt.core.compiler.problem.missingOverrideAnnotation=ignore
+org.eclipse.jdt.core.compiler.problem.missingOverrideAnnotationForInterfaceMethodImplementation=enabled
+org.eclipse.jdt.core.compiler.problem.missingSerialVersion=warning
+org.eclipse.jdt.core.compiler.problem.missingSynchronizedOnInheritedMethod=ignore
+org.eclipse.jdt.core.compiler.problem.noEffectAssignment=warning
+org.eclipse.jdt.core.compiler.problem.noImplicitStringConversion=warning
+org.eclipse.jdt.core.compiler.problem.nonExternalizedStringLiteral=ignore
+org.eclipse.jdt.core.compiler.problem.nonnullParameterAnnotationDropped=warning
+org.eclipse.jdt.core.compiler.problem.nullAnnotationInferenceConflict=error
+org.eclipse.jdt.core.compiler.problem.nullReference=warning
+org.eclipse.jdt.core.compiler.problem.nullSpecViolation=error
+org.eclipse.jdt.core.compiler.problem.nullUncheckedConversion=warning
+org.eclipse.jdt.core.compiler.problem.overridingPackageDefaultMethod=warning
+org.eclipse.jdt.core.compiler.problem.parameterAssignment=ignore
+org.eclipse.jdt.core.compiler.problem.possibleAccidentalBooleanAssignment=ignore
+org.eclipse.jdt.core.compiler.problem.potentialNullReference=ignore
+org.eclipse.jdt.core.compiler.problem.potentiallyUnclosedCloseable=ignore
+org.eclipse.jdt.core.compiler.problem.rawTypeReference=warning
+org.eclipse.jdt.core.compiler.problem.redundantNullAnnotation=warning
+org.eclipse.jdt.core.compiler.problem.redundantNullCheck=ignore
+org.eclipse.jdt.core.compiler.problem.redundantSpecificationOfTypeArguments=ignore
+org.eclipse.jdt.core.compiler.problem.redundantSuperinterface=ignore
+org.eclipse.jdt.core.compiler.problem.reportMethodCanBePotentiallyStatic=ignore
+org.eclipse.jdt.core.compiler.problem.reportMethodCanBeStatic=ignore
+org.eclipse.jdt.core.compiler.problem.specialParameterHidingField=disabled
+org.eclipse.jdt.core.compiler.problem.staticAccessReceiver=warning
+org.eclipse.jdt.core.compiler.problem.suppressOptionalErrors=disabled
+org.eclipse.jdt.core.compiler.problem.suppressWarnings=enabled
+org.eclipse.jdt.core.compiler.problem.syntacticNullAnalysisForFields=disabled
+org.eclipse.jdt.core.compiler.problem.syntheticAccessEmulation=ignore
+org.eclipse.jdt.core.compiler.problem.typeParameterHiding=warning
+org.eclipse.jdt.core.compiler.problem.unavoidableGenericTypeProblems=enabled
+org.eclipse.jdt.core.compiler.problem.uncheckedTypeOperation=warning
+org.eclipse.jdt.core.compiler.problem.unclosedCloseable=warning
+org.eclipse.jdt.core.compiler.problem.undocumentedEmptyBlock=ignore
+org.eclipse.jdt.core.compiler.problem.unhandledWarningToken=ignore
+org.eclipse.jdt.core.compiler.problem.unnecessaryElse=ignore
+org.eclipse.jdt.core.compiler.problem.unnecessaryTypeCheck=ignore
+org.eclipse.jdt.core.compiler.problem.unqualifiedFieldAccess=ignore
+org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownException=ignore
+org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionExemptExceptionAndThrowable=enabled
+org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionIncludeDocCommentReference=enabled
+org.eclipse.jdt.core.compiler.problem.unusedDeclaredThrownExceptionWhenOverriding=disabled
+org.eclipse.jdt.core.compiler.problem.unusedImport=warning
+org.eclipse.jdt.core.compiler.problem.unusedLabel=warning
+org.eclipse.jdt.core.compiler.problem.unusedLocal=warning
+org.eclipse.jdt.core.compiler.problem.unusedObjectAllocation=ignore
+org.eclipse.jdt.core.compiler.problem.unusedParameter=ignore
+org.eclipse.jdt.core.compiler.problem.unusedParameterIncludeDocCommentReference=enabled
+org.eclipse.jdt.core.compiler.problem.unusedParameterWhenImplementingAbstract=disabled
+org.eclipse.jdt.core.compiler.problem.unusedParameterWhenOverridingConcrete=disabled
+org.eclipse.jdt.core.compiler.problem.unusedPrivateMember=warning
+org.eclipse.jdt.core.compiler.problem.unusedTypeParameter=ignore
+org.eclipse.jdt.core.compiler.problem.unusedWarningToken=warning
+org.eclipse.jdt.core.compiler.problem.varargsArgumentNeedCast=warning
+org.eclipse.jdt.core.compiler.source=1.7

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

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

README.md

+New stages for the Kieker Analysis Component
\ No newline at end of file

pom.xml

+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>net.kieker-monitoring</groupId>
+	<artifactId>kieker-stages</artifactId>
+	<version>1.0-SNAPSHOT</version>
+
+	<licenses>
+		<license>
+			<name>Apache License, Version 2.0</name>
+			<url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+		</license>
+	</licenses>
+	
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<java.version>1.6</java.version>
+	</properties>
+
+
+
+	<repositories>
+		<repository>
+			<!-- for SNAPSHOT versions -->
+			<id>sonatype.oss.snapshots</id>
+			<url>https://oss.sonatype.org/content/repositories/snapshots/</url>
+		</repository>
+	</repositories>
+
+
+
+
+	<dependencies>
+
+		<dependency>
+			<groupId>net.kieker-monitoring</groupId>
+			<artifactId>kieker</artifactId>
+			<version>1.11-SNAPSHOT</version>
+		</dependency>
+		
+		
+	</dependencies>
+
+	<build>
+		<sourceDirectory>src</sourceDirectory>
+		<plugins>
+			<!-- we want JDK 1.6 source and binary compatibility -->
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<version>3.1</version>
+
+				<configuration>
+					<source>${java.version}</source>
+					<target>${java.version}</target>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+
+</project>
\ No newline at end of file

src/kieker/analysis/io/net/AbstractRecordTcpReader.java

+package kieker.analysis.io.net;
+
+import java.nio.ByteBuffer;
+
+import kieker.common.exception.RecordInstantiationException;
+import kieker.common.logging.Log;
+import kieker.common.record.AbstractMonitoringRecord;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.record.factory.CachedRecordFactoryCatalog;
+import kieker.common.record.factory.IRecordFactory;
+import kieker.common.util.registry.ILookup;
+
+public abstract class AbstractRecordTcpReader extends AbstractTcpReader {
+
+	private static final int INT_BYTES = AbstractMonitoringRecord.TYPE_SIZE_INT;
+	private static final int LONG_BYTES = AbstractMonitoringRecord.TYPE_SIZE_LONG;
+
+	private final CachedRecordFactoryCatalog recordFactories = CachedRecordFactoryCatalog.getInstance();
+	// BETTER use a non thread-safe implementation to increase performance. A thread-safe version is not necessary.
+	private final ILookup<String> stringRegistry;
+
+	/**
+	 * Default constructor with <code>port=10133</code> and <code>bufferCapacity=65535</code>
+	 */
+	public AbstractRecordTcpReader(final Log logger, final ILookup<String> stringRegistry) {
+		this(10133, 65535, logger, stringRegistry);
+	}
+
+	/**
+	 *
+	 * @param port
+	 *            accept connections on this port
+	 * @param bufferCapacity
+	 *            capacity of the receiving buffer
+	 */
+	public AbstractRecordTcpReader(final int port, final int bufferCapacity, final Log logger, final ILookup<String> stringRegistry) {
+		super(port, bufferCapacity, logger);
+		this.stringRegistry = stringRegistry;
+	}
+
+	@Override
+	protected final boolean onBufferReceived(final ByteBuffer buffer) {
+		// identify record class
+		if (buffer.remaining() < INT_BYTES) {
+			return false;
+		}
+		final int clazzId = buffer.getInt();
+		final String recordClassName = this.stringRegistry.get(clazzId);
+
+		// identify logging timestamp
+		if (buffer.remaining() < LONG_BYTES) {
+			return false;
+		}
+		final long loggingTimestamp = buffer.getLong();
+
+		// identify record data
+		final IRecordFactory<? extends IMonitoringRecord> recordFactory = this.recordFactories.get(recordClassName);
+		if (buffer.remaining() < recordFactory.getRecordSizeInBytes()) {
+			return false;
+		}
+
+		try {
+			final IMonitoringRecord record = recordFactory.create(buffer, this.stringRegistry);
+			record.setLoggingTimestamp(loggingTimestamp);
+
+			onRecordReceived(record);
+		} catch (final RecordInstantiationException ex) {
+			super.logger.error("Failed to create: " + recordClassName, ex);
+		}
+
+		return true;
+	}
+
+	protected abstract void onRecordReceived(IMonitoringRecord record);
+
+}

src/kieker/analysis/io/net/AbstractTcpReader.java

+package kieker.analysis.io.net;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+import kieker.common.logging.Log;
+
+public abstract class AbstractTcpReader implements Runnable {
+
+	private final int port;
+	private final int bufferCapacity;
+	@SuppressWarnings("PMD.LoggerIsNotStaticFinal")
+	protected final Log logger;
+
+	private volatile boolean terminated;
+
+	public AbstractTcpReader(final int port, final int bufferCapacity, final Log logger) {
+		super();
+		this.port = port;
+		this.bufferCapacity = bufferCapacity;
+		this.logger = logger;
+	}
+
+	@Override
+	public final void run() {
+		ServerSocketChannel serversocket = null;
+		try {
+			serversocket = ServerSocketChannel.open();
+			serversocket.socket().bind(new InetSocketAddress(this.port));
+			if (logger.isDebugEnabled()) {
+				logger.debug("Listening on port " + this.port);
+			}
+
+			final SocketChannel socketChannel = serversocket.accept();
+			try {
+				final ByteBuffer buffer = ByteBuffer.allocateDirect(bufferCapacity);
+				while (socketChannel.read(buffer) != -1 && !terminated) {
+					process(buffer);
+				}
+			} finally {
+				socketChannel.close();
+			}
+		} catch (final IOException ex) {
+			logger.error("Error while reading.", ex);
+		} finally {
+			if (null != serversocket) {
+				try {
+					serversocket.close();
+				} catch (final IOException e) {
+					logger.debug("Failed to close TCP connection.", e);
+				}
+			}
+		}
+	}
+
+	private void process(final ByteBuffer buffer) {
+		buffer.flip();
+		try {
+			while (buffer.hasRemaining()) {
+				buffer.mark();
+				boolean success = this.onBufferReceived(buffer);
+				if (!success) {
+					buffer.reset();
+					buffer.compact();
+					return;
+				}
+			}
+			buffer.clear();
+		} catch (final BufferUnderflowException ex) {
+			logger.warn("Unexpected buffer underflow. Resetting and compacting buffer.", ex);
+			buffer.reset();
+			buffer.compact();
+		}
+	}
+
+	/**
+	 * @param buffer
+	 *            to be read from
+	 * @return <ul>
+	 *         <li><code>true</code> when there were enough bytes to perform the read operation
+	 *         <li><code>false</code> otherwise. In this case, the buffer is reset, compacted, and filled with new content.
+	 */
+	protected abstract boolean onBufferReceived(final ByteBuffer buffer);
+
+	public void terminate() {
+		this.terminated = true;
+	}
+
+	public int getPort() {
+		return port;
+	}
+}

src/kieker/analysis/io/net/NewTcpReader.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.io.net;
+
+import java.nio.ByteBuffer;
+
+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;
+import kieker.common.record.IMonitoringRecord;
+import kieker.common.record.misc.RegistryRecord;
+import kieker.common.util.registry.ILookup;
+import kieker.common.util.registry.Lookup;
+
+/**
+ * This is a reader which reads the records from a TCP port.
+ *
+ * @author Jan Waller, Christian Wulf
+ *
+ * @since 1.8
+ */
+@Plugin(description = "A reader which reads records from a TCP port",
+		outputPorts = {
+			@OutputPort(name = NewTcpReader.OUTPUT_PORT_NAME_RECORDS, eventTypes = { IMonitoringRecord.class }, description = "Output Port of the TCPReader")
+		},
+		configuration = {
+			@Property(name = NewTcpReader.CONFIG_PROPERTY_NAME_PORT1, defaultValue = "10133",
+					description = "The first port of the server used for the TCP connection."),
+			@Property(name = NewTcpReader.CONFIG_PROPERTY_NAME_PORT2, defaultValue = "10134",
+					description = "The second port of the server used for the TCP connection.")
+		})
+public final class NewTcpReader extends AbstractReaderPlugin {
+
+	/** The name of the output port delivering the received records. */
+	public static final String OUTPUT_PORT_NAME_RECORDS = "monitoringRecords";
+
+	/** The name of the configuration determining the TCP port. */
+	public static final String CONFIG_PROPERTY_NAME_PORT1 = "port1";
+	/** The name of the configuration determining the TCP port. */
+	public static final String CONFIG_PROPERTY_NAME_PORT2 = "port2";
+
+	private static final int MESSAGE_BUFFER_SIZE = 65535;
+
+	private final int port1;
+	private final int port2;
+	private final ILookup<String> stringRegistry = new Lookup<String>();
+
+	private AbstractRecordTcpReader tcpMonitoringRecordReader;
+	private AbstractTcpReader tcpStringRecordReader;
+
+	private Thread tcpStringRecordReaderThread;
+
+	public NewTcpReader(final Configuration configuration, final IProjectContext projectContext) {
+		super(configuration, projectContext);
+		this.port1 = this.configuration.getIntProperty(CONFIG_PROPERTY_NAME_PORT1);
+		this.port2 = this.configuration.getIntProperty(CONFIG_PROPERTY_NAME_PORT2);
+
+		this.tcpMonitoringRecordReader = new AbstractRecordTcpReader(port1, MESSAGE_BUFFER_SIZE, LOG, stringRegistry) {
+			@Override
+			protected void onRecordReceived(final IMonitoringRecord record) {
+				NewTcpReader.super.deliver(OUTPUT_PORT_NAME_RECORDS, record);
+			}
+		};
+
+		this.tcpStringRecordReader = new AbstractTcpReader(port2, MESSAGE_BUFFER_SIZE, LOG) {
+			@Override
+			protected boolean onBufferReceived(final ByteBuffer buffer) {
+				RegistryRecord.registerRecordInRegistry(buffer, stringRegistry);
+				return true;
+			}
+		};
+	}
+
+	@Override
+	public boolean init() {
+		this.tcpStringRecordReaderThread = new Thread(tcpStringRecordReader);
+		this.tcpStringRecordReaderThread.start();
+		return super.init();
+	}
+
+	@Override
+	public Configuration getCurrentConfiguration() {
+		final Configuration configuration = new Configuration();
+		configuration.setProperty(CONFIG_PROPERTY_NAME_PORT1, Integer.toString(this.port1));
+		configuration.setProperty(CONFIG_PROPERTY_NAME_PORT2, Integer.toString(this.port2));
+		return configuration;
+	}
+
+	@Override
+	public boolean read() {
+		tcpMonitoringRecordReader.run();
+		return true;
+	}
+
+	@Override
+	public void terminate(final boolean error) {
+		this.log.info("Shutdown requested.");
+		this.tcpMonitoringRecordReader.terminate();
+
+		this.tcpStringRecordReader.terminate();
+		this.tcpStringRecordReaderThread.interrupt();
+	}
+
+	public int getPort1() {
+		return port1;
+	}
+
+	public int getPort2() {
+		return port2;
+	}
+}