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 a6975558..77478dcf 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 @@ -212,7 +212,7 @@ public abstract class AbstractStateMachineFactory extends LifecycleObjectS // we wrap previously build machine with a distributed // state machine and set it to use given ensemble. if (stateMachineConfigurationConfig.getStateMachineEnsemble() != null) { - DistributedStateMachine distributedStateMachine = new DistributedStateMachine<>( + DistributedStateMachine distributedStateMachine = new DistributedStateMachine( stateMachineConfigurationConfig.getStateMachineEnsemble(), machine); distributedStateMachine.setAutoStartup(stateMachineConfigurationConfig.isAutoStart()); distributedStateMachine.afterPropertiesSet(); 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 b54e0a41..552f01d3 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 @@ -75,13 +75,15 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem @Override protected void onInit() throws Exception { - delegate.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + // TODO: should we register with all, not just top one? + delegate.getStateMachineAccessor().doWithRegion(new StateMachineFunction>() { @Override public void apply(StateMachineAccess function) { function.addStateMachineInterceptor(interceptor); } }); + } @Override @@ -176,6 +178,9 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem @Override public void preStateChange(State state, Message message, Transition transition, StateMachine stateMachine) { + if (log.isTraceEnabled()) { + log.trace("Received preStateChange from " + stateMachine + " for delegate " + delegate); + } // only handle if state change originates from this dist machine if (message != null && ObjectUtils.nullSafeEquals(delegate.getId(), 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 c66fca69..4797a9a1 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 @@ -417,6 +417,8 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo if (currentState != null) { buf.append(StringUtils.collectionToCommaDelimitedString(currentState.getIds())); } + buf.append(" / id="); + buf.append(id); return buf.toString(); } diff --git a/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlan.java b/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlan.java index 7dff6401..eac1da68 100644 --- a/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlan.java +++ b/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlan.java @@ -22,13 +22,19 @@ import static org.junit.Assert.assertThat; import java.util.ArrayList; import java.util.Collection; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Map.Entry; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.hamcrest.Matcher; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.test.StateMachineTestPlanBuilder.StateMachineTestPlanStep; import org.springframework.statemachine.test.support.LatchStateMachineListener; +import org.springframework.util.StringUtils; /** * {@code StateMachineTestPlan} is fully constructed plan how @@ -41,18 +47,25 @@ import org.springframework.statemachine.test.support.LatchStateMachineListener; */ public class StateMachineTestPlan { - private final List> stateMachines; + private final static Log log = LogFactory.getLog(StateMachineTestPlan.class); + private final Map> stateMachines; private final List> steps; + private Integer defaultAwaitTime = 10; /** * Instantiates a new state machine test plan. * * @param stateMachines the state machines * @param steps the steps + * @param defaultAwaitTime the default await time in seconds */ - public StateMachineTestPlan(List> stateMachines, List> steps) { + public StateMachineTestPlan(Map> stateMachines, List> steps, + Integer defaultAwaitTime) { this.stateMachines = stateMachines; this.steps = steps; + if (defaultAwaitTime != null) { + this.defaultAwaitTime = defaultAwaitTime; + } } /** @@ -62,39 +75,144 @@ public class StateMachineTestPlan { */ public void test() throws Exception { - List> listeners = new ArrayList>(); - for (StateMachine stateMachine : stateMachines) { + Map, LatchStateMachineListener> listeners = + new HashMap, LatchStateMachineListener>(); + for (StateMachine stateMachine : stateMachines.values()) { LatchStateMachineListener listener = new LatchStateMachineListener(); - listeners.add(listener); + listeners.put(stateMachine, listener); stateMachine.addStateListener(listener); stateMachine.start(); } + log.info("Running test plan for machines " + + StringUtils.collectionToCommaDelimitedString(stateMachines.values())); - + int stepCounter = 0; for (StateMachineTestPlanStep step : steps) { - for (LatchStateMachineListener listener : listeners) { - listener.reset(step.expectStateChanged != null ? step.expectStateChanged : 0, 0, 0, 0, 0, 0, 0, 0, 0); + log.info("Running test plan step " + stepCounter++); + for (LatchStateMachineListener listener : listeners.values()) { + listener.reset( + step.expectStateChanged != null ? step.expectStateChanged : 0, + step.expectStateEntered != null ? step.expectStateEntered : 0, + step.expectStateExited != null ? step.expectStateExited : 0, + step.expectEventNotAccepted != null ? step.expectEventNotAccepted : 0, + step.expectTransition != null ? step.expectTransition : 0, + step.expectTransitionStarted != null ? step.expectTransitionStarted : 0, + step.expectTransitionEnded != null ? step.expectTransitionEnded : 0, + step.expectStateMachineStarted != null ? step.expectStateMachineStarted : 0, + step.expectStateMachineStopped != null ? step.expectStateMachineStopped : 0); } - if (step.sendEvent != null) { - stateMachines.get(0).sendEvent(step.sendEvent); - } - - if (step.expectStateChanged != null) { - for (LatchStateMachineListener listener : listeners) { - assertThat(listener.getStateChangedLatch().await(5, TimeUnit.SECONDS), is(true)); - assertThat(listener.getStateChanged().size(), is(step.expectStateChanged)); + if (step.expectStateMachineStarted != null) { + for (Entry, LatchStateMachineListener> entry : listeners.entrySet()) { + assertThat("StateMachineStarted Await not matched for machine " + entry.getKey(), entry.getValue() + .getStateMachineStartedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat("StateMachineStarted count not matched for machine " + entry.getKey(), entry.getValue() + .getStateMachineStarted().size(), is(step.expectStateMachineStarted)); } } - if (step.expectState != null) { - for (StateMachine stateMachine : stateMachines) { + if (step.sendEvent != null) { + StateMachine sendVia = null; + if (step.sendEventMachineId != null) { + sendVia = stateMachines.get(step.sendEventMachineId); + } else { + sendVia = stateMachines.values().iterator().next(); + } + assertThat("Error finding machine to send via", sendVia, notNullValue()); + log.info("Sending test event " + step.sendEvent + " via machine " + sendVia); + sendVia.sendEvent(step.sendEvent); + } + + if (step.expectStateChanged != null) { + for (Entry, LatchStateMachineListener> entry : listeners.entrySet()) { + assertThat("StateChanged Await not matched for machine " + entry.getKey(), entry.getValue() + .getStateChangedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat("StateChanged count not matched for machine " + entry.getKey(), entry.getValue() + .getStateChanged().size(), is(step.expectStateChanged)); + } + } + + if (step.expectStateEntered != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getStateEnteredLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getStateEntered().size(), is(step.expectStateEntered)); + } + } + + if (step.expectStateExited != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getStateExitedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getStateExited().size(), is(step.expectStateExited)); + } + } + + if (step.expectEventNotAccepted != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getEventNotAcceptedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getEventNotAccepted().size(), is(step.expectEventNotAccepted)); + } + } + + if (step.expectTransition != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getTransitionLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getTransition().size(), is(step.expectTransition)); + } + } + + if (step.expectTransitionStarted != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getTransitionStartedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getTransitionStarted().size(), is(step.expectTransitionStarted)); + } + } + + if (step.expectTransitionEnded != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getTransitionEndedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getTransitionEnded().size(), is(step.expectTransitionEnded)); + } + } + + if (step.expectStateMachineStopped != null) { + for (LatchStateMachineListener listener : listeners.values()) { + assertThat(listener.getStateMachineStoppedLatch().await(defaultAwaitTime, TimeUnit.SECONDS), is(true)); + assertThat(listener.getStateMachineStopped().size(), is(step.expectStateMachineStopped)); + } + } + + if (!step.expectStates.isEmpty()) { + for (StateMachine stateMachine : stateMachines.values()) { assertThat(stateMachine.getState(), notNullValue()); Collection> itemMatchers = new ArrayList>(); - itemMatchers.add(is(step.expectState)); + for (S expectState : step.expectStates) { + itemMatchers.add(is(expectState)); + } assertThat(stateMachine.getState().getIds(), containsInAnyOrder(itemMatchers)); } } + + if (!step.expectVariableKeys.isEmpty()) { + for (StateMachine stateMachine : stateMachines.values()) { + Map variables = stateMachine.getExtendedState().getVariables(); + for (Object key : step.expectVariableKeys) { + assertThat("Key " + key + " doesn't exist in extended state variables", + variables.containsKey(key), is(true)); + } + } + } + + if (!step.expectVariables.isEmpty()) { + for (StateMachine stateMachine : stateMachines.values()) { + Map variables = stateMachine.getExtendedState().getVariables(); + for (Entry entry : step.expectVariables.entrySet()) { + assertThat("Key " + entry.getKey() + " doesn exist in extended state variables", + variables.containsKey(entry.getKey()), is(true)); + assertThat("Variable " + entry.getKey() + " doesn't match in extended state variables", + variables.get(entry.getKey()), is(entry.getValue())); + } + } + } } } diff --git a/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlanBuilder.java b/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlanBuilder.java index b1f28deb..b44a5072 100644 --- a/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlanBuilder.java +++ b/spring-statemachine-test/src/main/java/org/springframework/statemachine/test/StateMachineTestPlanBuilder.java @@ -16,7 +16,11 @@ package org.springframework.statemachine.test; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.springframework.statemachine.StateMachine; @@ -30,8 +34,9 @@ import org.springframework.statemachine.StateMachine; */ public class StateMachineTestPlanBuilder { - private List> stateMachines = new ArrayList>(); + private Map> stateMachines = new HashMap>(); private final List> steps = new ArrayList>(); + private Integer defaultAwaitTime; /** * Gets a new instance of this builder. @@ -49,7 +54,33 @@ public class StateMachineTestPlanBuilder { * @return the state machine test plan builder */ public StateMachineTestPlanBuilder stateMachine(StateMachine stateMachine) { - this.stateMachines.add(stateMachine); + return stateMachine(stateMachine, stateMachine); + } + + /** + * Associate a state machine with this builder. + * + * @param stateMachine the state machine + * @param machineId the machine id to use for sending + * @return the state machine test plan builder + */ + public StateMachineTestPlanBuilder stateMachine(StateMachine stateMachine, Object machineId) { + this.stateMachines.put(machineId, stateMachine); + return this; + } + + /** + * Sets default await time. This is in seconds how long a latch + * will be waited for listening various callbacks. + * + * @param seconds the default await time in seconds + * @return the state machine test plan builder + */ + public StateMachineTestPlanBuilder defaultAwaitTime(int seconds) { + if (seconds < 0) { + throw new IllegalArgumentException("Default await time cannot be negative, was " + seconds); + } + this.defaultAwaitTime = seconds; return this; } @@ -68,7 +99,7 @@ public class StateMachineTestPlanBuilder { * @return the state machine test plan */ public StateMachineTestPlan build() { - return new StateMachineTestPlan(stateMachines, steps); + return new StateMachineTestPlan(stateMachines, steps, defaultAwaitTime); } /** @@ -77,8 +108,20 @@ public class StateMachineTestPlanBuilder { public class StateMachineTestPlanStepBuilder { E sendEvent; - S expectState; + Object sendEventMachineId; + final Collection expectStates = new ArrayList(); Integer expectStateChanged; + Integer expectStateEntered; + Integer expectStateExited; + Integer expectEventNotAccepted; + Integer expectTransition; + Integer expectTransitionStarted; + Integer expectTransitionEnded; + Integer expectStateMachineStarted; + Integer expectStateMachineStopped; + final Collection expectVariableKeys = new ArrayList(); + final Map expectVariables = new HashMap(); + /** * Expect a state {@code S}. @@ -87,18 +130,68 @@ public class StateMachineTestPlanBuilder { * @return the state machine test plan step builder */ public StateMachineTestPlanStepBuilder expectState(S state) { - this.expectState = state; + this.expectStates.add(state); return this; } /** - * Send an event {@code E}. + * Expect a states {@code S}. + * + * @param states the states + * @return the state machine test plan step builder + */ + @SuppressWarnings("unchecked") + public StateMachineTestPlanStepBuilder expectStates(S... states) { + this.expectStates.addAll(Arrays.asList(states)); + return this; + } + + /** + * Send an event {@code E}. In case multiple state machines + * exists, a random one will be chosen to send this event. * * @param event the event * @return the state machine test plan step builder */ public StateMachineTestPlanStepBuilder sendEvent(E event) { + return sendEvent(event, null); + } + + /** + * Send an event {@code E} into a state machine identified + * by {@code machineId}. + * + * @param event the event + * @param machineId the machine identifier for sending event + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder sendEvent(E event, Object machineId) { this.sendEvent = event; + this.sendEventMachineId = machineId; + return this; + } + + /** + * Expect variable to exist in extended state variables. + * + * @param key the key + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectVariable(Object key) { + this.expectVariableKeys.add(key); + return this; + } + + /** + * Expect variable to exist in extended state variables and match + * with the value. + * + * @param key the key + * @param value the value + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectVariable(Object key, Object value) { + this.expectVariables.put(key, value); return this; } @@ -116,6 +209,118 @@ public class StateMachineTestPlanBuilder { return this; } + /** + * Expect state enter happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectStateEntered(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectStateEntered = count; + return this; + } + + /** + * Expect state exit happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectStateExited(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectStateExited = count; + return this; + } + + /** + * Expect event not accepter happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectEventNotAccepted(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectEventNotAccepted = count; + return this; + } + + /** + * Expect transition happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectTransition(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectTransition = count; + return this; + } + + /** + * Expect transition start happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectTransitionStarted(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectTransitionStarted = count; + return this; + } + + /** + * Expect transition end happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectTransitionEnded(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectTransitionEnded = count; + return this; + } + + /** + * Expect state machine start happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectStateMachineStarted(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectStateMachineStarted = count; + return this; + } + + /** + * Expect state machine stop happening {@code count} times. + * + * @param count the count + * @return the state machine test plan step builder + */ + public StateMachineTestPlanStepBuilder expectStateMachineStopped(int count) { + if (count < 0) { + throw new IllegalArgumentException("Expected count cannot be negative, was " + count); + } + this.expectStateMachineStopped = count; + return this; + } + /** * Add a new step and return {@link StateMachineTestPlanBuilder} * for chaining. @@ -123,7 +328,10 @@ public class StateMachineTestPlanBuilder { * @return the state machine test plan builder for chaining */ public StateMachineTestPlanBuilder and() { - steps.add(new StateMachineTestPlanStep(sendEvent, expectState, expectStateChanged)); + steps.add(new StateMachineTestPlanStep(sendEvent, sendEventMachineId, expectStates, + expectStateChanged, expectStateEntered, expectStateExited, expectEventNotAccepted, + expectTransition, expectTransitionStarted, expectTransitionEnded, expectStateMachineStarted, + expectStateMachineStopped, expectVariableKeys, expectVariables)); return StateMachineTestPlanBuilder.this; } @@ -131,12 +339,39 @@ public class StateMachineTestPlanBuilder { static class StateMachineTestPlanStep { E sendEvent; - S expectState; + Object sendEventMachineId; + final Collection expectStates; Integer expectStateChanged; - public StateMachineTestPlanStep(E sendEvent, S expectState, Integer expectStateChanged) { + Integer expectStateEntered; + Integer expectStateExited; + Integer expectEventNotAccepted; + Integer expectTransition; + Integer expectTransitionStarted; + Integer expectTransitionEnded; + Integer expectStateMachineStarted; + Integer expectStateMachineStopped; + final Collection expectVariableKeys; + final Map expectVariables; + + public StateMachineTestPlanStep(E sendEvent, Object sendEventMachineId, Collection expectStates, Integer expectStateChanged, + Integer expectStateEntered, Integer expectStateExited, Integer expectEventNotAccepted, + Integer expectTransition, Integer expectTransitionStarted, Integer expectTransitionEnded, + Integer expectStateMachineStarted, Integer expectStateMachineStopped, + Collection expectVariableKeys, Map expectVariables) { this.sendEvent = sendEvent; - this.expectState = expectState; + this.sendEventMachineId = sendEventMachineId; + this.expectStates = expectStates; this.expectStateChanged = expectStateChanged; + this.expectStateEntered = expectStateEntered; + this.expectStateExited = expectStateExited; + this.expectEventNotAccepted = expectEventNotAccepted; + this.expectTransition = expectTransition; + this.expectTransitionStarted = expectTransitionStarted; + this.expectTransitionEnded = expectTransitionEnded; + this.expectStateMachineStarted = expectStateMachineStarted; + this.expectStateMachineStopped = expectStateMachineStopped; + this.expectVariableKeys = expectVariableKeys; + this.expectVariables = expectVariables; } } diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java index 0849eef1..b06327c8 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java @@ -32,6 +32,7 @@ import org.apache.curator.framework.imps.CuratorFrameworkState; import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex; import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode; import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode.Mode; +import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.data.Stat; import org.springframework.statemachine.StateMachine; @@ -182,9 +183,18 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj if (stateWrapper != null) { stat.setVersion(stateWrapper.version); } + if (log.isDebugEnabled()) { + log.debug("Requesting persist write " + context + " with version " + stat.getVersion() + " for ensemble " + uuid); + } persist.write(context, stat); stateRef.set(new StateWrapper(context, stat.getVersion())); } catch (Exception e) { + if (e instanceof StateMachineException) { + if (((StateMachineException)e).contains(KeeperException.BadVersionException.class)) { + notifyError(new StateMachineEnsembleException("Cas error during write id=[" + uuid + + "] for context=[" + context + "]", e)); + } + } throw new StateMachineException("Error persisting data", e); } } @@ -275,9 +285,16 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj private void mayNotifyStateChanged(StateWrapper wrapper) { StateWrapper notifyWrapper = notifyRef.get(); if (notifyWrapper == null) { + if (log.isDebugEnabled()) { + log.debug("notifyWrapper null, wrapper=[" + wrapper + "] for " + this); + } notifyRef.set(wrapper); notifyStateChanged(wrapper.context); } else if (wrapper.version > notifyWrapper.version) { + if (log.isDebugEnabled()) { + log.debug("Wrapper version higher that notifyWrapper version, notifyWrapper=[" + notifyWrapper + + "], wrapper=[" + wrapper + "] for " + this); + } notifyRef.set(wrapper); notifyStateChanged(wrapper.context); } @@ -285,11 +302,16 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj private void traceLogWrappers(StateWrapper currentWrapper, StateWrapper notifyWrapper, StateWrapper newWrapper) { if (log.isTraceEnabled()) { - log.trace("Wrappers \ncurrentWrapper=[" + currentWrapper + "] \nnotifyWrapper=[" + notifyWrapper - + "] \nnewWrapper=[" + newWrapper + "]"); + log.trace("Wrappers id=" + uuid + "\ncurrentWrapper=[" + currentWrapper + "] \nnotifyWrapper=[" + + notifyWrapper + "] \nnewWrapper=[" + newWrapper + "]"); } } + @Override + public String toString() { + return "ZookeeperStateMachineEnsemble [uuid=" + uuid + "]"; + } + private class StateWatcher implements CuratorWatcher { // zk is not really reliable for watching events because @@ -304,7 +326,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj @Override public void process(WatchedEvent event) throws Exception { if (log.isTraceEnabled()) { - log.trace("Process WatchedEvent: " + event); + log.trace("Process WatchedEvent: id=" + uuid + " " + event); } switch (event.getType()) { case NodeDataChanged: @@ -333,14 +355,25 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj // check if we're behind more than a log size meaning we can't // replay full history, notify and break out from a loop if (i + logSize < ver) { - notifyError(new StateMachineEnsembleException("Current version behind more that log size")); + notifyError(new StateMachineEnsembleException("Current version behind more than log size")); break; } if (log.isDebugEnabled()) { log.debug("Replay position " + i + " with version " + ver); log.debug("Context in position " + i + " " + context); } + StateWrapper wrapper = new StateWrapper(context, ver); + + // need to set stateRef when replaying if its + // context is not set or otherwise just set + // if stateRef still is currentWrapper + StateWrapper currentWrapperx = stateRef.get(); + if (currentWrapperx.context == null) { + stateRef.set(wrapper); + } else if (wrapper.version == currentWrapperx.version + 1){ + stateRef.set(wrapper); + } mayNotifyStateChanged(wrapper); } catch (Exception e) { log.error("error reading log", e); 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 a1d9a3d9..d294224b 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 @@ -17,6 +17,7 @@ package org.springframework.statemachine.zookeeper; import static org.hamcrest.Matchers.greaterThan; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; import static org.junit.Assert.assertThat; @@ -29,12 +30,14 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.curator.framework.CuratorFramework; +import org.apache.zookeeper.KeeperException; import org.junit.Test; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.messaging.Message; import org.springframework.statemachine.ExtendedState; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.StateMachineException; import org.springframework.statemachine.access.StateMachineAccessor; import org.springframework.statemachine.ensemble.EnsembleListeger; import org.springframework.statemachine.ensemble.StateMachineEnsembleException; @@ -229,6 +232,146 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { } } + @Test + public void testContextEventsNotMissedBurstNoOverflow2() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + TestEnsembleListener listener1 = new TestEnsembleListener(); + ensemble1.addEnsembleListener(listener1); + TestEnsembleListener listener2 = new TestEnsembleListener(); + ensemble2.addEnsembleListener(listener2); + + ensemble1.afterPropertiesSet(); + ensemble1.start(); + ensemble2.afterPropertiesSet(); + ensemble2.start(); + + listener1.reset(0, 10); + listener2.reset(0, 10); + + for (int i = 0; i < 10; i++) { + ensemble1.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } + + assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener1.events.size(), is(10)); + assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener2.events.size(), is(10)); + + for (int i = 0; i < 10; i++) { + assertThat(listener1.events.get(i).getEvent(), is("E" + i)); + assertThat(listener2.events.get(i).getEvent(), is("E" + i)); + } + } + + @Test + public void testContextEventsNotMissedBurstNoOverflow3() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + TestEnsembleListener listener1 = new TestEnsembleListener(); + ensemble1.addEnsembleListener(listener1); + TestEnsembleListener listener2 = new TestEnsembleListener(); + ensemble2.addEnsembleListener(listener2); + + ensemble1.afterPropertiesSet(); + ensemble1.start(); + ensemble2.afterPropertiesSet(); + ensemble2.start(); + + listener1.reset(0, 10); + listener2.reset(0, 10); + + Exception e = null; + try { + for (int i = 0; i < 10; i++) { + if (((i % 2) == 0)) { + ensemble1.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } else { + ensemble2.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } + } + } catch (Exception ee) { + e = ee; + } + + if (e != null) { + assertThat(e, instanceOf(StateMachineException.class)); + assertThat(((StateMachineException)e).contains(KeeperException.BadVersionException.class), is(true)); + } else { + // miracle happened and no cas error, well then check events + assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener1.events.size(), is(10)); + assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener2.events.size(), is(10)); + + for (int i = 0; i < 10; i++) { + assertThat(listener1.events.get(i).getEvent(), is("E" + i)); + assertThat(listener2.events.get(i).getEvent(), is("E" + i)); + } + } + } + + @Test + public void testContextEventsNotMissedBurstNoOverflow4() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + TestEnsembleListener listener1 = new TestEnsembleListener(); + ensemble1.addEnsembleListener(listener1); + TestEnsembleListener listener2 = new TestEnsembleListener(); + ensemble2.addEnsembleListener(listener2); + + ensemble1.afterPropertiesSet(); + ensemble1.start(); + ensemble2.afterPropertiesSet(); + ensemble2.start(); + + for (int i = 0; i < 10; i++) { + listener1.reset(0, 1); + listener2.reset(0, 1); + if (((i % 2) == 0)) { + ensemble1.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } else { + ensemble2.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } + assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener1.events.size(), is(1)); + assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener2.events.size(), is(1)); + } + } + @Test public void testContextEventsNotMissedSlowNoOverflow() throws Exception { context.register(ZkServerConfig.class, BaseConfig.class); @@ -261,6 +404,108 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { } } + @Test + public void testContextEventsNotMissedSlowNoOverflow2() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + TestEnsembleListener listener1 = new TestEnsembleListener(); + TestEnsembleListener listener2 = new TestEnsembleListener(); + ensemble1.addEnsembleListener(listener1); + ensemble2.addEnsembleListener(listener2); + + ensemble1.afterPropertiesSet(); + ensemble1.start(); + ensemble2.afterPropertiesSet(); + ensemble2.start(); + listener1.reset(0, 10); + listener2.reset(0, 10); + + for (int i = 0; i < 10; i++) { + ensemble1.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + Thread.sleep(500); + } + + assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener1.events.size(), is(10)); + assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener2.events.size(), is(10)); + + for (int i = 0; i < 10; i++) { + assertThat(listener1.events.get(i).getEvent(), is("E" + i)); + assertThat(listener2.events.get(i).getEvent(), is("E" + i)); + } + } + + @Test + public void testDoesNotThrowCasError() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + + TestEnsembleListener listener1 = new TestEnsembleListener(); + TestEnsembleListener listener2 = new TestEnsembleListener(); + ensemble1.addEnsembleListener(listener1); + ensemble2.addEnsembleListener(listener2); + + ensemble1.afterPropertiesSet(); + ensemble1.start(); + ensemble2.afterPropertiesSet(); + ensemble2.start(); + listener1.reset(0, 9); + listener2.reset(0, 9); + + for (int i = 0; i < 9; i++) { + ensemble1.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } + + assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener1.events.size(), is(9)); + assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener2.events.size(), is(9)); + + for (int i = 0; i < 9; i++) { + assertThat(listener1.events.get(i).getEvent(), is("E" + i)); + assertThat(listener2.events.get(i).getEvent(), is("E" + i)); + } + + listener1.reset(0, 1); + listener2.reset(0, 1); + + // should not throw BadVersionException when we immediately + for (int i = 9; i < 10; i++) { + ensemble2.setState(new DefaultStateMachineContext("S" + i, "E" + i, + new HashMap(), new DefaultExtendedState())); + } + + assertThat(listener1.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener1.events.size(), is(1)); + assertThat(listener2.eventLatch.await(3, TimeUnit.SECONDS), is(true)); + assertThat(listener2.events.size(), is(1)); + + for (int i = 0; i < 1; i++) { + assertThat(listener1.events.get(i).getEvent(), is("E" + (i+9))); + assertThat(listener2.events.get(i).getEvent(), is("E" + (i+9))); + } + } + @Test public void testEventsOverflow() throws Exception { context.register(ZkServerConfig.class, BaseConfig.class); diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java index 141ae63a..d14b7872 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java @@ -19,6 +19,7 @@ import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.is; import static org.junit.Assert.assertThat; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -29,14 +30,19 @@ import org.springframework.context.SmartLifecycle; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.action.Action; import org.springframework.statemachine.config.EnableStateMachine; +import org.springframework.statemachine.config.StateMachineBuilder; import org.springframework.statemachine.config.StateMachineConfigurerAdapter; import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; import org.springframework.statemachine.ensemble.DistributedStateMachine; import org.springframework.statemachine.ensemble.StateMachineEnsemble; +import org.springframework.statemachine.guard.Guard; import org.springframework.statemachine.listener.StateMachineListenerAdapter; import org.springframework.statemachine.state.State; import org.springframework.statemachine.test.StateMachineTestPlan; @@ -111,6 +117,67 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { assertThat(machine2.getState().getIds(), containsInAnyOrder("S2")); } + @Test + @SuppressWarnings("unchecked") + public void testStateChangesManualSetupSendDifferentMachines() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class, Config1.class, Config2.class); + context.refresh(); + + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + TestListener listener1 = new TestListener(); + TestListener listener2 = new TestListener(); + machine1.addStateListener(listener1); + machine2.addStateListener(listener2); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ZookeeperStateMachineEnsemble ensemble2 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ensemble1.afterPropertiesSet(); + ensemble2.afterPropertiesSet(); + ensemble1.start(); + ensemble2.start(); + + DistributedStateMachine machine1s = + new DistributedStateMachine(ensemble1, machine1); + + DistributedStateMachine machine2s = + new DistributedStateMachine(ensemble2, machine2); + + machine1s.afterPropertiesSet(); + machine2s.afterPropertiesSet(); + + machine1s.start(); + machine2s.start(); + + listener1.reset(1); + listener2.reset(1); + machine1s.sendEvent("E1"); + assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener1.stateChangedCount, is(1)); + assertThat(listener2.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener2.stateChangedCount, is(1)); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S1")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S1")); + + listener1.reset(1); + listener2.reset(1); + machine2s.sendEvent("E2"); + assertThat(listener1.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener1.stateChangedCount, is(1)); + assertThat(listener2.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener2.stateChangedCount, is(1)); + assertThat(machine1.getState().getIds(), containsInAnyOrder("S2")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("S2")); + } + @Test @SuppressWarnings("unchecked") public void testLifecycle() throws Exception { @@ -151,6 +218,45 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { plan.test(); } + @Test + public void testVariousChangesInShowcase() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + StateMachine machine1 = + buildTestStateMachine(curatorClient); + + StateMachine machine2 = + buildTestStateMachine(curatorClient); + + StateMachineTestPlan plan = + StateMachineTestPlanBuilder.builder() + .defaultAwaitTime(2) + .stateMachine(machine1) + .stateMachine(machine2) + .step() + .expectStates("S0", "S1", "S11") + .expectVariable("foo") + .expectVariable("foo", 0) + .and() + .step() + .sendEvent("C", machine1) + .expectStateChanged(3) + .expectStates("S0", "S2", "S21", "S211") + .and() + .step() + .sendEvent("C", machine2) + .expectStateChanged(2) + .expectStates("S0", "S1", "S11") + .and() + .build(); + + plan.test(); + } + @Test @SuppressWarnings("unchecked") public void testJoinLaterShouldSyncState() throws Exception { @@ -330,4 +436,161 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { } + private static StateMachineEnsemble stateMachineEnsemble(CuratorFramework curatorClient) { + ZookeeperStateMachineEnsemble ensemble = new ZookeeperStateMachineEnsemble( + curatorClient, "/foo"); + ensemble.afterPropertiesSet(); + ensemble.start(); + return ensemble; + } + + private StateMachine buildTestStateMachine(CuratorFramework curatorClient) + throws Exception { + StateMachineBuilder.Builder builder = StateMachineBuilder.builder(); + + builder.configureConfiguration() + .withConfiguration() + .taskExecutor(new SyncTaskExecutor()) + .autoStartup(true) + .and() + .withDistributed() + .ensemble(stateMachineEnsemble(curatorClient)); + + builder.configureStates() + .withStates() + .initial("S0", fooAction()) + .state("S0") + .and() + .withStates() + .parent("S0") + .initial("S1") + .state("S1") + .and() + .withStates() + .parent("S1") + .initial("S11") + .state("S11") + .state("S12") + .and() + .withStates() + .parent("S0") + .state("S2") + .and() + .withStates() + .parent("S2") + .initial("S21") + .state("S21") + .and() + .withStates() + .parent("S21") + .initial("S211") + .state("S211") + .state("S212"); + + builder.configureTransitions() + .withExternal() + .source("S1").target("S1").event("A") + .guard(foo1Guard()) + .and() + .withExternal() + .source("S1").target("S11").event("B") + .and() + .withExternal() + .source("S21").target("S211").event("B") + .and() + .withExternal() + .source("S1").target("S2").event("C") + .and() + .withExternal() + .source("S2").target("S1").event("C") + .and() + .withExternal() + .source("S1").target("S0").event("D") + .and() + .withExternal() + .source("S211").target("S21").event("D") + .and() + .withExternal() + .source("S0").target("S211").event("E") + .and() + .withExternal() + .source("S1").target("S211").event("F") + .and() + .withExternal() + .source("S2").target("S11").event("F") + .and() + .withExternal() + .source("S11").target("S211").event("G") + .and() + .withExternal() + .source("S211").target("S0").event("G") + .and() + .withInternal() + .source("S0").event("H") + .guard(foo0Guard()) + .action(fooAction()) + .and() + .withInternal() + .source("S2").event("H") + .guard(foo1Guard()) + .action(fooAction()) + .and() + .withInternal() + .source("S1").event("H") + .and() + .withExternal() + .source("S11").target("S12").event("I") + .and() + .withExternal() + .source("S211").target("S212").event("I") + .and() + .withExternal() + .source("S12").target("S212").event("I"); + + return builder.build(); + } + + private static FooGuard foo0Guard() { + return new FooGuard(0); + } + + private static FooGuard foo1Guard() { + return new FooGuard(1); + } + + private static FooAction fooAction() { + return new FooAction(); + } + + private static class FooGuard implements Guard { + + private final int match; + + public FooGuard(int match) { + this.match = match; + } + + @Override + public boolean evaluate(StateContext context) { + Object foo = context.getExtendedState().getVariables().get("foo"); + return !(foo == null || !foo.equals(match)); + } + } + + private static class FooAction implements Action { + + @Override + public void execute(StateContext context) { + Map variables = context.getExtendedState().getVariables(); + Integer foo = context.getExtendedState().get("foo", Integer.class); + if (foo == null) { + variables.put("foo", 0); + } else if (foo == 0) { + variables.put("foo", 1); + } else if (foo == 1) { + variables.put("foo", 0); + } + } + } + }