removed InitializingSignal, added abort method in RuntimeServiceFacade

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

@@ -24,7 +24,6 @@ import org.jctools.queues.spec.Ordering;
 import org.jctools.queues.spec.Preference;
 
 import teetime.framework.signal.ISignal;
-import teetime.framework.signal.InitializingSignal;
 import teetime.framework.signal.StartingSignal;
 import teetime.util.framework.concurrent.queue.PCBlockingQueue;
 import teetime.util.framework.concurrent.queue.putstrategy.PutStrategy;
@@ -65,15 +64,6 @@ public abstract class AbstractInterThreadPipe<T> extends AbstractPipe<T> {
 		// do nothing
 	}
 
-	@Override
-	public final void waitForInitializingSignal() throws InterruptedException {
-		final ISignal signal = signalQueue.take();
-		if (!(signal instanceof InitializingSignal)) {
-			throw new IllegalStateException("Expected InitializingSignal, but was " + signal.getClass().getSimpleName());
-		}
-		cachedTargetStage.onSignal(signal, getTargetPort());
-	}
-
 	@Override
 	public final void waitForStartSignal() throws InterruptedException {
 		final ISignal signal = signalQueue.take();

src/main/java/teetime/framework/AbstractIntraThreadPipe.java

@@ -57,9 +57,4 @@ public abstract class AbstractIntraThreadPipe<T> extends AbstractPipe<T> {
 		// do nothing
 	}
 
-	@SuppressWarnings("PMD.EmptyMethodInAbstractClassShouldBeAbstract")
-	@Override
-	public void waitForInitializingSignal() throws InterruptedException {
-		// do nothing
-	}
 }

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

@@ -85,11 +85,6 @@ public abstract class AbstractStage extends Stage {
 		return signalAlreadyReceived;
 	}
 
-	@Override
-	public void onInitializing() throws Exception {
-		changeState(StageState.INITIALIZED);
-	}
-
 	private void changeState(final StageState newState) {
 		currentState = newState;
 		logger.trace(newState.toString());

src/main/java/teetime/framework/InputPort.java

@@ -47,8 +47,4 @@ public class InputPort<T> extends AbstractPort<T> {
 		pipe.waitForStartSignal();
 	}
 
-	public void waitForInitializingSignal() throws InterruptedException {
-		pipe.waitForInitializingSignal();
-	};
-
 }

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

@@ -32,10 +32,6 @@ final class RunnableConsumerStage extends AbstractRunnableStage {
 
 	@Override
 	protected void beforeStageExecution() throws InterruptedException {
-		logger.trace("waitForInitializingSignal");
-		for (InputPort<?> inputPort : stage.getInputPorts()) {
-			inputPort.waitForInitializingSignal();
-		}
 		logger.trace("waitForStartingSignal");
 		for (InputPort<?> inputPort : stage.getInputPorts()) {
 			inputPort.waitForStartSignal();

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

@@ -17,14 +17,12 @@ package teetime.framework;
 
 import java.util.concurrent.Semaphore;
 
-import teetime.framework.signal.InitializingSignal;
 import teetime.framework.signal.StartingSignal;
 import teetime.framework.signal.TerminatingSignal;
 
 public final class RunnableProducerStage extends AbstractRunnableStage {
 
 	private final Semaphore startSemaphore = new Semaphore(0);
-	private final Semaphore initSemaphore = new Semaphore(0);
 
 	RunnableProducerStage(final Stage stage) {
 		super(stage);
@@ -32,8 +30,6 @@ public final class RunnableProducerStage extends AbstractRunnableStage {
 
 	@Override
 	protected void beforeStageExecution() throws InterruptedException {
-		waitForInitializingSignal();
-		this.stage.onSignal(new InitializingSignal(), null);
 		waitForStartingSignal();
 		this.stage.onSignal(new StartingSignal(), null);
 	}
@@ -49,19 +45,10 @@ public final class RunnableProducerStage extends AbstractRunnableStage {
 		this.stage.onSignal(terminatingSignal, null);
 	}
 
-	public void triggerInitializingSignal() {
-		initSemaphore.release();
-	}
-
 	public void triggerStartingSignal() {
 		startSemaphore.release();
 	}
 
-	private void waitForInitializingSignal() throws InterruptedException {
-		logger.trace("waitForInitializingSignal");
-		initSemaphore.acquire();
-	}
-
 	private void waitForStartingSignal() throws InterruptedException {
 		logger.trace("waitForStartingSignal");
 		startSemaphore.acquire();

src/main/java/teetime/framework/RuntimeServiceFacade.java

@@ -26,4 +26,8 @@ public final class RuntimeServiceFacade {
 	public void startWithinNewThread(final Stage previousStage, final Stage stage) {
 		previousStage.getOwningContext().getThreadService().startStageAtRuntime(stage);
 	}
+
+	public void abortExecution(final Stage stage) {
+		stage.getOwningContext().abortConfigurationRun();
+	}
 }

src/main/java/teetime/framework/Stage.java

@@ -165,9 +165,6 @@ public abstract class Stage {
 	 * @throws Exception
 	 *             an arbitrary exception if an error occurs during the initialization
 	 */
-	@SuppressWarnings("PMD.SignatureDeclareThrowsException")
-	public abstract void onInitializing() throws Exception;
-
 	@SuppressWarnings("PMD.SignatureDeclareThrowsException")
 	public abstract void onStarting() throws Exception;
 

src/main/java/teetime/framework/TeeTimeThread.java

@@ -24,12 +24,6 @@ public class TeeTimeThread extends Thread {
 		this.runnable = runnable;
 	}
 
-	public void sendInitializingSignal() {
-		if (runnable instanceof RunnableProducerStage) {
-			((RunnableProducerStage) runnable).triggerInitializingSignal();
-		}
-	}
-
 	public void sendStartingSignal() {
 		if (runnable instanceof RunnableProducerStage) {
 			((RunnableProducerStage) runnable).triggerStartingSignal();

src/main/java/teetime/framework/ThreadService.java

@@ -58,7 +58,6 @@ class ThreadService extends AbstractService<ThreadService> {
 
 		Set<Stage> newThreadableStages = initialize(startStage);
 		startThreads(newThreadableStages);
-		sendInitializingSignal(newThreadableStages);
 	}
 
 	void startStageAtRuntime(final Stage newStage) {
@@ -66,7 +65,6 @@ class ThreadService extends AbstractService<ThreadService> {
 
 		Set<Stage> newThreadableStages = initialize(newStage);
 		startThreads(newThreadableStages);
-		sendInitializingSignal(newThreadableStages);
 
 		sendStartingSignal(newThreadableStages);
 	}
@@ -130,12 +128,6 @@ class ThreadService extends AbstractService<ThreadService> {
 		}
 	}
 
-	private void sendInitializingSignal(final Set<Stage> threadableStages) {
-		for (Stage stage : threadableStages) {
-			((TeeTimeThread) stage.getOwningThread()).sendInitializingSignal();
-		}
-	}
-
 	private void sendStartingSignal(final Set<Stage> newThreadableStages) {
 		for (Stage stage : newThreadableStages) {
 			((TeeTimeThread) stage.getOwningThread()).sendStartingSignal();

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

@@ -91,11 +91,6 @@ public final class DummyPipe implements IPipe<Object> {
 
 	}
 
-	@Override
-	public void waitForInitializingSignal() throws InterruptedException {
-
-	}
-
 	@Override
 	public void close() {
 

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

@@ -97,8 +97,6 @@ public interface IPipe<T> {
 
 	void waitForStartSignal() throws InterruptedException;
 
-	void waitForInitializingSignal() throws InterruptedException;
-
 	void close();
 
 }

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

@@ -101,11 +101,6 @@ public class InstantiationPipe<T> implements IPipe<T> {
 		throw new IllegalStateException(ERROR_MESSAGE);
 	}
 
-	@Override
-	public void waitForInitializingSignal() throws InterruptedException {
-		throw new IllegalStateException(ERROR_MESSAGE);
-	}
-
 	@Override
 	public void close() {
 		throw new IllegalStateException(ERROR_MESSAGE);

src/main/java/teetime/framework/signal/InitializingSignal.java

-/**
- * Copyright (C) 2015 Christian Wulf, Nelson Tavares de Sousa (http://christianwulf.github.io/teetime)
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *         http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package teetime.framework.signal;
-
-import java.util.List;
-import java.util.Set;
-
-import teetime.framework.InputPort;
-import teetime.framework.Stage;
-
-public final class InitializingSignal extends AbstractSignal {
-
-	@Override
-	public void trigger(final Stage stage) {
-		try {
-			stage.onInitializing();
-		} catch (final Exception e) { // NOCS NOPMD (Stages can throw any arbitrary Exception)
-			this.catchedExceptions.add(e);
-			LOGGER.error("Exception while sending the initializing signal", e);
-		}
-	}
-
-	@Override
-	public boolean mayBeTriggered(final Set<InputPort<?>> receivedInputPorts, final List<InputPort<?>> allInputPorts) {
-		return true;
-	}
-
-}

src/main/java/teetime/framework/signal/StartingSignal.java

@@ -19,6 +19,7 @@ import java.util.List;
 import java.util.Set;
 
 import teetime.framework.InputPort;
+import teetime.framework.RuntimeServiceFacade;
 import teetime.framework.Stage;
 
 public final class StartingSignal extends AbstractSignal {
@@ -29,6 +30,7 @@ public final class StartingSignal extends AbstractSignal {
 			stage.onStarting();
 		} catch (final Exception e) { // NOCS NOPMD (Stages can throw any arbitrary Exception)
 			this.catchedExceptions.add(e);
+			RuntimeServiceFacade.INSTANCE.abortExecution(stage);
 			LOGGER.error("Exception while sending the start signal", e);
 		}
 	}

src/main/java/teetime/stage/basic/distributor/dynamic/CreatePortAction.java

@@ -18,11 +18,10 @@ package teetime.stage.basic.distributor.dynamic;
 import java.util.ArrayList;
 import java.util.List;
 
-import teetime.framework.RuntimeServiceFacade;
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
+import teetime.framework.RuntimeServiceFacade;
 import teetime.framework.pipe.SpScPipeFactory;
-import teetime.framework.signal.InitializingSignal;
 import teetime.framework.signal.StartingSignal;
 import teetime.util.framework.port.PortAction;
 
@@ -51,7 +50,6 @@ public class CreatePortAction<T> implements PortAction<DynamicDistributor<T>> {
 
 		RuntimeServiceFacade.INSTANCE.startWithinNewThread(dynamicDistributor, inputPort.getOwningStage());
 
-		newOutputPort.sendSignal(new InitializingSignal());
 		newOutputPort.sendSignal(new StartingSignal());
 
 		// FIXME pass the new thread to the analysis so that it can terminate the thread at the end

src/test/java/teetime/framework/RunnableProducerStageTest.java

@@ -40,12 +40,8 @@ public class RunnableProducerStageTest {
 
 		thread.start();
 
-		// Not running and not initialized
-		assertFalse(testStage.executed && testStage.initialized);
-		runnable.triggerInitializingSignal();
-
 		// Not running, but initialized
-		assertFalse(testStage.executed && !testStage.initialized);
+		assertFalse(testStage.executed);
 		runnable.triggerStartingSignal();
 
 		thread.join();

src/test/java/teetime/framework/RunnableTestStage.java

@@ -25,10 +25,4 @@ class RunnableTestStage extends AbstractProducerStage<Object> {
 		this.terminate();
 	}
 
-	@Override
-	public void onInitializing() throws Exception {
-		super.onInitializing();
-		initialized = true;
-	}
-
 }

src/test/java/teetime/framework/pipe/SpScPipeTest.java

@@ -27,7 +27,6 @@ import teetime.framework.AbstractInterThreadPipe;
 import teetime.framework.InputPort;
 import teetime.framework.OutputPort;
 import teetime.framework.signal.ISignal;
-import teetime.framework.signal.InitializingSignal;
 import teetime.framework.signal.StartingSignal;
 import teetime.framework.signal.TerminatingSignal;
 import teetime.framework.signal.ValidatingSignal;
@@ -47,14 +46,11 @@ public class SpScPipeTest {
 		List<ISignal> signals = new ArrayList<ISignal>();
 		signals.add(new StartingSignal());
 		signals.add(new TerminatingSignal());
-		signals.add(new InitializingSignal());
 		signals.add(new ValidatingSignal());
 		signals.add(new StartingSignal());
 		signals.add(new TerminatingSignal());
-		signals.add(new InitializingSignal());
 		signals.add(new ValidatingSignal());
 		signals.add(new StartingSignal());
-		signals.add(new InitializingSignal());
 		signals.add(new TerminatingSignal());
 		signals.add(new ValidatingSignal());
 

src/test/java/teetime/stage/basic/merger/MergerTestingPipe.java

@@ -107,9 +107,6 @@ class MergerTestingPipe implements IPipe {
 	@Override
 	public void waitForStartSignal() throws InterruptedException {}
 
-	@Override
-	public void waitForInitializingSignal() throws InterruptedException {}
-
 	@Override
 	public void close() {}