added FIXMEs

src/main/java/teetime/framework/AbstractInterThreadPipe.java

@@ -21,15 +21,17 @@ public abstract class AbstractInterThreadPipe extends AbstractPipe {
 	@Override
 	public void sendSignal(final ISignal signal) {
 		this.signalQueue.offer(signal);
-		System.out.println("send signal: " + signal + " to " + cachedTargetStage);
 
 		Thread owningThread = cachedTargetStage.getOwningThread();
-		if (null != owningThread && (owningThread.getState() == State.WAITING || owningThread.getState() == State.TIMED_WAITING)) {
+		if (null != owningThread && isThreadWaiting(owningThread)) { // FIXME remove the null check for performance
 			owningThread.interrupt();
-			System.out.println("interrupted " + owningThread);
 		}
 	}
 
+	protected boolean isThreadWaiting(final Thread thread) {
+		return thread.getState() == State.WAITING || thread.getState() == State.TIMED_WAITING;
+	}
+
 	/**
 	 * Retrieves and removes the head of the signal queue
 	 *

src/main/java/teetime/framework/pipe/SpScPipe.java

 package teetime.framework.pipe;
 
-import java.lang.Thread.State;
 import java.util.Queue;
 
 import org.jctools.queues.QueueFactory;
@@ -41,7 +40,7 @@ public final class SpScPipe extends AbstractInterThreadPipe {
 		}
 
 		Thread owningThread = cachedTargetStage.getOwningThread();
-		if (null != owningThread && (owningThread.getState() == State.WAITING || owningThread.getState() == State.TIMED_WAITING)) {
+		if (null != owningThread && isThreadWaiting(owningThread)) { // FIXME remove the null check for performance
 			synchronized (cachedTargetStage) {
 				cachedTargetStage.notify();
 				// LOGGER.trace("Notified: " + cachedTargetStage);

src/test/java/teetime/framework/RunnableConsumerStageTest.java

@@ -9,8 +9,6 @@ import org.junit.Test;
 
 import teetime.util.Pair;
 
-import com.google.common.base.Joiner;
-
 public class RunnableConsumerStageTest {
 
 	@Test
@@ -22,7 +20,7 @@ public class RunnableConsumerStageTest {
 		Thread thread = new Thread(new Runnable() {
 			@Override
 			public void run() {
-				start(analysis);
+				start(analysis); // FIXME react on exceptions
 			}
 		});
 		thread.start();
@@ -42,7 +40,7 @@ public class RunnableConsumerStageTest {
 		Thread thread = new Thread(new Runnable() {
 			@Override
 			public void run() {
-				start(analysis);
+				start(analysis); // FIXME react on exceptions
 			}
 		});
 		thread.start();
@@ -70,8 +68,9 @@ public class RunnableConsumerStageTest {
 	private void start(final Analysis analysis) {
 		Collection<Pair<Thread, Throwable>> exceptions = analysis.start();
 		for (Pair<Thread, Throwable> pair : exceptions) {
-			System.out.println(pair.getSecond());
-			System.out.println(Joiner.on("\n").join(pair.getSecond().getStackTrace()));
+			// System.out.println(pair.getSecond());
+			// System.out.println(Joiner.on("\n").join(pair.getSecond().getStackTrace()));
+			throw new RuntimeException(pair.getSecond());
 		}
 		assertEquals(0, exceptions.size());
 	}