first (more or less) complete prototype

src/main/java/teetime/framework/Analysis.java

@@ -9,6 +9,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import teetime.framework.exceptionHandling.DefaultListener;
+import teetime.framework.exceptionHandling.StageExceptionListener;
 import teetime.framework.signal.ValidatingSignal;
 import teetime.framework.validation.AnalysisNotValidException;
 import teetime.util.Pair;
@@ -26,6 +28,8 @@ public class Analysis implements UncaughtExceptionHandler {
 
 	private final AnalysisConfiguration configuration;
 
+	private final StageExceptionListener listener;
+
 	private final List<Thread> consumerThreads = new LinkedList<Thread>();
 	private final List<Thread> finiteProducerThreads = new LinkedList<Thread>();
 	private final List<Thread> infiniteProducerThreads = new LinkedList<Thread>();
@@ -39,11 +43,20 @@ public class Analysis implements UncaughtExceptionHandler {
 	 *            to be used for the analysis
 	 */
 	public Analysis(final AnalysisConfiguration configuration) {
-		this(configuration, false);
+		this(configuration, false, new DefaultListener());
 	}
 
 	public Analysis(final AnalysisConfiguration configuration, final boolean validationEnabled) {
+		this(configuration, validationEnabled, new DefaultListener());
+	}
+
+	public Analysis(final AnalysisConfiguration configuration, final StageExceptionListener listener) {
+		this(configuration, false, listener);
+	}
+
+	public Analysis(final AnalysisConfiguration configuration, final boolean validationEnabled, final StageExceptionListener listener) {
 		this.configuration = configuration;
+		this.listener = listener;
 		if (validationEnabled) {
 			validateStages();
 		}
@@ -70,13 +83,22 @@ public class Analysis implements UncaughtExceptionHandler {
 	public void init() {
 		final List<Stage> threadableStageJobs = this.configuration.getThreadableStageJobs();
 		for (Stage stage : threadableStageJobs) {
+			StageExceptionListener newListener;
+			try {
+				newListener = listener.getClass().newInstance();
+			} catch (InstantiationException e) {
+				throw new IllegalStateException(e);
+			} catch (IllegalAccessException e) {
+				throw new IllegalStateException(e);
+			}
 			switch (stage.getTerminationStrategy()) {
 			case BY_SIGNAL: {
-				RunnableConsumerStage runnable;
+				RunnableConsumerStage runnable = null;
+				newListener.setHeadStage(runnable);
 				if (stage instanceof AbstractConsumerStage<?>) {
-					runnable = new RunnableConsumerStage(stage, ((AbstractConsumerStage<?>) stage).getIdleStrategy()); // FIXME remove this word-around
+					runnable = new RunnableConsumerStage(stage, ((AbstractConsumerStage<?>) stage).getIdleStrategy(), newListener); // FIXME remove this word-around
 				} else {
-					runnable = new RunnableConsumerStage(stage);
+					runnable = new RunnableConsumerStage(stage, newListener);
 				}
 				final Thread thread = new Thread(runnable);
 				stage.setOwningThread(thread);
@@ -84,13 +106,19 @@ public class Analysis implements UncaughtExceptionHandler {
 				break;
 			}
 			case BY_SELF_DECISION: {
-				final Thread thread = new Thread(new RunnableProducerStage(stage));
+				RunnableProducerStage runnable = null;
+				newListener.setHeadStage(runnable);
+				runnable = new RunnableProducerStage(stage, newListener);
+				final Thread thread = new Thread(runnable);
 				stage.setOwningThread(thread);
 				this.finiteProducerThreads.add(thread);
 				break;
 			}
 			case BY_INTERRUPT: {
-				final Thread thread = new Thread(new RunnableProducerStage(stage));
+				RunnableProducerStage runnable = null;
+				newListener.setHeadStage(runnable);
+				runnable = new RunnableProducerStage(stage, newListener);
+				final Thread thread = new Thread(runnable);
 				stage.setOwningThread(thread);
 				this.infiniteProducerThreads.add(thread);
 				break;

src/main/java/teetime/framework/RunnableConsumerStage.java

@@ -2,6 +2,7 @@ package teetime.framework;
 
 import java.util.Arrays;
 
+import teetime.framework.exceptionHandling.StageExceptionListener;
 import teetime.framework.idle.IdleStrategy;
 import teetime.framework.idle.YieldStrategy;
 import teetime.framework.pipe.IPipe;
@@ -17,12 +18,12 @@ final class RunnableConsumerStage extends RunnableStage {
 	 * @param stage
 	 *            to execute within an own thread
 	 */
-	public RunnableConsumerStage(final Stage stage) {
-		this(stage, new YieldStrategy());
+	public RunnableConsumerStage(final Stage stage, final StageExceptionListener exceptionListener) {
+		this(stage, new YieldStrategy(), exceptionListener);
 	}
 
-	public RunnableConsumerStage(final Stage stage, final IdleStrategy idleStrategy) {
-		super(stage);
+	public RunnableConsumerStage(final Stage stage, final IdleStrategy idleStrategy, final StageExceptionListener exceptionListener) {
+		super(stage, exceptionListener);
 		this.idleStrategy = idleStrategy;
 	}
 

src/main/java/teetime/framework/RunnableProducerStage.java

 package teetime.framework;
 
+import teetime.framework.exceptionHandling.StageExceptionListener;
 import teetime.framework.signal.StartingSignal;
 import teetime.framework.signal.TerminatingSignal;
 
 public final class RunnableProducerStage extends RunnableStage {
 
-	public RunnableProducerStage(final Stage stage) {
-		super(stage);
+	public RunnableProducerStage(final Stage stage, final StageExceptionListener listener) {
+		super(stage, listener);
 	}
 
 	@Override

src/main/java/teetime/framework/RunnableStage.java

@@ -3,17 +3,28 @@ package teetime.framework;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import teetime.framework.exceptionHandling.DefaultListener;
 import teetime.framework.exceptionHandling.StageException;
+import teetime.framework.exceptionHandling.StageExceptionListener;
 
-abstract class RunnableStage implements Runnable {
+public abstract class RunnableStage implements Runnable {
 
 	protected final Stage stage;
 	@SuppressWarnings("PMD.LoggerIsNotStaticFinal")
 	protected final Logger logger;
+	private final StageExceptionListener listener;
 
 	public RunnableStage(final Stage stage) {
 		this.stage = stage;
 		this.logger = LoggerFactory.getLogger(stage.getClass());
+		this.listener = new DefaultListener();
+		listener.setHeadStage(this);
+	}
+
+	public RunnableStage(final Stage stage, final StageExceptionListener exceptionListener) {
+		this.stage = stage;
+		this.logger = LoggerFactory.getLogger(stage.getClass());
+		this.listener = exceptionListener;
 	}
 
 	@Override
@@ -27,7 +38,7 @@ abstract class RunnableStage implements Runnable {
 				try {
 					executeStage();
 				} catch (StageException e) {
-					// TODO: handle exception
+					this.listener.onStageException(e, e.getThrowingStage());
 				}
 			} while (!this.stage.shouldBeTerminated());
 
@@ -44,6 +55,11 @@ abstract class RunnableStage implements Runnable {
 		this.logger.debug("Finished runnable stage. (" + this.stage.getId() + ")");
 	}
 
+	public final void abortExecution() {
+		this.stage.terminate();
+		// TODO: flag error and throw exception
+	}
+
 	protected abstract void beforeStageExecution();
 
 	protected abstract void executeStage();

src/main/java/teetime/framework/exceptionHandling/DefaultListener.java

+package teetime.framework.exceptionHandling;
+
+import teetime.framework.Stage;
+
+public class DefaultListener extends StageExceptionListener {
+
+	public DefaultListener() {
+		super();
+		// TODO Auto-generated constructor stub
+	}
+
+	@Override
+	public void onStageException(final Exception e, final Stage throwingStage) {
+		// TODO Auto-generated method stub
+	}
+}

src/main/java/teetime/framework/exceptionHandling/StageExceptionListener.java

 package teetime.framework.exceptionHandling;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import teetime.framework.RunnableStage;
 import teetime.framework.Stage;
 
 /**
  * Represent a minimalistic StageExceptionListener. Listener which extend from this one, must a least implement this functionality.
- *
+ * This abstract class provides a Logger {@link #logger} and a method to terminate the threads execution {@link #terminateExecution()}.
  */
 public abstract class StageExceptionListener {
 
-	private final Thread thread;
+	private RunnableStage runnable;
 
-	public StageExceptionListener(final Thread thread) {
-		this.thread = thread;
+	/**
+	 * The default logger, which can be used by all subclasses
+	 */
+	protected final Logger logger;
+
+	public StageExceptionListener() {
+		this.logger = LoggerFactory.getLogger(this.getClass().getCanonicalName());
 	}
 
 	/**
@@ -25,12 +34,13 @@ public abstract class StageExceptionListener {
 	public abstract void onStageException(Exception e, Stage throwingStage);
 
 	/**
-	 * Retrieves the thread in which the exception occurred.
-	 *
-	 * @return exception throwing thread
+	 * This method can be used to terminate the execution of the thread.
 	 */
-	public Thread getThread() {
-		return thread;
+	protected final void terminateExecution() {
+		this.runnable.abortExecution();
 	}
 
+	public final void setHeadStage(final RunnableStage headStage) {
+		this.runnable = headStage;
+	}
 }

src/performancetest/java/teetime/examples/experiment09pipeimpls/MethodCallThroughputAnalysis9.java

@@ -20,6 +20,7 @@ import java.util.List;
 import teetime.framework.OldHeadPipeline;
 import teetime.framework.RunnableProducerStage;
 import teetime.framework.Stage;
+import teetime.framework.exceptionHandling.DefaultListener;
 import teetime.framework.pipe.IPipeFactory;
 import teetime.stage.CollectorSink;
 import teetime.stage.NoopFilter;
@@ -44,7 +45,7 @@ public class MethodCallThroughputAnalysis9 {
 
 	public void init(final IPipeFactory pipeFactory) {
 		Stage pipeline = this.buildPipeline(pipeFactory);
-		this.runnable = new RunnableProducerStage(pipeline);
+		this.runnable = new RunnableProducerStage(pipeline, new DefaultListener());
 	}
 
 	/**

src/performancetest/java/teetime/examples/experiment11/MethodCallThroughputAnalysis11.java

@@ -17,9 +17,10 @@ package teetime.examples.experiment11;
 
 import java.util.List;
 
-import teetime.framework.Stage;
 import teetime.framework.OldHeadPipeline;
 import teetime.framework.RunnableProducerStage;
+import teetime.framework.Stage;
+import teetime.framework.exceptionHandling.DefaultListener;
 import teetime.framework.pipe.UnorderedGrowablePipe;
 import teetime.stage.CollectorSink;
 import teetime.stage.NoopFilter;
@@ -44,7 +45,7 @@ public class MethodCallThroughputAnalysis11 {
 
 	public void init() {
 		Stage pipeline = this.buildPipeline(this.numInputObjects, this.inputObjectCreator);
-		this.runnable = new RunnableProducerStage(pipeline);
+		this.runnable = new RunnableProducerStage(pipeline, new DefaultListener());
 	}
 
 	private OldHeadPipeline<ObjectProducer<TimestampObject>, CollectorSink<TimestampObject>> buildPipeline(final long numInputObjects,

src/performancetest/java/teetime/examples/experiment15/MethodCallThroughputAnalysis15.java

@@ -21,6 +21,7 @@ import teetime.framework.AnalysisConfiguration;
 import teetime.framework.OldHeadPipeline;
 import teetime.framework.RunnableProducerStage;
 import teetime.framework.Stage;
+import teetime.framework.exceptionHandling.DefaultListener;
 import teetime.framework.pipe.IPipeFactory;
 import teetime.framework.pipe.OrderedGrowableArrayPipe;
 import teetime.framework.pipe.PipeFactoryRegistry.PipeOrdering;
@@ -64,10 +65,10 @@ public class MethodCallThroughputAnalysis15 extends AnalysisConfiguration {
 
 	public void init() {
 		OldHeadPipeline<Clock, Sink<Long>> clockPipeline = this.buildClockPipeline();
-		this.clockRunnable = new RunnableProducerStage(clockPipeline);
+		this.clockRunnable = new RunnableProducerStage(clockPipeline, new DefaultListener());
 
 		Stage pipeline = this.buildPipeline(this.clock);
-		this.runnable = new RunnableProducerStage(pipeline);
+		this.runnable = new RunnableProducerStage(pipeline, new DefaultListener());
 	}
 
 	private OldHeadPipeline<Clock, Sink<Long>> buildClockPipeline() {