From 4b858b1d9bade74ceb50977185b69122ff988610 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Fri, 3 Jul 2015 14:02:17 +0100 Subject: [PATCH] Add skeleton for tasks recipe - Relates to #74 - Tasks recipe not yet fully working, need to get this work in to be able to work with other tickets. - Add childs to StateMachineContext - A lot of fixes to regions concept throughout a code base. Some of the concepts were really broken if there were a complex recursive set of regions, etc. - Add task executor/scheduler to StateMachineBuilder --- build.gradle | 4 + .../statemachine/StateMachineContext.java | 7 +- .../config/AbstractStateMachineFactory.java | 51 +- .../config/ObjectStateMachineFactory.java | 10 +- .../config/StateMachineBuilder.java | 4 +- .../StateMachineConfigurationConfig.java | 16 +- .../DefaultJoinTransitionConfigurer.java | 7 + .../configurers/JoinTransitionConfigurer.java | 10 + .../ensemble/DistributedStateMachine.java | 2 +- .../statemachine/state/RegionState.java | 8 +- .../statemachine/state/StateMachineState.java | 4 +- .../support/AbstractStateMachine.java | 3 +- .../support/DefaultStateMachineContext.java | 27 +- .../transition/TransitionTests.java | 33 +- .../recipes/support/RunnableAction.java | 95 +++ .../recipes/tasks/TasksHandler.java | 561 ++++++++++++++++++ .../recipes/TasksHandlerTests.java | 314 ++++++++++ .../src/test/resources/log4j.properties | 8 + .../ZookeeperStateMachinePersist.java | 2 +- .../ZookeeperStateMachineEnsembleTests.java | 6 +- .../ZookeeperStateMachinePersistTests.java | 2 +- 21 files changed, 1128 insertions(+), 46 deletions(-) create mode 100644 spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java create mode 100644 spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java create mode 100644 spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/TasksHandlerTests.java create mode 100644 spring-statemachine-recipes/src/test/resources/log4j.properties diff --git a/build.gradle b/build.gradle index 6c2eb380..ca727b5c 100644 --- a/build.gradle +++ b/build.gradle @@ -155,6 +155,10 @@ configure(recipeProjects()) { project('spring-statemachine-recipes-common') { dependencies { compile project(":spring-statemachine-core") + testCompile "org.springframework:spring-test:$springVersion" + testCompile "org.hamcrest:hamcrest-core:$hamcrestVersion" + testCompile "org.hamcrest:hamcrest-library:$hamcrestVersion" + testCompile "junit:junit:$junitVersion" } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java index 7c66d07f..249829de 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/StateMachineContext.java @@ -15,6 +15,7 @@ */ package org.springframework.statemachine; +import java.util.List; import java.util.Map; /** @@ -28,11 +29,11 @@ import java.util.Map; public interface StateMachineContext { /** - * Gets the state machine. + * Gets the child contexts if any. * - * @return the state machine + * @return the child contexts */ - StateMachine getStateMachine(); + List> getChilds(); /** * Gets the state. diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java index edc64fb7..46f1a0e0 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/AbstractStateMachineFactory.java @@ -151,19 +151,20 @@ public abstract class AbstractStateMachineFactory extends LifecycleObjectS if (initialCount > 1) { for (Collection> regionStateDatas : regionsStateDatas) { - machine = buildMachine(machineMap, stateMap, regionStateDatas, transitionsData, getBeanFactory(), - contextEvents, defaultExtendedState, stateMachineTransitions, getTaskExecutor(), - getTaskScheduler()); + machine = buildMachine(machineMap, stateMap, regionStateDatas, transitionsData, resolveBeanFactory(), + contextEvents, defaultExtendedState, stateMachineTransitions, resolveTaskExecutor(), + resolveTaskScheduler()); regionStack.push(new MachineStackItem(machine)); } Collection> regions = new ArrayList>(); - for (MachineStackItem si : regionStack) { - regions.add(si.machine); + while (!regionStack.isEmpty()) { + MachineStackItem pop = regionStack.pop(); + regions.add(pop.machine); } S parent = (S)peek.getParent(); - RegionState rstate = new RegionState(parent, regions, null, null, null, - new DefaultPseudoState(PseudoStateKind.INITIAL)); + RegionState rstate = buildRegionStateInternal(parent, regions, null, stateData != null ? stateData.getEntryActions() : null, + stateData != null ? stateData.getExitActions() : null, new DefaultPseudoState(PseudoStateKind.INITIAL)); if (stateData != null) { stateMap.put(stateData.getState(), rstate); } else { @@ -172,13 +173,13 @@ public abstract class AbstractStateMachineFactory extends LifecycleObjectS states.add(rstate); Transition initialTransition = new InitialTransition(rstate); StateMachine m = buildStateMachineInternal(states, new ArrayList>(), rstate, - initialTransition, null, defaultExtendedState, null, contextEvents, getBeanFactory(), - getTaskExecutor(), getTaskScheduler()); + initialTransition, null, defaultExtendedState, null, contextEvents, resolveBeanFactory(), + resolveTaskExecutor(), resolveTaskScheduler()); machine = m; } } else { - machine = buildMachine(machineMap, stateMap, stateDatas, transitionsData, getBeanFactory(), - contextEvents, defaultExtendedState, stateMachineTransitions, getTaskExecutor(), getTaskScheduler()); + machine = buildMachine(machineMap, stateMap, stateDatas, transitionsData, resolveBeanFactory(), + contextEvents, defaultExtendedState, stateMachineTransitions, resolveTaskExecutor(), resolveTaskScheduler()); if (peek.isInitial() || (!peek.isInitial() && !machineMap.containsKey(peek.getParent()))) { machineMap.put(peek.getParent(), machine); } @@ -214,6 +215,30 @@ public abstract class AbstractStateMachineFactory extends LifecycleObjectS this.contextEvents = contextEvents; } + private BeanFactory resolveBeanFactory() { + if (stateMachineConfigurationConfig.getBeanFactory() != null) { + return stateMachineConfigurationConfig.getBeanFactory(); + } else { + return getBeanFactory(); + } + } + + private TaskExecutor resolveTaskExecutor() { + if (stateMachineConfigurationConfig.getTaskExecutor() != null) { + return stateMachineConfigurationConfig.getTaskExecutor(); + } else { + return getTaskExecutor(); + } + } + + private TaskScheduler resolveTaskScheduler() { + if (stateMachineConfigurationConfig.getTaskScheduler() != null) { + return stateMachineConfigurationConfig.getTaskScheduler(); + } else { + return getTaskScheduler(); + } + } + private int getInitialCount(Collection> stateDatas) { int count = 0; for (StateData stateData : stateDatas) { @@ -506,4 +531,8 @@ public abstract class AbstractStateMachineFactory extends LifecycleObjectS return iterator; } + protected abstract RegionState buildRegionStateInternal(S id, Collection> regions, Collection deferred, + Collection> entryActions, Collection> exitActions, + PseudoState pseudoState); + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/ObjectStateMachineFactory.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/ObjectStateMachineFactory.java index 6b2317f1..f44aa31e 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/ObjectStateMachineFactory.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/ObjectStateMachineFactory.java @@ -21,15 +21,17 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.core.task.TaskExecutor; import org.springframework.messaging.Message; import org.springframework.scheduling.TaskScheduler; -import org.springframework.statemachine.ObjectStateMachine; import org.springframework.statemachine.ExtendedState; +import org.springframework.statemachine.ObjectStateMachine; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.action.Action; import org.springframework.statemachine.config.builders.StateMachineConfigurationConfig; import org.springframework.statemachine.config.builders.StateMachineStates; import org.springframework.statemachine.config.builders.StateMachineTransitions; +import org.springframework.statemachine.region.Region; import org.springframework.statemachine.state.ObjectState; import org.springframework.statemachine.state.PseudoState; +import org.springframework.statemachine.state.RegionState; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -80,4 +82,10 @@ public class ObjectStateMachineFactory extends AbstractStateMachineFactory return new ObjectState(id, deferred, entryActions, exitActions, pseudoState); } + @Override + protected RegionState buildRegionStateInternal(S id, Collection> regions, Collection deferred, + Collection> entryActions, Collection> exitActions, PseudoState pseudoState) { + return new RegionState(id, regions, deferred, entryActions, exitActions, pseudoState); + } + } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/StateMachineBuilder.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/StateMachineBuilder.java index 44fe306e..9733edff 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/StateMachineBuilder.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/StateMachineBuilder.java @@ -121,8 +121,8 @@ public class StateMachineBuilder { if (stateMachineConfigurationConfig.getTaskExecutor() != null) { stateMachineFactory.setTaskExecutor(stateMachineConfigurationConfig.getTaskExecutor()); } - if (stateMachineConfigurationConfig.getTaskScheculer() != null) { - stateMachineFactory.setTaskScheduler(stateMachineConfigurationConfig.getTaskScheculer()); + if (stateMachineConfigurationConfig.getTaskScheduler() != null) { + stateMachineFactory.setTaskScheduler(stateMachineConfigurationConfig.getTaskScheduler()); } return stateMachineFactory.getStateMachine(); diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/builders/StateMachineConfigurationConfig.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/builders/StateMachineConfigurationConfig.java index 8eadd128..077145f7 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/builders/StateMachineConfigurationConfig.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/builders/StateMachineConfigurationConfig.java @@ -32,7 +32,7 @@ public class StateMachineConfigurationConfig { private final BeanFactory beanFactory; private final TaskExecutor taskExecutor; - private final TaskScheduler taskScheculer; + private final TaskScheduler taskScheduler; private final StateMachineEnsemble ensemble; /** @@ -40,14 +40,14 @@ public class StateMachineConfigurationConfig { * * @param beanFactory the bean factory * @param taskExecutor the task executor - * @param taskScheculer the task scheculer + * @param taskScheduler the task scheduler * @param ensemble the state machine ensemble */ public StateMachineConfigurationConfig(BeanFactory beanFactory, TaskExecutor taskExecutor, - TaskScheduler taskScheculer, StateMachineEnsemble ensemble) { + TaskScheduler taskScheduler, StateMachineEnsemble ensemble) { this.beanFactory = beanFactory; this.taskExecutor = taskExecutor; - this.taskScheculer = taskScheculer; + this.taskScheduler = taskScheduler; this.ensemble = ensemble; } @@ -70,12 +70,12 @@ public class StateMachineConfigurationConfig { } /** - * Gets the task scheculer. + * Gets the task scheduler. * - * @return the task scheculer + * @return the task scheduler */ - public TaskScheduler getTaskScheculer() { - return taskScheculer; + public TaskScheduler getTaskScheduler() { + return taskScheduler; } /** diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/DefaultJoinTransitionConfigurer.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/DefaultJoinTransitionConfigurer.java index ef2e1ae6..904fd446 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/DefaultJoinTransitionConfigurer.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/DefaultJoinTransitionConfigurer.java @@ -16,6 +16,7 @@ package org.springframework.statemachine.config.configurers; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import org.springframework.statemachine.config.builders.StateMachineTransitionBuilder; @@ -50,6 +51,12 @@ public class DefaultJoinTransitionConfigurer return this; } + @Override + public JoinTransitionConfigurer sources(Collection sources) { + this.sources.addAll(sources); + return this; + } + @Override public JoinTransitionConfigurer target(S target) { this.target = target; diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/JoinTransitionConfigurer.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/JoinTransitionConfigurer.java index 8f477315..6f7d82cf 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/JoinTransitionConfigurer.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/JoinTransitionConfigurer.java @@ -15,6 +15,8 @@ */ package org.springframework.statemachine.config.configurers; +import java.util.Collection; + import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; import org.springframework.statemachine.config.common.annotation.AnnotationConfigurerBuilder; import org.springframework.statemachine.transition.Transition; @@ -39,6 +41,14 @@ public interface JoinTransitionConfigurer */ JoinTransitionConfigurer source(S source); + /** + * Specify a source states {@code S} for this {@link Transition}. + * + * @param sources the sources + * @return configurer for chaining + */ + JoinTransitionConfigurer sources(Collection sources); + /** * Specify a target state {@code S} for this {@link Transition}. * diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java index 400b8bea..a6cf9d62 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java @@ -160,7 +160,7 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem public void preStateChange(State state, Message message, Transition transition, StateMachine stateMachine) { if (message != null && ObjectUtils.nullSafeEquals(uuid, message.getHeaders().get("uuid"))) { - ensemble.setState(new DefaultStateMachineContext(delegate, transition.getTarget() + ensemble.setState(new DefaultStateMachineContext(transition.getTarget() .getId(), message.getPayload(), message.getHeaders(), stateMachine.getExtendedState())); } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/RegionState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/RegionState.java index fe832501..c2e4f940 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/RegionState.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/RegionState.java @@ -78,7 +78,7 @@ public class RegionState extends AbstractState { * @param pseudoState the pseudo state */ public RegionState(S id, Collection> regions, Collection deferred, - Collection> entryActions, Collection> exitActions, PseudoState pseudoState) { + Collection> entryActions, Collection> exitActions, PseudoState pseudoState) { super(id, deferred, entryActions, exitActions, pseudoState, regions); } @@ -92,7 +92,7 @@ public class RegionState extends AbstractState { * @param exitActions the exit actions */ public RegionState(S id, Collection> regions, Collection deferred, - Collection> entryActions, Collection> exitActions) { + Collection> entryActions, Collection> exitActions) { super(id, deferred, entryActions, exitActions, null, regions); } @@ -168,7 +168,9 @@ public class RegionState extends AbstractState { ArrayList> states = new ArrayList>(); states.add(this); for (Region r : getRegions()) { - states.addAll(r.getStates()); + for (State s : r.getStates()) { + states.addAll(s.getStates()); + } } return states; } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateMachineState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateMachineState.java index 0dad7038..b64364d7 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateMachineState.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/state/StateMachineState.java @@ -136,10 +136,10 @@ public class StateMachineState extends AbstractState { // don't stop if it looks like we're coming back // stop would cause start with entry which would // enable default transition and state - if (getSubmachine().getState() != null + if (getSubmachine().getState() != null && context.getTransition() != null && context.getTransition().getSource().getId() != getSubmachine().getState().getId()) { getSubmachine().stop(); - } else if (!StateMachineUtils.isSubstate(context.getTransition().getTarget(), context.getTransition() + } else if (context.getTransition() != null && !StateMachineUtils.isSubstate(context.getTransition().getTarget(), context.getTransition() .getSource())) { getSubmachine().stop(); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java index 8e72733c..b5e3a764 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java @@ -404,8 +404,7 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo } protected boolean acceptEvent(Message message) { - - boolean accepted = currentState.sendEvent(message); + boolean accepted = (currentState != null && currentState.sendEvent(message)); if (accepted) { return true; } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java index c66a2311..c1b1ba37 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java @@ -15,10 +15,11 @@ */ package org.springframework.statemachine.support; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import org.springframework.statemachine.ExtendedState; -import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.StateMachineContext; /** @@ -31,7 +32,7 @@ import org.springframework.statemachine.StateMachineContext; */ public class DefaultStateMachineContext implements StateMachineContext { - private final StateMachine stateMachine; + private final List> childs; private final S state; private final E event; private final Map eventHeaders; @@ -40,14 +41,26 @@ public class DefaultStateMachineContext implements StateMachineContext stateMachine, S state, E event, Map eventHeaders, ExtendedState extendedState) { - this.stateMachine = stateMachine; + public DefaultStateMachineContext(S state, E event, Map eventHeaders, ExtendedState extendedState) { + this(new ArrayList>(), state, event, eventHeaders, extendedState); + } + + /** + * Instantiates a new default state machine context. + * + * @param childs the child state machine contexts + * @param state the state + * @param event the event + * @param eventHeaders the event headers + * @param extendedState the extended state + */ + public DefaultStateMachineContext(List> childs, S state, E event, Map eventHeaders, ExtendedState extendedState) { + this.childs = childs; this.state = state; this.event = event; this.eventHeaders = eventHeaders; @@ -55,8 +68,8 @@ public class DefaultStateMachineContext implements StateMachineContext getStateMachine() { - return stateMachine; + public List> getChilds() { + return childs; } @Override diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java index e82bd4d6..9b172f2e 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/transition/TransitionTests.java @@ -113,12 +113,25 @@ public class TransitionTests extends AbstractStateMachineTests { public void testTriggerlessTransitionInRegionsDefinedInSubStates() throws Exception { context.register(BaseConfig.class, Config5.class); context.refresh(); + + TestAction testAction1 = context.getBean("testAction1", TestAction.class); + TestAction testAction20 = context.getBean("testAction20", TestAction.class); + TestAction testAction21 = context.getBean("testAction21", TestAction.class); + assertTrue(context.containsBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE)); ObjectStateMachine machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, ObjectStateMachine.class); machine.start(); assertThat(machine.getState().getIds(), contains(TestStates.S1)); machine.sendEvent(MessageBuilder.withPayload(TestEvents.E1).build()); + + assertThat(testAction1.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(testAction1.stateContexts.size(), is(1)); + assertThat(testAction20.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(testAction20.stateContexts.size(), is(1)); + assertThat(testAction21.onExecuteLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(testAction21.stateContexts.size(), is(1)); + assertThat(machine.getState().getIds(), containsInAnyOrder(TestStates.S2, TestStates.S201, TestStates.S211)); } @@ -322,16 +335,34 @@ public class TransitionTests extends AbstractStateMachineTests { .source(TestStates.S1) .target(TestStates.S2) .event(TestEvents.E1) + .action(testAction1()) .and() .withExternal() .state(TestStates.S2) .source(TestStates.S20) .target(TestStates.S201) + .action(testAction20()) .and() .withExternal() .state(TestStates.S2) .source(TestStates.S21) - .target(TestStates.S211); + .target(TestStates.S211) + .action(testAction21()); + } + + @Bean + public TestAction testAction1() { + return new TestAction(); + } + + @Bean + public TestAction testAction20() { + return new TestAction(); + } + + @Bean + public TestAction testAction21() { + return new TestAction(); } } 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 new file mode 100644 index 00000000..3e0bf2f3 --- /dev/null +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/support/RunnableAction.java @@ -0,0 +1,95 @@ +/* + * Copyright 2015 the original author or authors. + * + * 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 org.springframework.statemachine.recipes.support; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.action.Action; +import org.springframework.util.Assert; +import org.springframework.util.StopWatch; + +/** + * A {@link Action} which executes a {@link Runnable}. + * + * @author Janne Valkealahti + * + */ +public class RunnableAction implements Action { + + private static final Log log = LogFactory.getLog(RunnableAction.class); + private final Runnable runnable; + private final String id; + + /** + * Instantiates a new runnable action. + * + * @param runnable the runnable + */ + public RunnableAction(Runnable runnable) { + this(runnable, null); + } + + /** + * Instantiates a new runnable action. + * + * @param runnable the runnable + * @param id the optional id for logging + */ + public RunnableAction(Runnable runnable, String id) { + Assert.notNull(runnable, "Runnable must be set"); + this.runnable = runnable; + this.id = id; + } + + @Override + public final void execute(StateContext context) { + if (!shouldExecute(context)) { + return; + } + StopWatch watch = new StopWatch(); + String logId = (id == null ? "" : (" id=" + id)); + log.info("Executing runnable" + logId); + if (log.isDebugEnabled()) { + watch.start(); + } + try { + runnable.run(); + onSuccess(context); + } catch (Exception e) { + onError(context, e); + } + if (log.isDebugEnabled()) { + watch.stop(); + log.debug("Runnable execution took " + watch.getTotalTimeMillis() + " ms" + logId); + } + } + + public String getId() { + return id; + } + + protected boolean shouldExecute(StateContext context) { + return true; + } + + protected void onSuccess(StateContext context) { + } + + protected void onError(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 new file mode 100644 index 00000000..d716e468 --- /dev/null +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/tasks/TasksHandler.java @@ -0,0 +1,561 @@ +/* + * Copyright 2015 the original author or authors. + * + * 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 org.springframework.statemachine.recipes.tasks; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.core.task.TaskExecutor; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineException; +import org.springframework.statemachine.action.Action; +import org.springframework.statemachine.config.StateMachineBuilder; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; +import org.springframework.statemachine.ensemble.StateMachinePersist; +import org.springframework.statemachine.guard.Guard; +import org.springframework.statemachine.listener.AbstractCompositeListener; +import org.springframework.statemachine.recipes.support.RunnableAction; +import org.springframework.statemachine.support.tree.Tree; +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 + * using a state machine logic. + * + * + * + * @author Janne Valkealahti + * + */ +public class TasksHandler { + + private final static Log log = LogFactory.getLog(TasksHandler.class); + + public final static String STATE_READY = "READY"; + public final static String STATE_FORK = "FORK"; + public final static String STATE_TASKS = "TASKS"; + public final static String STATE_JOIN = "JOIN"; + public final static String STATE_CHOICE = "CHOICE"; + public final static String STATE_ERROR = "ERROR"; + public final static String STATE_AUTOMATIC = "AUTOMATIC"; + public final static String STATE_MANUAL = "MANUAL"; + + public final static String STATE_TASKS_PREFIX = "TASK_"; + public final static String STATE_TASKS_INITIAL_POSTFIX = "_INITIAL"; + + public final static String EVENT_RUN = "RUN"; + public final static String EVENT_FALLBACK = "FALLBACK"; + public final static String EVENT_CONTINUE = "CONTINUE"; + public final static String EVENT_FIX = "FIX"; + + private StateMachine stateMachine; + private final CompositeTasksListener listener = new CompositeTasksListener(); + + /** + * Instantiates a new tasks handler. Intentionally private instantiation + * meant to be called from builder. + * + * @param tasks the wrapped tasks + */ + private TasksHandler(List tasks) { + try { + this.stateMachine = buildStateMachine(tasks); + } catch (Exception e) { + throw new StateMachineException("Error building state machine from tasks", e); + } + } + + public void runTasks() { + stateMachine.sendEvent(EVENT_RUN); + } + + public void continueFromError() { + stateMachine.sendEvent(EVENT_CONTINUE); + } + + public void fixCurrentProblems() { + stateMachine.sendEvent(EVENT_FIX); + } + + /** + * Adds the tasks listener. + * + * @param listener the listener + */ + public void addTasksListener(TasksListener listener) { + this.listener.register(listener); + } + + /** + * Removes the tasks listener. + * + * @param listener the listener + */ + public void removeTasksListener(TasksListener listener) { + this.listener.unregister(listener); + } + + /** + * Gets the internal state machine used by executing tasks. + * + * @return the state machine + */ + public StateMachine getStateMachine() { + return stateMachine; + } + + /** + * Gets a new instance of a {@link Builder} which is used to build + * an instance of a {@code TasksHandler}. + * + * @return the tasks handler builder + */ + public static Builder builder() { + return new Builder(); + } + + private StateMachine buildStateMachine(List tasks) throws Exception { + StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); + + builder.configureConfiguration().withConfiguration() + .taskExecutor(taskExecutor()); + + StateMachineStateConfigurer stateMachineStateConfigurer = builder.configureStates(); + StateMachineTransitionConfigurer stateMachineTransitionConfigurer = builder.configureTransitions(); + + stateMachineStateConfigurer + .withStates() + .initial(STATE_READY) + .fork(STATE_FORK) + .state(STATE_TASKS, tasksEntryAction(), null) + .join(STATE_JOIN) + .choice(STATE_CHOICE) + .state(STATE_ERROR); + + stateMachineTransitionConfigurer + .withExternal() + .source(STATE_READY).target(STATE_FORK).event(EVENT_RUN) + .and() + .withFork() + .source(STATE_FORK).target(STATE_TASKS); + + Iterator> iterator = buildTasksIterator(tasks); + String parent = null; + Collection joinStates = new ArrayList(); + while (iterator.hasNext()) { + Node node = iterator.next(); + if (node.getData() == null) { + break; + } + String initial = STATE_TASKS_PREFIX + node.getData().id.toString() + STATE_TASKS_INITIAL_POSTFIX; + String task = STATE_TASKS_PREFIX + node.getData().id.toString(); + parent = node.getData().parent != null ? STATE_TASKS_PREFIX + node.getData().parent.toString() : STATE_TASKS; + + stateMachineStateConfigurer + .withStates() + .parent(parent) + .initial(initial) + .state(task, runnableAction(node.getData().runnable, node.getData().id.toString()), null); + + joinStates.add(task); + + stateMachineTransitionConfigurer + .withExternal() + .source(initial) + .target(task); + } + + stateMachineStateConfigurer + .withStates() + .parent(STATE_ERROR) + .initial(STATE_AUTOMATIC) + .state(STATE_AUTOMATIC, automaticAction(), null) + .state(STATE_MANUAL); + + stateMachineTransitionConfigurer + .withJoin() + .sources(joinStates) + .target(STATE_JOIN) + .and() + .withExternal() + .source(STATE_JOIN).target(STATE_CHOICE) + .and() + .withChoice() + .source(STATE_CHOICE) + .first(STATE_ERROR, tasksChoiceGuard()) + .last(STATE_READY) + .and() + .withExternal() + .source(STATE_ERROR).target(STATE_READY) + .event(EVENT_CONTINUE) + .and() + .withExternal() + .source(STATE_AUTOMATIC).target(STATE_MANUAL) + .event(EVENT_FALLBACK) + .and() + .withInternal() + .source(STATE_MANUAL) + .action(fixAction()) + .event(EVENT_FIX); + + return builder.build(); + } + + private static TaskExecutor taskExecutor() { + ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); + taskExecutor.afterPropertiesSet(); + taskExecutor.setCorePoolSize(5); + return taskExecutor; + } + + private static Iterator> buildTasksIterator(List tasks) { + Tree tree = new Tree(); + for (TaskWrapper wrapper : tasks) { + tree.add(wrapper, wrapper.id, wrapper.parent); + } + + TreeTraverser> traverser = new TreeTraverser>() { + @Override + public Iterable> children(Node root) { + return root.getChildren(); + } + }; + + + Iterable> postOrderTraversal = traverser.postOrderTraversal(tree.getRoot()); + Iterator> iterator = postOrderTraversal.iterator(); + return iterator; + } + + /** + * Builder pattern implementation building a {@link TasksHandler}. + */ + public static class Builder { + + private final List tasks = new ArrayList(); + + /** + * Define a top-level task. + * + * @param id the id + * @param runnable the runnable + * @return the builder + */ + public Builder task(Object id, Runnable runnable) { + tasks.add(new TaskWrapper(null, id, runnable)); + return this; + } + + /** + * Define a sub-task with a reference to its parent. + * + * @param parent the parent + * @param id the id + * @param runnable the runnable + * @return the builder + */ + public Builder task(Object parent, Object id, Runnable runnable) { + tasks.add(new TaskWrapper(parent, id, runnable)); + return this; + } + + /** + * Define a {@link StateMachinePersist} implementation if state machine + * should be persisted with state changes. + * + * @param persist the persist + * @return the builder + */ + public Builder persist(StateMachinePersist persist) { + return this; + } + + /** + * Builds the {@link TasksHandler}. + * + * @return the tasks handler + */ + public TasksHandler build() { + return new TasksHandler(tasks); + } + + } + + private TasksEntryAction tasksEntryAction() { + return new TasksEntryAction(); + } + + private static LocalRunnableAction runnableAction(Runnable runnable, String id) { + return new LocalRunnableAction(runnable, id); + } + + private static Guard tasksChoiceGuard() { + return new Guard() { + + @Override + public boolean evaluate(StateContext context) { + Map variables = context.getExtendedState().getVariables(); + for (Entry entry : variables.entrySet()) { + if (entry.getKey() instanceof String && ((String)entry.getKey()).startsWith(STATE_TASKS_PREFIX)) { + if (entry.getValue() instanceof Integer) { + Integer value = (Integer) entry.getValue(); + if (value < 0) { + if (log.isDebugEnabled()) { + log.debug("Task id=[" + entry.getKey() + "] has negative execution value, tasksChoiceGuard returns true"); + } + return true; + } + } + } + } + return false; + } + }; + } + + private Action automaticAction() { + return new Action() { + + @Override + public void execute(StateContext context) { + } + }; + } + + private Action fixAction() { + return new Action() { + + @Override + public void execute(StateContext context) { + Map variables = context.getExtendedState().getVariables(); + for (Entry entry : variables.entrySet()) { + if (entry.getKey() instanceof String && ((String)entry.getKey()).startsWith(STATE_TASKS_PREFIX)) { + if (entry.getValue() instanceof Integer) { + Integer value = (Integer) entry.getValue(); + if (value < 0) { + variables.put(entry.getValue(), 0); + } + } + } + } + } + }; + } + + /** + * {@code TasksListener} is a generic interface listening tasks + * execution events. + */ + public interface TasksListener { + + /** + * Called when all DAGs have either never executed or previous + * execution was fully successful. + */ + void onTasksStarted(); + + /** + * Called when some of a tasks in DAGs failed to execute and tasks + * execution in going to continue. + */ + void onTasksContinue(); + + /** + * Called before tasks is about to be executed. + * + * @param id the task id + */ + void onTaskPreExecute(Object id); + + /** + * Called after tasks has been executed regardless if task + * execution succeeded or not. + * + * @param id the task id + */ + void onTaskPostExecute(Object id); + + /** + * Called when task execution resulter an error of any kind. + * + * @param id the task id + * @param exception the exception + */ + void onTaskFailed(Object id, Exception exception); + + /** + * Called when all tasks has been executed successfully. + */ + void onTasksSuccess(); + + /** + * Called when after an execution of full DAGs if some of the + * tasks executed with an error. + */ + void onTasksError(); + } + + private class CompositeTasksListener extends AbstractCompositeListener implements + TasksListener { + + @Override + public void onTasksStarted() { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTasksStarted(); + } + } + + @Override + public void onTasksContinue() { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTasksContinue(); + } + } + + @Override + public void onTaskPreExecute(Object id) { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTaskPreExecute(id); + } + } + + @Override + public void onTaskPostExecute(Object id) { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTaskPostExecute(id); + } + } + + @Override + public void onTaskFailed(Object id, Exception exception) { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTaskFailed(id, exception); + } + } + + @Override + public void onTasksSuccess() { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTasksSuccess(); + } + } + + @Override + public void onTasksError() { + for (Iterator iterator = getListeners().reverse(); iterator.hasNext();) { + iterator.next().onTasksError(); + } + } + + } + + /** + * {@link Action} which is executed when TASKS state is entered. + */ + private class TasksEntryAction implements Action { + + @Override + public void execute(StateContext context) { + boolean hasErrors = false; + Map variables = context.getExtendedState().getVariables(); + for (Entry entry : variables.entrySet()) { + if (entry.getKey() instanceof String && ((String)entry.getKey()).startsWith(STATE_TASKS_PREFIX)) { + if (entry.getValue() instanceof Integer) { + Integer value = (Integer) entry.getValue(); + if (value < 0) { + hasErrors = true; + break; + } + } + } + } + if (hasErrors) { + listener.onTasksContinue(); + } else { + listener.onTasksStarted(); + } + } + + } + + /** + * {@link Action} which is execution with every registered {@link Runnable}. + */ + private static class LocalRunnableAction extends RunnableAction { + + public LocalRunnableAction(Runnable runnable, String id) { + super(runnable, id); + } + + @Override + protected boolean shouldExecute(StateContext context) { + return super.shouldExecute(context); + } + + @Override + protected void onSuccess(StateContext context) { + changeCount(1, context); + } + + @Override + protected void onError(StateContext context, Exception e) { + changeCount(-1, context); + } + + private void changeCount(int delta, StateContext context) { + Map variables = context.getExtendedState().getVariables(); + Integer count; + String key = STATE_TASKS_PREFIX + getId(); + if (variables.containsKey(key)) { + count = (Integer) variables.get(key); + } else { + count = 0; + } + count =+ delta; + variables.put(key, count); + } + + } + + /** + * Wrapping a {@link Runnable} with a task identifier and parent if task + * is a subtask. If parent is null it indicates that a task is a top-level + * task with optional child tasks creating a dag task graph. + */ + private static class TaskWrapper { + final Object parent; + final Object id; + final Runnable runnable; + + public TaskWrapper(Object parent, Object id, Runnable runnable) { + this.parent = parent; + this.id = id; + this.runnable = runnable; + } + + } + +} 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 new file mode 100644 index 00000000..e85e1278 --- /dev/null +++ b/spring-statemachine-recipes/src/test/java/org/springframework/statemachine/recipes/TasksHandlerTests.java @@ -0,0 +1,314 @@ +/* + * Copyright 2015 the original author or authors. + * + * 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 org.springframework.statemachine.recipes; + +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertThat; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.listener.StateMachineListenerAdapter; +import org.springframework.statemachine.recipes.tasks.TasksHandler; +import org.springframework.statemachine.recipes.tasks.TasksHandler.TasksListener; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.transition.Transition; + +public class TasksHandlerTests { + + @Test + public void testRunOnceSimpleNoFailures() throws InterruptedException { + TasksHandler handler = TasksHandler.builder() + .task("1", sleepRunnable()) + .task("2", sleepRunnable()) + .task("3", sleepRunnable()) + .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)); + + 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)); + } + + @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)); + Map variables = machine.getExtendedState().getVariables(); + assertThat(variables.size(), is(3)); + } + + @Test + public void testDagSingleRoot() throws InterruptedException { + TasksHandler handler = TasksHandler.builder() + .task("1", sleepRunnable()) + .task("1", "12", sleepRunnable()) + .task("1", "13", sleepRunnable()) + .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)); + + handler.runTasks(); + + assertThat(listener.stateChangedLatch.await(12, 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)); + } + + @Test + public void testDagMultiRoot() throws InterruptedException { + TasksHandler handler = TasksHandler.builder() + .task("1", sleepRunnable()) + .task("1", "12", sleepRunnable()) + .task("1", "13", sleepRunnable()) + .task("2", sleepRunnable()) + .task("2", "22", sleepRunnable()) + .task("2", "23", sleepRunnable()) + .task("3", sleepRunnable()) + .task("3", "32", sleepRunnable()) + .task("3", "33", sleepRunnable()) + .build(); + + TestListener listener = new TestListener(); + listener.reset(22, 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(20, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(22)); + assertThat(machine.getState().getIds(), contains(TasksHandler.STATE_READY)); + Map variables = machine.getExtendedState().getVariables(); + assertThat(variables.size(), is(9)); + } + + @Test + public void testEvents() throws InterruptedException { + TestTasksListener tasksListener = new TestTasksListener(); + + TasksHandler handler = TasksHandler.builder() + .task("1", sleepRunnable()) + .task("2", sleepRunnable()) + .task("3", sleepRunnable()) + .build(); + + handler.addTasksListener(tasksListener); + + 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)); + + 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)); + } + + private static Runnable sleepRunnable() { + return new Runnable() { + + @Override + public void run() { + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + } + } + }; + } + + private static Runnable failRunnable() { + return new Runnable() { + + @Override + public void run() { + throw new RuntimeException(); + } + }; + } + + static class TestListener extends StateMachineListenerAdapter { + + volatile CountDownLatch stateMachineStartedLatch = new CountDownLatch(1); + volatile CountDownLatch stateChangedLatch = new CountDownLatch(1); + volatile CountDownLatch stateEnteredLatch = new CountDownLatch(2); + volatile CountDownLatch stateExitedLatch = new CountDownLatch(0); + volatile CountDownLatch transitionLatch = new CountDownLatch(0); + volatile int stateChangedCount = 0; + volatile int transitionCount = 0; + List> statesEntered = new ArrayList>(); + List> statesExited = new ArrayList>(); + + @Override + public void stateMachineStarted(StateMachine stateMachine) { + stateMachineStartedLatch.countDown(); + } + + @Override + public void stateChanged(State from, State to) { + stateChangedCount++; + stateChangedLatch.countDown(); + } + + @Override + public void stateEntered(State state) { + statesEntered.add(state); + stateEnteredLatch.countDown(); + } + + @Override + public void stateExited(State state) { + statesExited.add(state); + stateExitedLatch.countDown(); + } + + @Override + public void transitionEnded(Transition transition) { + transitionCount++; + transitionLatch.countDown(); + } + + public void reset(int c1, int c2, int c3) { + reset(c1, c2, c3, 0); + } + + public void reset(int c1, int c2, int c3, int c4) { + stateChangedLatch = new CountDownLatch(c1); + stateEnteredLatch = new CountDownLatch(c2); + stateExitedLatch = new CountDownLatch(c3); + transitionLatch = new CountDownLatch(c4); + stateChangedCount = 0; + transitionCount = 0; + statesEntered.clear(); + statesExited.clear(); + } + + } + + private class TestTasksListener implements TasksListener { + + volatile CountDownLatch onTasksStartedLatch = new CountDownLatch(1); + volatile CountDownLatch onTasksContinueLatch = new CountDownLatch(1); + volatile CountDownLatch onTaskPreExecuteLatch = new CountDownLatch(1); + volatile CountDownLatch onTaskPostExecuteLatch = new CountDownLatch(1); + volatile CountDownLatch onTaskFailedLatch = new CountDownLatch(1); + volatile CountDownLatch onTasksSuccessLatch = new CountDownLatch(1); + volatile CountDownLatch onTasksErrorLatch = new CountDownLatch(1); + + volatile int onTasksStarted; + + @Override + public void onTasksStarted() { + onTasksStarted++; + onTasksStartedLatch.countDown(); + } + + @Override + public void onTasksContinue() { + onTasksContinueLatch.countDown(); + } + + @Override + public void onTaskPreExecute(Object id) { + onTaskPreExecuteLatch.countDown(); + } + + @Override + public void onTaskPostExecute(Object id) { + onTaskPostExecuteLatch.countDown(); + } + + @Override + public void onTaskFailed(Object id, Exception exception) { + onTaskFailedLatch.countDown(); + } + + @Override + public void onTasksSuccess() { + onTasksSuccessLatch.countDown(); + } + + @Override + public void onTasksError() { + onTasksErrorLatch.countDown(); + } + + public void reset(int c1, int c2, int c3, int c4, int c5, int c6, int c7) { + onTasksStartedLatch = new CountDownLatch(c1); + onTasksContinueLatch = new CountDownLatch(c2); + onTaskPreExecuteLatch = new CountDownLatch(c3); + onTaskPostExecuteLatch = new CountDownLatch(c4); + onTaskFailedLatch = new CountDownLatch(c5); + onTasksSuccessLatch = new CountDownLatch(c6); + onTasksErrorLatch = new CountDownLatch(c7); + onTasksStarted = 0; + } + + } + +} diff --git a/spring-statemachine-recipes/src/test/resources/log4j.properties b/spring-statemachine-recipes/src/test/resources/log4j.properties new file mode 100644 index 00000000..c7fdcd54 --- /dev/null +++ b/spring-statemachine-recipes/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=INFO, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2} [%t] - %m%n + +log4j.category.org.springframework.statemachine=DEBUG + diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java index 0b1be583..de43665d 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java @@ -125,7 +125,7 @@ public class ZookeeperStateMachinePersist implements StateMachinePersist read(Kryo kryo, Input input, Class> clazz) { E event = (E) kryo.readClassAndObject(input); S state = (S) kryo.readClassAndObject(input); - return new DefaultStateMachineContext(null, state, event, null, null); + return new DefaultStateMachineContext(state, event, null, null); } } diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java index e7e733cf..81183a8e 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java @@ -73,8 +73,8 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { assertThat(curatorClient.checkExists().forPath("/foo/data/current"), notNullValue()); - ensemble.setState(new DefaultStateMachineContext(null, "S1","E1", null, null)); - ensemble.setState(new DefaultStateMachineContext(null, "S2","E1", null, null)); + ensemble.setState(new DefaultStateMachineContext("S1","E1", null, null)); + ensemble.setState(new DefaultStateMachineContext("S2","E1", null, null)); } @@ -109,7 +109,7 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { assertThat(listener1.joinedLatch.await(2, TimeUnit.SECONDS), is(true)); assertThat(listener2.joinedLatch.await(2, TimeUnit.SECONDS), is(true)); - ensemble1.setState(new DefaultStateMachineContext(stateMachine1, "S1", "E1", null, null)); + ensemble1.setState(new DefaultStateMachineContext("S1", "E1", null, null)); assertThat(listener2.eventLatch.await(2, TimeUnit.SECONDS), is(true)); } diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java index fa8cc719..7cfbd8d7 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java @@ -46,7 +46,7 @@ public class ZookeeperStateMachinePersistTests extends AbstractZookeeperTests { curatorClient, "/KryoStateMachinePersistTests"); StateMachineContext contextOut = - new DefaultStateMachineContext(null, "S1", "E1", null, null); + new DefaultStateMachineContext("S1", "E1", null, null); persist.write(contextOut, new Stat()); StateMachineContext contextIn = persist.read(new Stat());