Merge remote-tracking branch 'origin/master' into uncaught-exceptions

Conflicts:
	src/main/java/teetime/framework/AbstractRunnableStage.java
	src/main/java/teetime/framework/Configuration.java

.checkstyle

@@ -12,4 +12,5 @@
     <file-match-pattern match-pattern="." include-pattern="true"/>
   </fileset>
   <filter name="WriteProtectedFiles" enabled="true"/>
+  <filter name="DerivedFiles" enabled="true"/>
 </fileset-config>

conf/quality-config/pmd-ruleset.xml

@@ -65,6 +65,14 @@
 		<exclude name="AvoidUsingVolatile" />
 	</rule>
 
+	<!-- UR means "undefined reference" which is already detected by the compiler. 
+		so we deactivate it. -->
+	<rule ref="rulesets/java/controversial.xml/DataflowAnomalyAnalysis">
+		<properties>
+			<property name="violationSuppressRegex" value="^Found 'UR'-anomaly.*" />
+		</properties>
+	</rule>
+
 	<rule ref="rulesets/java/coupling.xml">
 		<exclude name="LawOfDemeter" />
 	</rule>

src/main/java/teetime/framework/AbstractPort.java

@@ -19,6 +19,7 @@ import teetime.framework.pipe.IPipe;
 
 public abstract class AbstractPort<T> {
 
+	protected IPipe pipe;
 	/**
 	 * The type of this port.
 	 * <p>
@@ -29,8 +30,6 @@ public abstract class AbstractPort<T> {
 	private final Stage owningStage;
 	private final String name;
 
-	protected IPipe pipe;
-
 	protected AbstractPort(final Class<T> type, final Stage owningStage, final String name) {
 		super();
 		this.type = type;

src/main/java/teetime/framework/AbstractRunnableStage.java

@@ -39,6 +39,7 @@ abstract class AbstractRunnableStage implements Runnable {
 
 	@Override
 	public final void run() {
+		final Stage stage = this.stage; // should prevent the stage to be reloaded after a volatile read
 		this.logger.debug("Executing runnable stage...");
 
 		try {
@@ -67,7 +68,7 @@ abstract class AbstractRunnableStage implements Runnable {
 			}
 		}
 
-		logger.debug("Finished runnable stage. (" + this.stage.getId() + ")");
+		logger.debug("Finished runnable stage. (" + stage.getId() + ")");
 	}
 
 	protected abstract void beforeStageExecution() throws InterruptedException;

src/main/java/teetime/framework/AbstractStage.java

@@ -38,12 +38,12 @@ public abstract class AbstractStage extends Stage {
 
 	@Override
 	protected List<InputPort<?>> getInputPorts() {
-		return inputPorts.getOpenedPorts();
+		return inputPorts.getOpenedPorts(); // TODO consider to publish a read-only version
 	}
 
 	@Override
 	protected List<OutputPort<?>> getOutputPorts() {
-		return outputPorts.getOpenedPorts();
+		return outputPorts.getOpenedPorts(); // TODO consider to publish a read-only version
 	}
 
 	@Override
@@ -74,7 +74,7 @@ public abstract class AbstractStage extends Stage {
 	 * @return <code>true</code> if this stage has already received the given <code>signal</code>, <code>false</code> otherwise
 	 */
 	protected boolean signalAlreadyReceived(final ISignal signal, final InputPort<?> inputPort) {
-		boolean signalAlreadyReceived = this.triggeredSignalTypes.contains(signal.getClass());
+		final boolean signalAlreadyReceived = this.triggeredSignalTypes.contains(signal.getClass());
 		if (signalAlreadyReceived) {
 			if (logger.isTraceEnabled()) {
 				logger.trace("Got signal again: " + signal + " from input port: " + inputPort);

src/main/java/teetime/framework/Configuration.java

@@ -32,6 +32,14 @@ public abstract class Configuration extends AbstractCompositeStage {
 
 	private final IExceptionListenerFactory factory;
 
+	protected Configuration() {
+		this(new TerminatingExceptionListenerFactory());
+	}
+
+	protected Configuration(final IExceptionListenerFactory factory) {
+		this.factory = factory;
+	}
+
 	boolean isExecuted() {
 		return executed;
 	}
@@ -44,11 +52,4 @@ public abstract class Configuration extends AbstractCompositeStage {
 		return factory;
 	}
 
-	protected Configuration() {
-		this(new TerminatingExceptionListenerFactory());
-	}
-
-	protected Configuration(final IExceptionListenerFactory factory) {
-		this.factory = factory;
-	}
 }

src/main/java/teetime/framework/ExecutionInstantiation.java

@@ -66,10 +66,7 @@ class ExecutionInstantiation {
 			final OutputPort outputPort, final InstantiationPipe pipe) {
 		Stage targetStage = pipe.getTargetPort().getOwningStage();
 
-		int targetColor = DEFAULT_COLOR;
-		if (colors.containsKey(targetStage)) {
-			targetColor = colors.get(targetStage);
-		}
+		int targetColor = colors.containsKey(targetStage) ? colors.get(targetStage) : DEFAULT_COLOR;
 
 		if (threadableStages.contains(targetStage) && targetColor != color) {
 			if (pipe.capacity() != 0) {
@@ -94,13 +91,16 @@ class ExecutionInstantiation {
 		int color = DEFAULT_COLOR;
 		Map<Stage, Integer> colors = new HashMap<Stage, Integer>();
 		Set<Stage> threadableStageJobs = configuration.getThreadableStages().keySet();
-		Integer createdConnections = 0;
+		int numCreatedConnections = 0;
 		for (Stage threadableStage : threadableStageJobs) {
 			color++;
 			colors.put(threadableStage, color);
-			createdConnections = colorAndConnectStages(color, colors, threadableStage, configuration);
+			numCreatedConnections += colorAndConnectStages(color, colors, threadableStage, configuration);
+		}
+
+		if (LOGGER.isDebugEnabled()) {
+			LOGGER.debug("Created " + numCreatedConnections + " connections");
 		}
-		LOGGER.debug("Created " + createdConnections + " connections");
 	}
 
 }

src/main/java/teetime/util/framework/concurrent/SignalingCounter.java

@@ -53,8 +53,9 @@ public class SignalingCounter {
 		}
 
 		final Object cond = conditions.get(number);
-		while (counter != number) {
-			synchronized (cond) {
+
+		synchronized (cond) { // counter must be wrapped by synchronized to get the latest value
+			while (counter != number) {
 				cond.wait();
 			}
 		}

src/main/java/teetime/util/framework/concurrent/queue/putstrategy/YieldPutStrategy.java

@@ -17,20 +17,17 @@ package teetime.util.framework.concurrent.queue.putstrategy;
 
 import java.util.Queue;
 
-public class YieldPutStrategy<E> implements PutStrategy<E>
-{
+public class YieldPutStrategy<E> implements PutStrategy<E> {
+
 	@Override
-	public void backoffOffer(final Queue<E> q, final E e)
-	{
-		while (!q.offer(e))
-		{
+	public void backoffOffer(final Queue<E> q, final E e) {
+		while (!q.offer(e)) {
 			Thread.yield();
 		}
 	}
 
 	@Override
-	public void signal()
-	{
+	public void signal() {
 		// Nothing
 	}
 }

src/main/java/teetime/util/framework/concurrent/queue/takestrategy/SCParkTakeStrategy.java

@@ -21,25 +21,22 @@ import java.util.concurrent.locks.LockSupport;
 
 public final class SCParkTakeStrategy<E> implements TakeStrategy<E> {
 
-	public volatile int storeFence = 0;
+	public volatile int storeFence = 0; // NOCS
 
 	private final AtomicReference<Thread> t = new AtomicReference<Thread>(null);
 
 	@Override
 	// Make sure the offer is visible before unpark
-	public void signal()
-	{
+	public void signal() {
 		storeFence = 1; // store barrier
 
 		LockSupport.unpark(t.get()); // t.get() load barrier
 	}
 
 	@Override
-	public E waitPoll(final Queue<E> q) throws InterruptedException
-	{
+	public E waitPoll(final Queue<E> q) throws InterruptedException {
 		E e = q.poll();
-		if (e != null)
-		{
+		if (e != null) {
 			return e;
 		}
 

src/test/java/teetime/stage/io/File2SeqOfWordsTest.java

@@ -30,8 +30,8 @@ public class File2SeqOfWordsTest {
 
 	@Test
 	public void testExecute() throws Exception {
-		File2SeqOfWords stage = new File2SeqOfWords(14);
-		List<String> outputSeqOfWords = new ArrayList<String>();
+		final File2SeqOfWords stage = new File2SeqOfWords(14);
+		final List<String> outputSeqOfWords = new ArrayList<String>();
 		StageTester.test(stage).send(Arrays.asList(new File("./src/test/resources/data/input.txt"))).to(stage.getInputPort()).and().receive(outputSeqOfWords)
 				.from(stage.getOutputPort()).start();
 		assertEquals(outputSeqOfWords.get(0), "Lorem ipsum");

src/test/java/teetime/stage/string/TokenizerTest.java

@@ -39,7 +39,7 @@ public class TokenizerTest {
 
 	@Test
 	public void tokenizerShouldJustDelaySingleToken() {
-		List<String> results = new ArrayList<String>();
+		final List<String> results = new ArrayList<String>();
 
 		test(tokenizer).and().send("Hello World").to(tokenizer.getInputPort()).and().receive(results).from(tokenizer.getOutputPort()).start();
 
@@ -48,7 +48,7 @@ public class TokenizerTest {
 
 	@Test
 	public void tokenizerShouldSplitMultipleToken() {
-		List<String> results = new ArrayList<String>();
+		final List<String> results = new ArrayList<String>();
 
 		test(tokenizer).and().send("Hello;World").to(tokenizer.getInputPort()).and().receive(results).from(tokenizer.getOutputPort()).start();