Merge remote-tracking branch 'origin/master' into site

pom.xml

@@ -87,12 +87,6 @@
 			<version>1.1.2</version>
 			<scope>runtime</scope>
 		</dependency>
-		<dependency>
-			<groupId>org.codehaus.groovy</groupId>
-			<artifactId>groovy-all</artifactId>
-			<version>2.3.6</version>
-			<scope>runtime</scope>
-		</dependency>
 		<dependency>
 			<groupId>com.google.guava</groupId>
 			<artifactId>guava</artifactId>

src/main/java/teetime/framework/AnalysisConfiguration.java

@@ -12,7 +12,7 @@ public class AnalysisConfiguration {
 
 	public AnalysisConfiguration() {}
 
-	List<Stage> getThreadableStageJobs() {
+	List<Stage> getThreadableStageJobs() { // NOPMD only Analysis should use this method
 		return this.threadableStageJobs;
 	}
 

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

 package teetime.framework;
 
 import java.util.List;
-import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -9,7 +10,9 @@ import org.slf4j.LoggerFactory;
 import teetime.framework.signal.ISignal;
 import teetime.framework.validation.InvalidPortConnection;
 
-public abstract class Stage {
+public abstract class Stage { // NOPMD (should not start with "Abstract")
+
+	private static final ConcurrentMap<String, Integer> INSTANCES_COUNTER = new ConcurrentHashMap<String, Integer>();
 
 	private final String id;
 	/**
@@ -18,8 +21,8 @@ public abstract class Stage {
 	protected final Logger logger; // NOPMD
 
 	protected Stage() {
-		this.id = UUID.randomUUID().toString(); // the id should only be represented by a UUID, not additionally by the class name
-		this.logger = LoggerFactory.getLogger(this.getClass().getName() + "(" + this.id + ")");
+		this.id = this.createId();
+		this.logger = LoggerFactory.getLogger(this.id);
 	}
 
 	public String getId() {
@@ -31,6 +34,23 @@ public abstract class Stage {
 		return this.getClass().getName() + ": " + this.getId();
 	}
 
+	private String createId() {
+		String simpleName = this.getClass().getSimpleName();
+
+		Integer numInstances = INSTANCES_COUNTER.get(simpleName);
+		if (null == numInstances) {
+			numInstances = 0;
+		}
+
+		String newId = simpleName + "-" + numInstances;
+		INSTANCES_COUNTER.put(simpleName, ++numInstances);
+		return newId;
+	}
+
+	static void clearInstanceCounters() { // NOPMD (package-private to clear map in tests)
+		INSTANCES_COUNTER.clear();
+	}
+
 	// public abstract Stage getParentStage();
 	//
 	// public abstract void setParentStage(Stage parentStage, int index);

src/main/java/teetime/stage/MultipleInstanceOfFilter.java

+package teetime.stage;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import teetime.framework.AbstractConsumerStage;
+import teetime.framework.OutputPort;
+
+/**
+ * @author Nils Christian Ehmke
+ */
+public final class MultipleInstanceOfFilter<I> extends AbstractConsumerStage<I> {
+
+	private final Map<Class<? extends I>, OutputPort<? super I>> outputPortsMap = new HashMap<Class<? extends I>, OutputPort<? super I>>();
+	private Entry<Class<? extends I>, OutputPort<? super I>>[] cachedOutputPortsMap;
+
+	@SuppressWarnings("unchecked")
+	public <T extends I> OutputPort<T> getOutputPortForType(final Class<T> clazz) {
+		if (!this.outputPortsMap.containsKey(clazz)) {
+			this.outputPortsMap.put(clazz, super.createOutputPort());
+		}
+		return (OutputPort<T>) this.outputPortsMap.get(clazz);
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public void onStarting() throws Exception {
+		super.onStarting();
+
+		// We cache the map to avoid the creating of iterators during runtime
+		cachedOutputPortsMap = (Entry<Class<? extends I>, OutputPort<? super I>>[]) outputPortsMap.entrySet().toArray(new Entry<?, ?>[outputPortsMap.size()]);
+	}
+
+	@Override
+	protected void execute(final I element) {
+		for (Entry<Class<? extends I>, OutputPort<? super I>> outputPortMapEntry : cachedOutputPortsMap) {
+			if (outputPortMapEntry.getKey().isInstance(element)) {
+				outputPortMapEntry.getValue().send(element);
+			}
+		}
+	}
+}

src/main/resources/logback.groovy

-import ch.qos.logback.classic.filter.ThresholdFilter
-
-statusListener(OnConsoleStatusListener)
-
-appender("FILE", FileAppender) {
-  file = "teetime.log"
-  append = false
-  filter(ThresholdFilter) {
-	level = INFO
-  }
-  encoder(PatternLayoutEncoder) {
-    pattern = "%msg%n"
-  }
-}
-
-appender("CONSOLE", ConsoleAppender) {
-  encoder(PatternLayoutEncoder) {
-    pattern = "%d{HH:mm:ss.SSS} %level %logger - %msg%n"
-  }
-}
-
-root WARN, ["CONSOLE"]
-
-//logger "teetime.framework", INFO
-//logger "teetime.stage", INFO
-
-logger "util.TimingsReader", TRACE, ["FILE"]
-logger "util.BucketTimingsReader", TRACE, ["FILE"]
\ No newline at end of file

src/main/resources/logback.xml

+<configuration>
+	<statusListener class="ch.qos.logback.core.status.OnConsoleStatusListener" />
+
+	<appender name="FILE" class="ch.qos.logback.core.FileAppender">
+		<file>teetime.log</file>
+		<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+			<level>INFO</level>
+		</filter>
+		<append>false</append>
+		<!-- encoders are assigned the type ch.qos.logback.classic.encoder.PatternLayoutEncoder 
+			by default -->
+		<encoder>
+			<pattern>%msg%n</pattern>
+		</encoder>
+	</appender>
+
+	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+			<pattern>%d{HH:mm:ss.SSS} %level %logger - %msg%n</pattern>
+		</encoder>
+	</appender>
+
+	<logger name="util.TimingsReader" level="TRACE">
+		<appender-ref ref="FILE" />
+	</logger>
+
+	<logger name="util.BucketTimingsReader" level="TRACE">
+		<appender-ref ref="FILE" />
+	</logger>
+
+	<root level="WARN">
+		<appender-ref ref="CONSOLE" />
+	</root>
+</configuration>
\ No newline at end of file

src/test/java/teetime/framework/StageTest.java

+package teetime.framework;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import teetime.stage.Cache;
+import teetime.stage.Counter;
+
+public class StageTest {
+
+	@Test
+	public void testId() {
+		Stage.clearInstanceCounters();
+
+		Counter<Object> counter0 = new Counter<Object>();
+		Counter<Object> counter1 = new Counter<Object>();
+		Assert.assertEquals("Counter-0", counter0.getId());
+		Assert.assertEquals("Counter-1", counter1.getId());
+
+		for (int i = 0; i < 100; i++) {
+			Cache<Object> cache = new Cache<Object>();
+			Assert.assertEquals("Cache-" + i, cache.getId());
+		}
+	}
+
+}

src/test/java/teetime/stage/MultipleInstanceOfFilterTest.java

+package teetime.stage;
+
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import org.junit.Test;
+
+import teetime.framework.Analysis;
+import teetime.framework.AnalysisConfiguration;
+import teetime.framework.OutputPort;
+import teetime.framework.pipe.IPipeFactory;
+import teetime.framework.pipe.PipeFactoryRegistry.PipeOrdering;
+import teetime.framework.pipe.PipeFactoryRegistry.ThreadCommunication;
+import teetime.util.Pair;
+
+public class MultipleInstanceOfFilterTest {
+
+	private static class TestConfiguration extends AnalysisConfiguration {
+
+		public TestConfiguration(final List<Number> initialInput, final List<Integer> integerList, final List<Float> floatList) {
+			// Create the stages
+			final InitialElementProducer<Number> producer = new InitialElementProducer<Number>(initialInput.toArray(new Number[0]));
+			final MultipleInstanceOfFilter<Number> filter = new MultipleInstanceOfFilter<Number>();
+			final CollectorSink<Integer> integerSink = new CollectorSink<Integer>(integerList);
+			final CollectorSink<Float> floatSink = new CollectorSink<Float>(floatList);
+
+			// Connect the stages
+			final IPipeFactory factory = PIPE_FACTORY_REGISTRY.getPipeFactory(ThreadCommunication.INTRA, PipeOrdering.ARBITRARY, false);
+			factory.create(producer.getOutputPort(), filter.getInputPort());
+			factory.create(filter.getOutputPortForType(Integer.class), integerSink.getInputPort());
+			factory.create(filter.getOutputPortForType(Float.class), floatSink.getInputPort());
+
+			super.addThreadableStage(producer);
+		}
+
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void filteringShouldWork() {
+		final List<Number> initialInput = new ArrayList<Number>(Arrays.asList(1, 1.5f, 2, 2.5f, 3, 3.5f));
+		final List<Integer> integerList = new ArrayList<Integer>();
+		final List<Float> floatList = new ArrayList<Float>();
+
+		final Analysis analysis = new Analysis(new TestConfiguration(initialInput, integerList, floatList));
+		analysis.init();
+		final Collection<Pair<Thread, Throwable>> errors = analysis.start();
+
+		assertThat(errors, is(empty()));
+
+		assertThat(integerList, contains(1, 2, 3));
+		assertThat(floatList, contains(1.5f, 2.5f, 3.5f));
+	}
+
+	@Test
+	public void outputPortForSameTypeShouldBeCached() {
+		final MultipleInstanceOfFilter<Number> filter = new MultipleInstanceOfFilter<Number>();
+
+		final OutputPort<Float> firstPort = filter.getOutputPortForType(Float.class);
+		final OutputPort<Float> secondPort = filter.getOutputPortForType(Float.class);
+
+		assertThat(firstPort, is(secondPort));
+	}
+
+}

src/test/resources/logback.groovy

-import ch.qos.logback.classic.filter.ThresholdFilter
-
-statusListener(OnConsoleStatusListener)
-
-appender("FILE", FileAppender) {
-  file = "src/test/data/load-logs/timings-results.txt"
-  append = false
-  filter(ThresholdFilter) {
-	level = INFO
-  }
-  encoder(PatternLayoutEncoder) {
-    pattern = "%msg%n"
-  }
-}
-
-appender("CONSOLE", ConsoleAppender) {
-  encoder(PatternLayoutEncoder) {
-    pattern = "%d{HH:mm:ss.SSS} %level %logger - %msg%n"
-  }
-}
-
-root ERROR, ["CONSOLE"]
-
-//logger "teetime.framework", INFO
-logger "teetime.stage", INFO
-
-logger "util", INFO
\ No newline at end of file

src/test/resources/logback.xml

+<configuration>
+	<statusListener class="ch.qos.logback.core.status.OnConsoleStatusListener" />
+
+	<appender name="FILE" class="ch.qos.logback.core.FileAppender">
+		<file>src/test/data/load-logs/timings-results.txt</file>
+		<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+			<level>INFO</level>
+		</filter>
+		<append>false</append>
+		<!-- encoders are assigned the type ch.qos.logback.classic.encoder.PatternLayoutEncoder 
+			by default -->
+		<encoder>
+			<pattern>%msg%n</pattern>
+		</encoder>
+	</appender>
+
+	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+			<pattern>%d{HH:mm:ss.SSS} %level %logger - %msg%n</pattern>
+		</encoder>
+	</appender>
+	
+	<logger name="teetime.stage" level="INFO" />
+	<logger name="util" level="INFO" />
+
+	<root level="ERROR">
+		<appender-ref ref="CONSOLE" />
+	</root>
+</configuration>
\ No newline at end of file