Merge remote-tracking branch 'remotes/origin/multipleInstanceOfFilter'

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/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));
+	}
+
+}