diff --git a/pom.xml b/pom.xml
index 3fb6ea771d72ca076fbd70a149729e747aeb7a97..7354b67771a257fa131e00fba908704523014a16 100644
--- a/pom.xml
+++ b/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>
diff --git a/src/main/java/teetime/framework/AnalysisConfiguration.java b/src/main/java/teetime/framework/AnalysisConfiguration.java
index c6a0436bc33783ac7d044c817729389d3a61fafe..bb8cbb8b1bbb51a47396920211da5d92aebc0765 100644
--- a/src/main/java/teetime/framework/AnalysisConfiguration.java
+++ b/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;
}
diff --git a/src/main/java/teetime/framework/Stage.java b/src/main/java/teetime/framework/Stage.java
index 99f6c0b8d15139c066de47dd9c287e63dd797da9..f947b21502906d732e8ac7e1dcee3a708d0d3fe6 100644
--- a/src/main/java/teetime/framework/Stage.java
+++ b/src/main/java/teetime/framework/Stage.java
@@ -1,7 +1,8 @@
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);
diff --git a/src/main/java/teetime/stage/MultipleInstanceOfFilter.java b/src/main/java/teetime/stage/MultipleInstanceOfFilter.java
new file mode 100644
index 0000000000000000000000000000000000000000..23aea142b7ca1f37170242cd4ad7c1229b0f459f
--- /dev/null
+++ b/src/main/java/teetime/stage/MultipleInstanceOfFilter.java
@@ -0,0 +1,43 @@
+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);
+ }
+ }
+ }
+}
diff --git a/src/main/resources/logback.groovy b/src/main/resources/logback.groovy
deleted file mode 100644
index 371a869615c1d9ebe0097d7e8f02f5bb597806d2..0000000000000000000000000000000000000000
--- a/src/main/resources/logback.groovy
+++ /dev/null
@@ -1,28 +0,0 @@
-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
diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml
new file mode 100644
index 0000000000000000000000000000000000000000..b17a05f640444952762360a6eb16dddbe52fab56
--- /dev/null
+++ b/src/main/resources/logback.xml
@@ -0,0 +1,34 @@
+<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
diff --git a/src/test/java/teetime/framework/StageTest.java b/src/test/java/teetime/framework/StageTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..b3ac19814975d3cde840d12f2815a829a4c45dec
--- /dev/null
+++ b/src/test/java/teetime/framework/StageTest.java
@@ -0,0 +1,26 @@
+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());
+ }
+ }
+
+}
diff --git a/src/test/java/teetime/stage/MultipleInstanceOfFilterTest.java b/src/test/java/teetime/stage/MultipleInstanceOfFilterTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..4740dea2998c6d47bc5b9947d48c6c516e292be7
--- /dev/null
+++ b/src/test/java/teetime/stage/MultipleInstanceOfFilterTest.java
@@ -0,0 +1,72 @@
+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));
+ }
+
+}
diff --git a/src/main/java/util/test/PerformanceTest.java b/src/test/java/util/test/PerformanceTest.java
similarity index 100%
rename from src/main/java/util/test/PerformanceTest.java
rename to src/test/java/util/test/PerformanceTest.java
diff --git a/src/test/resources/logback.groovy b/src/test/resources/logback.groovy
deleted file mode 100644
index a888fc48590a7920e3c82e1ad9ab1caf1c38bc8c..0000000000000000000000000000000000000000
--- a/src/test/resources/logback.groovy
+++ /dev/null
@@ -1,27 +0,0 @@
-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
diff --git a/src/test/resources/logback.xml b/src/test/resources/logback.xml
new file mode 100644
index 0000000000000000000000000000000000000000..0c6041095f144317b7397cdd45a4bd82c385a8d1
--- /dev/null
+++ b/src/test/resources/logback.xml
@@ -0,0 +1,29 @@
+<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