From ddc19392c6905958def2134609cee530d7551883 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sat, 4 Jul 2015 18:13:07 +0100 Subject: [PATCH] Further fixes for tasks recipe - Relates to #74 - Add missing features for listener - Tweak rest of a missing tasks features --- .../recipes/support/RunnableAction.java | 21 ++- .../recipes/tasks/TasksHandler.java | 167 +++++++++++++++--- .../recipes/TasksHandlerTests.java | 157 +++++++++++++++- 3 files changed, 310 insertions(+), 35 deletions(-) diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java index 3e0bf2f3..094c8150 100644 --- a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java @@ -57,7 +57,7 @@ public class RunnableAction implements Action { @Override public final void execute(StateContext context) { - if (!shouldExecute(context)) { + if (!shouldExecute(id, context)) { return; } StopWatch watch = new StopWatch(); @@ -67,10 +67,13 @@ public class RunnableAction implements Action { 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 { return id; } - protected boolean shouldExecute(StateContext context) { + protected boolean shouldExecute(String id, StateContext context) { return true; } - protected void onSuccess(StateContext context) { + protected void onPreExecute(String id, StateContext context) { } - protected void onError(StateContext context, Exception e) { + protected void onPostExecute(String id, StateContext context) { + } + + protected void onSuccess(String id, StateContext context) { + } + + protected void onError(String id, StateContext context, Exception e) { } } \ No newline at end of file diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java index d716e468..86664093 100644 --- a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java +++ b/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 tasks) { + private TasksHandler(List 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 buildStateMachine(List tasks) throws Exception { + private StateMachine buildStateMachine(List tasks, TaskExecutor taskExecutor) + throws Exception { StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); + int taskCount = topLevelTaskCount(tasks); + builder.configureConfiguration().withConfiguration() - .taskExecutor(taskExecutor()); + .taskExecutor(taskExecutor != null ? taskExecutor : taskExecutor(taskCount)); StateMachineStateConfigurer stateMachineStateConfigurer = builder.configureStates(); StateMachineTransitionConfigurer 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 tasks) { + Tree tree = new Tree(); + for (TaskWrapper wrapper : tasks) { + tree.add(wrapper, wrapper.id, wrapper.parent); + } + return tree.getRoot().getChildren().size(); + } + private static Iterator> buildTasksIterator(List tasks) { Tree tree = new Tree(); for (TaskWrapper wrapper : tasks) { @@ -256,13 +282,15 @@ public class TasksHandler { public static class Builder { private final List tasks = new ArrayList(); + 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 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 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 tasksChoiceGuard() { return new Guard() { @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 continueAction() { + return new Action() { + + @Override + public void execute(StateContext context) { + listener.onTasksContinue(); + } + }; + } + private Action automaticAction() { return new Action() { @@ -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 fixAction() { return new Action() { @@ -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 iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTaskSuccess(id); + } + } + @Override public void onTasksSuccess() { for (Iterator 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 context) { - return super.shouldExecute(context); + protected boolean shouldExecute(String id, StateContext context) { + return super.shouldExecute(id, context); } @Override - protected void onSuccess(StateContext context) { + protected void onPreExecute(String id, StateContext context) { + listener.onTaskPreExecute(id); + } + + @Override + protected void onPostExecute(String id, StateContext context) { + listener.onTaskPostExecute(id); + } + + @Override + protected void onSuccess(String id, StateContext context) { + listener.onTaskSuccess(id); changeCount(1, context); } @Override - protected void onError(StateContext context, Exception e) { + protected void onError(String id, StateContext context, Exception e) { + listener.onTaskFailed(id, e); changeCount(-1, context); } diff --git a/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/TasksHandlerTests.java b/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/TasksHandlerTests.java index 2c59cf98..b2db22c8 100644 --- a/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/TasksHandlerTests.java +++ b/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 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 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 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 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 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; + } + } }