Further fixes for tasks recipe

- Relates to #74
- Add missing features for listener
- Tweak rest of a missing tasks features

spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java

@@ -57,7 +57,7 @@ public class RunnableAction implements Action<String, String> {
 
 	@Override
 	public final void execute(StateContext<String, String> context) {
-		if (!shouldExecute(context)) {
+		if (!shouldExecute(id, context)) {
 			return;
 		}
 		StopWatch watch = new StopWatch();
@@ -67,10 +67,13 @@ public class RunnableAction implements Action<String, String> {
 			watch.start();
 		}
 		try {
+			onPreExecute(id, context);
 			runnable.run();
-			onSuccess(context);
+			onSuccess(id, context);
 		} catch (Exception e) {
-			onError(context, e);
+			onError(id, context, e);
+		} finally {
+			onPostExecute(id, context);
 		}
 		if (log.isDebugEnabled()) {
 			watch.stop();
@@ -82,14 +85,20 @@ public class RunnableAction implements Action<String, String> {
 		return id;
 	}
 
-	protected boolean shouldExecute(StateContext<String, String> context) {
+	protected boolean shouldExecute(String id, StateContext<String, String> context) {
 		return true;
 	}
 
-	protected void onSuccess(StateContext<String, String> context) {
+	protected void onPreExecute(String id, StateContext<String, String> context) {
 	}
 
-	protected void onError(StateContext<String, String> context, Exception e) {
+	protected void onPostExecute(String id, StateContext<String, String> context) {
+	}
+
+	protected void onSuccess(String id, StateContext<String, String> context) {
+	}
+
+	protected void onError(String id, StateContext<String, String> context, Exception e) {
 	}
 
 }

spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java

@@ -42,10 +42,11 @@ import org.springframework.statemachine.support.tree.Tree.Node;
 import org.springframework.statemachine.support.tree.TreeTraverser;
 
 /**
- * {@code TasksHandler} is a recipe for executing {@link Runnable} tasks
+ * {@code TasksHandler} is a recipe for executing arbitrary {@link Runnable} tasks
  * using a state machine logic.
  *
- *
+ * This recipe supports execution of multiple top-level tasks with a
+ * sub-states construct of DAGs.
  *
  * @author Janne Valkealahti
  *
@@ -76,26 +77,39 @@ public class TasksHandler {
 
 	/**
 	 * Instantiates a new tasks handler. Intentionally private instantiation
-	 * meant to be called from builder.
+	 * meant to be called from a builder.
 	 *
 	 * @param tasks the wrapped tasks
+	 * @param listener the tasks listener
 	 */
-	private TasksHandler(List<TaskWrapper> tasks) {
+	private TasksHandler(List<TaskWrapper> tasks, TasksListener listener, TaskExecutor taskExecutor) {
 		try {
-			this.stateMachine = buildStateMachine(tasks);
+			this.stateMachine = buildStateMachine(tasks, taskExecutor);
 		} catch (Exception e) {
 			throw new StateMachineException("Error building state machine from tasks", e);
 		}
+		if (listener != null) {
+			addTasksListener(listener);
+		}
 	}
 
+	/**
+	 * Request to execute current tasks logic.
+	 */
 	public void runTasks() {
 		stateMachine.sendEvent(EVENT_RUN);
 	}
 
+	/**
+	 * Request to continue from an error.
+	 */
 	public void continueFromError() {
 		stateMachine.sendEvent(EVENT_CONTINUE);
 	}
 
+	/**
+	 * Request to fix current problems.
+	 */
 	public void fixCurrentProblems() {
 		stateMachine.sendEvent(EVENT_FIX);
 	}
@@ -137,11 +151,14 @@ public class TasksHandler {
 		return new Builder();
 	}
 
-	private StateMachine<String, String> buildStateMachine(List<TaskWrapper> tasks) throws Exception {
+	private StateMachine<String, String> buildStateMachine(List<TaskWrapper> tasks, TaskExecutor taskExecutor)
+			throws Exception {
 		StateMachineBuilder.Builder<String, String> builder = StateMachineBuilder.builder();
 
+		int taskCount = topLevelTaskCount(tasks);
+
 		builder.configureConfiguration().withConfiguration()
-			.taskExecutor(taskExecutor());
+			.taskExecutor(taskExecutor != null ? taskExecutor : taskExecutor(taskCount));
 
 		StateMachineStateConfigurer<String, String> stateMachineStateConfigurer = builder.configureStates();
 		StateMachineTransitionConfigurer<String, String> stateMachineTransitionConfigurer = builder.configureTransitions();
@@ -211,6 +228,7 @@ public class TasksHandler {
 			.withExternal()
 				.source(STATE_ERROR).target(STATE_READY)
 				.event(EVENT_CONTINUE)
+				.action(continueAction())
 				.and()
 			.withExternal()
 				.source(STATE_AUTOMATIC).target(STATE_MANUAL)
@@ -224,13 +242,21 @@ public class TasksHandler {
 		return builder.build();
 	}
 
-	private static TaskExecutor taskExecutor() {
+	private static TaskExecutor taskExecutor(int taskCount) {
 		ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
 		taskExecutor.afterPropertiesSet();
-		taskExecutor.setCorePoolSize(5);
+		taskExecutor.setCorePoolSize(taskCount);
 		return taskExecutor;
 	}
 
+	private static int topLevelTaskCount(List<TaskWrapper> tasks) {
+		Tree<TaskWrapper> tree = new Tree<TaskWrapper>();
+		for (TaskWrapper wrapper : tasks) {
+			tree.add(wrapper, wrapper.id, wrapper.parent);
+		}
+		return tree.getRoot().getChildren().size();
+	}
+
 	private static Iterator<Node<TaskWrapper>> buildTasksIterator(List<TaskWrapper> tasks) {
 		Tree<TaskWrapper> tree = new Tree<TaskWrapper>();
 		for (TaskWrapper wrapper : tasks) {
@@ -256,13 +282,15 @@ public class TasksHandler {
 	public static class Builder {
 
 		private final List<TaskWrapper> tasks = new ArrayList<TaskWrapper>();
+		private TasksListener listener;
+		private TaskExecutor taskExecutor;
 
 		/**
 		 * Define a top-level task.
 		 *
 		 * @param id the id
 		 * @param runnable the runnable
-		 * @return the builder
+		 * @return the builder for chaining
 		 */
 		public Builder task(Object id, Runnable runnable) {
 			tasks.add(new TaskWrapper(null, id, runnable));
@@ -275,7 +303,7 @@ public class TasksHandler {
 		 * @param parent the parent
 		 * @param id the id
 		 * @param runnable the runnable
-		 * @return the builder
+		 * @return the builder for chaining
 		 */
 		public Builder task(Object parent, Object id, Runnable runnable) {
 			tasks.add(new TaskWrapper(parent, id, runnable));
@@ -287,32 +315,75 @@ public class TasksHandler {
 		 * should be persisted with state changes.
 		 *
 		 * @param persist the persist
-		 * @return the builder
+		 * @return the builder for chaining
 		 */
 		public Builder persist(StateMachinePersist<String, String, Void> persist) {
 			return this;
 		}
 
+		/**
+		 * Define a {@link TasksListener} to be registered.
+		 *
+		 * @param listener the tasks listener
+		 * @return the builder for chaining
+		 */
+		public Builder listener(TasksListener listener) {
+			this.listener = listener;
+			return this;
+		}
+
+		/**
+		 * Define a {@link TaskExecutor} to be used. Default executor will be
+		 * a {@link ThreadPoolTaskExecutor} set with a thread pool size of
+		 * a top-level task count.
+		 *
+		 * @param taskExecutor the task executor
+		 * @return the builder for chaining
+		 */
+		public Builder taskExecutor(TaskExecutor taskExecutor) {
+			this.taskExecutor = taskExecutor;
+			return this;
+		}
+
 		/**
 		 * Builds the {@link TasksHandler}.
 		 *
 		 * @return the tasks handler
 		 */
 		public TasksHandler build() {
-			return new TasksHandler(tasks);
+			return new TasksHandler(tasks, listener, taskExecutor);
 		}
 
 	}
 
+	/**
+	 * Gets a tasks entry action.
+	 *
+	 * @return the tasks entry action
+	 */
 	private TasksEntryAction tasksEntryAction() {
 		return new TasksEntryAction();
 	}
 
-	private static LocalRunnableAction runnableAction(Runnable runnable, String id) {
+	/**
+	 * Gets a loca runnable action.
+	 *
+	 * @param runnable the runnable
+	 * @param id the task id
+	 * @return the local runnable action
+	 */
+	private LocalRunnableAction runnableAction(Runnable runnable, String id) {
 		return new LocalRunnableAction(runnable, id);
 	}
 
-	private static Guard<String, String> tasksChoiceGuard() {
+	/**
+	 * Tasks choice guard. This {@link Guard} will check if related
+	 * extended state variables contains negative values for related
+	 * tasks id's and returns true if so, else false.
+	 *
+	 * @return the guard
+	 */
+	private Guard<String, String> tasksChoiceGuard() {
 		return new Guard<String, String>() {
 
 			@Override
@@ -326,16 +397,34 @@ public class TasksHandler {
 								if (log.isDebugEnabled()) {
 									log.debug("Task id=[" + entry.getKey() + "] has negative execution value, tasksChoiceGuard returns true");
 								}
+								listener.onTasksError();
 								return true;
 							}
 						}
 					}
 				}
+				listener.onTasksSuccess();
 				return false;
 			}
 		};
 	}
 
+	/**
+	 * {@link Action} which simply sends an event of continue
+	 * tasks into a state machine.
+	 *
+	 * @return the action
+	 */
+	private Action<String, String> continueAction() {
+		return new Action<String, String>() {
+
+			@Override
+			public void execute(StateContext<String, String> context) {
+				listener.onTasksContinue();
+			}
+		};
+	}
+
 	private Action<String, String> automaticAction() {
 		return new Action<String, String>() {
 
@@ -345,6 +434,12 @@ public class TasksHandler {
 		};
 	}
 
+	/**
+	 * {@link Action} which resets related extended state variables
+	 * to zero for tasks order to indicate a fixed tasks.
+	 *
+	 * @return the action
+	 */
 	private Action<String, String> fixAction() {
 		return new Action<String, String>() {
 
@@ -367,7 +462,9 @@ public class TasksHandler {
 
 	/**
 	 * {@code TasksListener} is a generic interface listening tasks
-	 * execution events.
+	 * execution events. Methods in this interface will be called in a
+	 * tasks execution position where user most likely will want to get
+	 * notified.
 	 */
 	public interface TasksListener {
 
@@ -399,13 +496,20 @@ public class TasksHandler {
 		void onTaskPostExecute(Object id);
 
 		/**
-		 * Called when task execution resulter an error of any kind.
+		 * Called when task execution result an error of any kind.
 		 *
 		 * @param id the task id
 		 * @param exception the exception
 		 */
 		void onTaskFailed(Object id, Exception exception);
 
+		/**
+		 * Called when task execution result without errors.
+		 *
+		 * @param id the task id
+		 */
+		void onTaskSuccess(Object id);
+
 		/**
 		 * Called when all tasks has been executed successfully.
 		 */
@@ -456,6 +560,13 @@ public class TasksHandler {
 			}
 		}
 
+		@Override
+		public void onTaskSuccess(Object id) {
+			for (Iterator<TasksListener> iterator = getListeners().reverse(); iterator.hasNext();) {
+				iterator.next().onTaskSuccess(id);
+			}
+		}
+
 		@Override
 		public void onTasksSuccess() {
 			for (Iterator<TasksListener> iterator = getListeners().reverse(); iterator.hasNext();) {
@@ -504,24 +615,36 @@ public class TasksHandler {
 	/**
 	 * {@link Action} which is execution with every registered {@link Runnable}.
 	 */
-	private static class LocalRunnableAction extends RunnableAction {
+	private class LocalRunnableAction extends RunnableAction {
 
 		public LocalRunnableAction(Runnable runnable, String id) {
 			super(runnable, id);
 		}
 
 		@Override
-		protected boolean shouldExecute(StateContext<String, String> context) {
-			return super.shouldExecute(context);
+		protected boolean shouldExecute(String id, StateContext<String, String> context) {
+			return super.shouldExecute(id, context);
 		}
 
 		@Override
-		protected void onSuccess(StateContext<String, String> context) {
+		protected void onPreExecute(String id, StateContext<String, String> context) {
+			listener.onTaskPreExecute(id);
+		}
+
+		@Override
+		protected void onPostExecute(String id, StateContext<String, String> context) {
+			listener.onTaskPostExecute(id);
+		}
+
+		@Override
+		protected void onSuccess(String id, StateContext<String, String> context) {
+			listener.onTaskSuccess(id);
 			changeCount(1, context);
 		}
 
 		@Override
-		protected void onError(StateContext<String, String> context, Exception e) {
+		protected void onError(String id, StateContext<String, String> context, Exception e) {
+			listener.onTaskFailed(id, e);
 			changeCount(-1, context);
 		}
 

spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/TasksHandlerTests.java

@@ -60,7 +60,7 @@ public class TasksHandlerTests {
 	}
 
 	@Test
-	public void testRunFailAndContinue() throws InterruptedException {
+	public void testRunFail() throws InterruptedException {
 		TasksHandler handler = TasksHandler.builder()
 				.task("1", sleepRunnable())
 				.task("2", sleepRunnable())
@@ -83,6 +83,35 @@ public class TasksHandlerTests {
 		assertThat(variables.size(), is(3));
 	}
 
+	@Test
+	public void testRunFailAndContinue() throws InterruptedException {
+		TasksHandler handler = TasksHandler.builder()
+				.task("1", sleepRunnable())
+				.task("2", sleepRunnable())
+				.task("3", failRunnable())
+				.build();
+
+		TestListener listener = new TestListener();
+		listener.reset(11, 0, 0);
+		StateMachine<String, String> machine = handler.getStateMachine();
+		machine.addStateListener(listener);
+		machine.start();
+		assertThat(listener.stateMachineStartedLatch.await(1, TimeUnit.SECONDS), is(true));
+
+		handler.runTasks();
+
+		assertThat(listener.stateChangedLatch.await(8, TimeUnit.SECONDS), is(true));
+		assertThat(listener.stateChangedCount, is(11));
+		assertThat(machine.getState().getIds(), contains(TasksHandler.STATE_ERROR, TasksHandler.STATE_AUTOMATIC));
+
+		handler.fixCurrentProblems();
+		handler.continueFromError();
+		listener.reset(1, 0, 0);
+		assertThat(listener.stateChangedLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(listener.stateChangedCount, is(1));
+		assertThat(machine.getState().getIds(), contains(TasksHandler.STATE_READY));
+	}
+
 	@Test
 	public void testDagSingleRoot() throws InterruptedException {
 		TasksHandler handler = TasksHandler.builder()
@@ -139,17 +168,15 @@ public class TasksHandlerTests {
 	}
 
 	@Test
-	public void testEvents() throws InterruptedException {
+	public void testEvents1() throws InterruptedException {
 		TestTasksListener tasksListener = new TestTasksListener();
-
 		TasksHandler handler = TasksHandler.builder()
 				.task("1", sleepRunnable())
 				.task("2", sleepRunnable())
 				.task("3", sleepRunnable())
+				.listener(tasksListener)
 				.build();
 
-		handler.addTasksListener(tasksListener);
-
 		TestListener listener = new TestListener();
 		listener.reset(10, 0, 0);
 		StateMachine<String, String> machine = handler.getStateMachine();
@@ -157,16 +184,93 @@ public class TasksHandlerTests {
 		machine.start();
 		assertThat(listener.stateMachineStartedLatch.await(1, TimeUnit.SECONDS), is(true));
 
+		tasksListener.reset(1, 0, 3, 3, 0, 3, 1, 0);
 		handler.runTasks();
 
 		assertThat(listener.stateChangedLatch.await(8, TimeUnit.SECONDS), is(true));
 		assertThat(listener.stateChangedCount, is(10));
 		assertThat(machine.getState().getIds(), contains(TasksHandler.STATE_READY));
-		Map<Object, Object> variables = machine.getExtendedState().getVariables();
-		assertThat(variables.size(), is(3));
 
 		assertThat(tasksListener.onTasksStartedLatch.await(1, TimeUnit.SECONDS), is(true));
 		assertThat(tasksListener.onTasksStarted, is(1));
+		assertThat(tasksListener.onTaskPreExecuteLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTaskPreExecute, is(3));
+		assertThat(tasksListener.onTaskPostExecuteLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTaskPostExecute, is(3));
+		assertThat(tasksListener.onTaskFailed, is(0));
+		assertThat(tasksListener.onTaskSuccessLatch.await(3, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTaskSuccess, is(3));
+		assertThat(tasksListener.onTasksSuccessLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTasksSuccess, is(1));
+	}
+
+	@Test
+	public void testEvents2() throws InterruptedException {
+		TestTasksListener tasksListener = new TestTasksListener();
+		TasksHandler handler = TasksHandler.builder()
+				.task("1", sleepRunnable())
+				.task("2", sleepRunnable())
+				.task("3", failRunnable())
+				.listener(tasksListener)
+				.build();
+
+		TestListener listener = new TestListener();
+		listener.reset(10, 0, 0);
+		StateMachine<String, String> machine = handler.getStateMachine();
+		machine.addStateListener(listener);
+		machine.start();
+		assertThat(listener.stateMachineStartedLatch.await(1, TimeUnit.SECONDS), is(true));
+
+		tasksListener.reset(1, 0, 0, 0, 1, 0, 0, 1);
+		handler.runTasks();
+
+		assertThat(listener.stateChangedLatch.await(8, TimeUnit.SECONDS), is(true));
+		assertThat(listener.stateChangedCount, is(10));
+
+		assertThat(tasksListener.onTasksStartedLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTasksStarted, is(1));
+		assertThat(tasksListener.onTaskSuccessLatch.await(2, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTaskSuccess, is(2));
+		assertThat(tasksListener.onTaskFailedLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTaskFailed, is(1));
+		assertThat(tasksListener.onTasksErrorLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTasksError, is(1));
+		assertThat(tasksListener.onTasksSuccess, is(0));
+	}
+
+	@Test
+	public void testEvents3() throws InterruptedException {
+		TestTasksListener tasksListener = new TestTasksListener();
+		TasksHandler handler = TasksHandler.builder()
+				.task("1", sleepRunnable())
+				.task("2", sleepRunnable())
+				.task("3", failRunnable())
+				.listener(tasksListener)
+				.build();
+
+		TestListener listener = new TestListener();
+		listener.reset(11, 0, 0);
+		StateMachine<String, String> machine = handler.getStateMachine();
+		machine.addStateListener(listener);
+		machine.start();
+		assertThat(listener.stateMachineStartedLatch.await(1, TimeUnit.SECONDS), is(true));
+
+		tasksListener.reset(0, 1, 0, 0, 0, 0, 0, 0);
+		handler.runTasks();
+
+		assertThat(listener.stateChangedLatch.await(8, TimeUnit.SECONDS), is(true));
+		assertThat(listener.stateChangedCount, is(11));
+		assertThat(machine.getState().getIds(), contains(TasksHandler.STATE_ERROR, TasksHandler.STATE_AUTOMATIC));
+
+		handler.fixCurrentProblems();
+		handler.continueFromError();
+		listener.reset(1, 0, 0);
+		assertThat(listener.stateChangedLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(listener.stateChangedCount, is(1));
+		assertThat(machine.getState().getIds(), contains(TasksHandler.STATE_READY));
+
+		assertThat(tasksListener.onTasksContinueLatch.await(1, TimeUnit.SECONDS), is(true));
+		assertThat(tasksListener.onTasksContinue, is(1));
 	}
 
 	private static Runnable sleepRunnable() {
@@ -257,10 +361,18 @@ public class TasksHandlerTests {
 		volatile CountDownLatch onTaskPreExecuteLatch = new CountDownLatch(1);
 		volatile CountDownLatch onTaskPostExecuteLatch = new CountDownLatch(1);
 		volatile CountDownLatch onTaskFailedLatch = new CountDownLatch(1);
+		volatile CountDownLatch onTaskSuccessLatch = new CountDownLatch(1);
 		volatile CountDownLatch onTasksSuccessLatch = new CountDownLatch(1);
 		volatile CountDownLatch onTasksErrorLatch = new CountDownLatch(1);
 
 		volatile int onTasksStarted;
+		volatile int onTasksContinue;
+		volatile int onTaskPreExecute;
+		volatile int onTaskPostExecute;
+		volatile int onTaskFailed;
+		volatile int onTaskSuccess;
+		volatile int onTasksSuccess;
+		volatile int onTasksError;
 
 		@Override
 		public void onTasksStarted() {
@@ -270,34 +382,65 @@ public class TasksHandlerTests {
 
 		@Override
 		public void onTasksContinue() {
+			onTasksContinue++;
 			onTasksContinueLatch.countDown();
 		}
 
 		@Override
 		public void onTaskPreExecute(Object id) {
+			onTaskPreExecute++;
 			onTaskPreExecuteLatch.countDown();
 		}
 
 		@Override
 		public void onTaskPostExecute(Object id) {
+			onTaskPostExecute++;
 			onTaskPostExecuteLatch.countDown();
 		}
 
 		@Override
 		public void onTaskFailed(Object id, Exception exception) {
+			onTaskFailed++;
 			onTaskFailedLatch.countDown();
 		}
 
+		@Override
+		public void onTaskSuccess(Object id) {
+			onTaskSuccess++;
+			onTaskSuccessLatch.countDown();
+		}
+
 		@Override
 		public void onTasksSuccess() {
+			onTasksSuccess++;
 			onTasksSuccessLatch.countDown();
 		}
 
 		@Override
 		public void onTasksError() {
+			onTasksError++;
 			onTasksErrorLatch.countDown();
 		}
 
+		public void reset(int c1, int c2, int c3, int c4, int c5, int c6, int c7, int c8) {
+			onTasksStartedLatch = new CountDownLatch(c1);
+			onTasksContinueLatch = new CountDownLatch(c2);
+			onTaskPreExecuteLatch = new CountDownLatch(c3);
+			onTaskPostExecuteLatch = new CountDownLatch(c4);
+			onTaskFailedLatch = new CountDownLatch(c5);
+			onTaskSuccessLatch = new CountDownLatch(c6);
+			onTasksSuccessLatch = new CountDownLatch(c7);
+			onTasksErrorLatch = new CountDownLatch(c8);
+			onTasksStarted = 0;
+			onTasksContinue = 0;
+			onTaskPreExecute = 0;
+			onTaskPostExecute = 0;
+			onTaskFailed = 0;
+			onTaskSuccess = 0;
+			onTasksSuccess = 0;
+			onTasksError = 0;
+		}
+
 	}
 
 }