From 15c6e507d85e0bd1d6c501f4b7b0303560d20f5a Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Wed, 15 Jul 2015 11:17:43 +0100 Subject: [PATCH] Tweak concepts around state reset - StateMachineAccess, replace state reset and variables with a StateMachineContext. - Additional fixes to reset state properly even if target transition is a super state. - Some polish - Add tests --- .../access/StateMachineAccess.java | 15 +- .../configurers/ConfigurationConfigurer.java | 1 - .../ensemble/DistributedStateMachine.java | 13 +- .../support/AbstractStateMachine.java | 38 ++- .../support/DefaultStateMachineContext.java | 9 +- .../statemachine/StateMachineResetTests.java | 289 ++++++++++++++++++ .../access/StateMachineAccessTests.java | 7 +- .../support/StateChangeInterceptorTests.java | 289 ++++++++++++++++++ .../persist/PersistStateMachineHandler.java | 3 +- .../main/java/demo/zookeeper/Application.java | 8 +- .../ZookeeperStateMachineEnsemble.java | 47 ++- .../ZookeeperStateMachinePersist.java | 29 +- .../ZookeeperStateMachineEnsembleTests.java | 7 +- .../zookeeper/ZookeeperStateMachineTests.java | 140 ++++++++- 14 files changed, 841 insertions(+), 54 deletions(-) create mode 100644 spring-statemachine-core/src/test/java/org/springframework/statemachine/StateMachineResetTests.java create mode 100644 spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java index a24f3cd1..5d2342b1 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java @@ -15,8 +15,8 @@ */ package org.springframework.statemachine.access; -import org.springframework.statemachine.ExtendedState; import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.support.StateChangeInterceptor; /** @@ -37,18 +37,11 @@ public interface StateMachineAccess { void setRelay(StateMachine stateMachine); /** - * Reset state. + * Reset state machine. * - * @param state the state + * @param stateMachineContext the state machine context */ - void resetState(S state); - - /** - * Sets the extended state. - * - * @param extendedState the new extended state - */ - void setExtendedState(ExtendedState extendedState); + void resetStateMachine(StateMachineContext stateMachineContext); /** * Adds the state change interceptor. diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/ConfigurationConfigurer.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/ConfigurationConfigurer.java index 010d2504..c39af0d6 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/ConfigurationConfigurer.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/ConfigurationConfigurer.java @@ -19,7 +19,6 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.core.task.TaskExecutor; import org.springframework.scheduling.TaskScheduler; import org.springframework.statemachine.config.builders.StateMachineConfigurationConfigurer; -import org.springframework.statemachine.config.builders.StateMachineConfigurer; import org.springframework.statemachine.config.common.annotation.AnnotationConfigurerBuilder; import org.springframework.statemachine.listener.StateMachineListener; 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 a6cf9d62..1ba8ec60 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 @@ -42,6 +42,9 @@ import org.springframework.util.ObjectUtils; * together with a {@link StateMachineEnsemble} order to provide a distributed state * machine. * + * Every distributed state machine will enter its initial state regardless of + * a distributed state status. + * * @author Janne Valkealahti * * @param the type of state @@ -168,7 +171,8 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem } /** - * + * Bridge for instructing delegating machine based on what + * is happening in an ensemble. */ private class LocalEnsembleListener implements EnsembleListeger { @@ -178,12 +182,15 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem // I'm now successfully joined, so set delegating // sm to current known state by a context. + if (log.isDebugEnabled()) { + log.debug("Joining with context " + context); + } + delegate.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { @Override public void apply(StateMachineAccess function) { - function.resetState(context.getState()); - function.setExtendedState(context.getExtendedState()); + function.resetStateMachine(context); } }); 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 84ec2b7c..33f6a8a8 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 @@ -37,6 +37,7 @@ import org.springframework.messaging.support.MessageBuilder; import org.springframework.statemachine.ExtendedState; import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.access.StateMachineAccess; import org.springframework.statemachine.access.StateMachineAccessor; import org.springframework.statemachine.access.StateMachineFunction; @@ -170,11 +171,6 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return extendedState; } - @Override - public void setExtendedState(ExtendedState extendedState) { - this.extendedState = extendedState; - } - public void setHistoryState(PseudoState history) { this.history = history; } @@ -422,13 +418,30 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo } @Override - public void resetState(S state) { + public void resetStateMachine(StateMachineContext stateMachineContext) { + S state = stateMachineContext.getState(); + boolean stateSet = false; for (State s : getStates()) { - if (s.getId().equals(state)) { - currentState = s; + for (State ss : s.getStates()) { + if (ss.getIds().contains(state)) { + currentState = s; + // TODO: not sure about starting submachine here, though + // needed if we only transit to super state + if (s.isSubmachineState()) { + StateMachine submachine = ((AbstractState)s).getSubmachine(); + submachine.start(); + } + stateSet = true; + break; + } + } + if (stateSet) { break; } } + if (stateSet && stateMachineContext.getExtendedState() != null) { + this.extendedState = stateMachineContext.getExtendedState(); + } } @Override @@ -479,8 +492,12 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo return true; } + private boolean isInitialTransition(Transition transition) { + return transition != null && transition.getKind() == TransitionKind.INITIAL; + } + private void switchToState(State state, Message message, Transition transition, StateMachine stateMachine) { - if (!callStateChangeInterceptors(state, message, transition, stateMachine)) { + if (!isInitialTransition(transition) && !callStateChangeInterceptors(state, message, transition, stateMachine)) { return; } // TODO: need to make below more clear when @@ -558,9 +575,6 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo if (isTargetSubOf && currentState == transition.getTarget()) { state = transition.getSource(); } -// else if (currentState == null && StateMachineUtils.isSubstate(findDeep, state)) { -// state = findDeep; -// } } boolean nonDeepStatePresent = false; 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 c1b1ba37..c01bbf1a 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 @@ -59,7 +59,8 @@ public class DefaultStateMachineContext implements StateMachineContext> childs, S state, E event, Map eventHeaders, ExtendedState extendedState) { + public DefaultStateMachineContext(List> childs, S state, E event, + Map eventHeaders, ExtendedState extendedState) { this.childs = childs; this.state = state; this.event = event; @@ -92,4 +93,10 @@ public class DefaultStateMachineContext implements StateMachineContext machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class); + + Map variables = new HashMap(); + variables.put("foo", 1); + ExtendedState extendedState = new DefaultExtendedState(variables); + DefaultStateMachineContext stateMachineContext = new DefaultStateMachineContext(States.S12, Events.I, null, extendedState); + + machine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.resetStateMachine(stateMachineContext); + } + }); + + machine.start(); + assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S1, States.S12)); + assertThat((Integer)machine.getExtendedState().getVariables().get("foo"), is(1)); + } + + @Test + public void testResetSubStates2() throws Exception { + context.register(Config1.class); + context.refresh(); + @SuppressWarnings("unchecked") + StateMachine machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class); + + Map variables = new HashMap(); + variables.put("foo", 1); + ExtendedState extendedState = new DefaultExtendedState(variables); + DefaultStateMachineContext stateMachineContext = new DefaultStateMachineContext(States.S211, Events.C, null, extendedState); + + machine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.resetStateMachine(stateMachineContext); + } + }); + + machine.start(); + assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S2, States.S21, States.S211)); + assertThat((Integer)machine.getExtendedState().getVariables().get("foo"), is(1)); + } + + @Test + public void testResetSubStates3() throws Exception { + context.register(Config1.class); + context.refresh(); + @SuppressWarnings("unchecked") + StateMachine machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class); + + Map variables = new HashMap(); + variables.put("foo", 1); + ExtendedState extendedState = new DefaultExtendedState(variables); + DefaultStateMachineContext stateMachineContext = new DefaultStateMachineContext(States.S2, Events.C, null, extendedState); + + machine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.resetStateMachine(stateMachineContext); + } + }); + + machine.start(); + assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S2, States.S21, States.S211)); + assertThat((Integer)machine.getExtendedState().getVariables().get("foo"), is(1)); + } + + @Configuration + @EnableStateMachine + static class Config1 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) + throws Exception { + states + .withStates() + .initial(States.S0, fooAction()) + .state(States.S0) + .and() + .withStates() + .parent(States.S0) + .initial(States.S1) + .state(States.S1) + .and() + .withStates() + .parent(States.S1) + .initial(States.S11) + .state(States.S11) + .state(States.S12) + .and() + .withStates() + .parent(States.S0) + .state(States.S2) + .and() + .withStates() + .parent(States.S2) + .initial(States.S21) + .state(States.S21) + .and() + .withStates() + .parent(States.S21) + .initial(States.S211) + .state(States.S211) + .state(States.S212); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) + throws Exception { + transitions + .withExternal() + .source(States.S1).target(States.S1).event(Events.A) + .guard(foo1Guard()) + .and() + .withExternal() + .source(States.S1).target(States.S11).event(Events.B) + .and() + .withExternal() + .source(States.S21).target(States.S211).event(Events.B) + .and() + .withExternal() + .source(States.S1).target(States.S2).event(Events.C) + .and() + .withExternal() + .source(States.S2).target(States.S1).event(Events.C) + .and() + .withExternal() + .source(States.S1).target(States.S0).event(Events.D) + .and() + .withExternal() + .source(States.S211).target(States.S21).event(Events.D) + .and() + .withExternal() + .source(States.S0).target(States.S211).event(Events.E) + .and() + .withExternal() + .source(States.S1).target(States.S211).event(Events.F) + .and() + .withExternal() + .source(States.S2).target(States.S11).event(Events.F) + .and() + .withExternal() + .source(States.S11).target(States.S211).event(Events.G) + .and() + .withExternal() + .source(States.S211).target(States.S0).event(Events.G) + .and() + .withInternal() + .source(States.S0).event(Events.H) + .guard(foo0Guard()) + .action(fooAction()) + .and() + .withInternal() + .source(States.S2).event(Events.H) + .guard(foo1Guard()) + .action(fooAction()) + .and() + .withInternal() + .source(States.S1).event(Events.H) + .and() + .withExternal() + .source(States.S11).target(States.S12).event(Events.I) + .and() + .withExternal() + .source(States.S211).target(States.S212).event(Events.I) + .and() + .withExternal() + .source(States.S12).target(States.S212).event(Events.I); + + } + + @Bean + public FooGuard foo0Guard() { + return new FooGuard(0); + } + + @Bean + public FooGuard foo1Guard() { + return new FooGuard(1); + } + + @Bean + public FooAction fooAction() { + return new FooAction(); + } + + } + + public static enum States { + S0, S1, S11, S12, S2, S21, S211, S212 + } + + public static enum Events { + A, B, C, D, E, F, G, H, I + } + + 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); + } + } + } + + 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)); + } + } + +} diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java index 081fa4bf..c9a3948a 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java @@ -26,6 +26,7 @@ import org.junit.Test; import org.springframework.messaging.Message; import org.springframework.statemachine.ExtendedState; import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.listener.StateMachineListener; import org.springframework.statemachine.state.State; import org.springframework.statemachine.support.StateChangeInterceptor; @@ -100,11 +101,7 @@ public class StateMachineAccessTests { } @Override - public void resetState(String state) { - } - - @Override - public void setExtendedState(ExtendedState extendedState) { + public void resetStateMachine(StateMachineContext stateMachineContext) { } @Override diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java new file mode 100644 index 00000000..12510590 --- /dev/null +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java @@ -0,0 +1,289 @@ +/* + * 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.support; + +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; + +import org.junit.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; +import org.springframework.statemachine.AbstractStateMachineTests; +import org.springframework.statemachine.StateContext; +import org.springframework.statemachine.StateMachine; +import org.springframework.statemachine.StateMachineSystemConstants; +import org.springframework.statemachine.access.StateMachineAccess; +import org.springframework.statemachine.access.StateMachineFunction; +import org.springframework.statemachine.action.Action; +import org.springframework.statemachine.config.EnableStateMachine; +import org.springframework.statemachine.config.EnumStateMachineConfigurerAdapter; +import org.springframework.statemachine.config.builders.StateMachineStateConfigurer; +import org.springframework.statemachine.config.builders.StateMachineTransitionConfigurer; +import org.springframework.statemachine.guard.Guard; +import org.springframework.statemachine.listener.StateMachineListenerAdapter; +import org.springframework.statemachine.state.State; +import org.springframework.statemachine.transition.Transition; + +public class StateChangeInterceptorTests extends AbstractStateMachineTests { + + @Override + protected AnnotationConfigApplicationContext buildContext() { + return new AnnotationConfigApplicationContext(); + } + + @Test + public void testIntercept() throws InterruptedException { + context.register(Config1.class); + context.refresh(); + @SuppressWarnings("unchecked") + StateMachine machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class); + TestListener listener = new TestListener(); + machine.addStateListener(listener); + TestStateChangeInterceptor interceptor = new TestStateChangeInterceptor(); + + machine.getStateMachineAccessor().doWithRegion(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.addStateChangeInterceptor(interceptor); + } + }); + + + machine.start(); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S1, States.S11)); + + assertThat((Integer)machine.getExtendedState().getVariables().get("foo"), is(0)); + + + listener.reset(3); + interceptor.reset(1); + machine.sendEvent(Events.C); + assertThat(listener.stateChangedLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(listener.stateChangedCount, is(3)); + assertThat(interceptor.preStateChangeLatch.await(2, TimeUnit.SECONDS), is(true)); + assertThat(interceptor.preStateChangeCount, is(1)); + assertThat(machine.getState().getIds(), containsInAnyOrder(States.S0, States.S2, States.S21, States.S211)); + } + + @Configuration + @EnableStateMachine + static class Config1 extends EnumStateMachineConfigurerAdapter { + + @Override + public void configure(StateMachineStateConfigurer states) + throws Exception { + states + .withStates() + .initial(States.S0, fooAction()) + .state(States.S0) + .and() + .withStates() + .parent(States.S0) + .initial(States.S1) + .state(States.S1) + .and() + .withStates() + .parent(States.S1) + .initial(States.S11) + .state(States.S11) + .state(States.S12) + .and() + .withStates() + .parent(States.S0) + .state(States.S2) + .and() + .withStates() + .parent(States.S2) + .initial(States.S21) + .state(States.S21) + .and() + .withStates() + .parent(States.S21) + .initial(States.S211) + .state(States.S211) + .state(States.S212); + } + + @Override + public void configure(StateMachineTransitionConfigurer transitions) + throws Exception { + transitions + .withExternal() + .source(States.S1).target(States.S1).event(Events.A) + .guard(foo1Guard()) + .and() + .withExternal() + .source(States.S1).target(States.S11).event(Events.B) + .and() + .withExternal() + .source(States.S21).target(States.S211).event(Events.B) + .and() + .withExternal() + .source(States.S1).target(States.S2).event(Events.C) + .and() + .withExternal() + .source(States.S2).target(States.S1).event(Events.C) + .and() + .withExternal() + .source(States.S1).target(States.S0).event(Events.D) + .and() + .withExternal() + .source(States.S211).target(States.S21).event(Events.D) + .and() + .withExternal() + .source(States.S0).target(States.S211).event(Events.E) + .and() + .withExternal() + .source(States.S1).target(States.S211).event(Events.F) + .and() + .withExternal() + .source(States.S2).target(States.S11).event(Events.F) + .and() + .withExternal() + .source(States.S11).target(States.S211).event(Events.G) + .and() + .withExternal() + .source(States.S211).target(States.S0).event(Events.G) + .and() + .withInternal() + .source(States.S0).event(Events.H) + .guard(foo0Guard()) + .action(fooAction()) + .and() + .withInternal() + .source(States.S2).event(Events.H) + .guard(foo1Guard()) + .action(fooAction()) + .and() + .withInternal() + .source(States.S1).event(Events.H) + .and() + .withExternal() + .source(States.S11).target(States.S12).event(Events.I) + .and() + .withExternal() + .source(States.S211).target(States.S212).event(Events.I) + .and() + .withExternal() + .source(States.S12).target(States.S212).event(Events.I); + + } + + @Bean + public FooGuard foo0Guard() { + return new FooGuard(0); + } + + @Bean + public FooGuard foo1Guard() { + return new FooGuard(1); + } + + @Bean + public FooAction fooAction() { + return new FooAction(); + } + + } + + public static enum States { + S0, S1, S11, S12, S2, S21, S211, S212 + } + + public static enum Events { + A, B, C, D, E, F, G, H, I + } + + 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); + } + } + } + + 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 TestListener extends StateMachineListenerAdapter { + + volatile CountDownLatch stateChangedLatch = new CountDownLatch(1); + volatile int stateChangedCount = 0; + + @Override + public void stateChanged(State from, State to) { + stateChangedCount++; + stateChangedLatch.countDown(); + } + + public void reset(int c1) { + stateChangedLatch = new CountDownLatch(c1); + stateChangedCount = 0; + } + + } + + private static class TestStateChangeInterceptor implements StateChangeInterceptor { + + volatile CountDownLatch preStateChangeLatch = new CountDownLatch(1); + volatile int preStateChangeCount = 0; + + @Override + public void preStateChange(State state, Message message, + Transition transition, StateMachine stateMachine) { + preStateChangeCount++; + preStateChangeLatch.countDown(); + + } + + public void reset(int c1) { + preStateChangeLatch = new CountDownLatch(c1); + preStateChangeCount = 0; + } + + } + +} diff --git a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java index a270a51a..5fc4a3aa 100644 --- a/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java +++ b/spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java @@ -24,6 +24,7 @@ import org.springframework.statemachine.access.StateMachineAccess; import org.springframework.statemachine.access.StateMachineFunction; import org.springframework.statemachine.listener.AbstractCompositeListener; import org.springframework.statemachine.state.State; +import org.springframework.statemachine.support.DefaultStateMachineContext; import org.springframework.statemachine.support.LifecycleObjectSupport; import org.springframework.statemachine.support.StateChangeInterceptor; import org.springframework.statemachine.transition.Transition; @@ -73,7 +74,7 @@ public class PersistStateMachineHandler extends LifecycleObjectSupport { stateMachine.stop(); List> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions(); for (StateMachineAccess a : withAllRegions) { - a.resetState(state); + a.resetStateMachine(new DefaultStateMachineContext(state, null, null, null)); } stateMachine.start(); stateMachine.sendEvent(event); diff --git a/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java b/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java index 0c424e67..a6a2ee12 100644 --- a/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java +++ b/spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java @@ -26,6 +26,7 @@ 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.StateMachineEnsemble; import org.springframework.statemachine.zookeeper.ZookeeperStateMachineEnsemble; @Configuration @@ -41,7 +42,7 @@ public class Application { public void configure(StateMachineConfigurationConfigurer config) throws Exception { config .withDistributed() - .ensemble(new ZookeeperStateMachineEnsemble(curatorClient(), "/foo")); + .ensemble(stateMachineEnsemble()); } @Override @@ -68,6 +69,11 @@ public class Application { .event("PUSH"); } + @Bean + public StateMachineEnsemble stateMachineEnsemble() throws Exception { + return new ZookeeperStateMachineEnsemble(curatorClient(), "/foo"); + } + @Bean public CuratorFramework curatorClient() throws Exception { CuratorFramework client = CuratorFrameworkFactory.builder().defaultData(new byte[0]) 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 dfbb84ec..3f252976 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 @@ -24,6 +24,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.api.CuratorWatcher; +import org.apache.curator.framework.api.transaction.CuratorTransaction; +import org.apache.curator.framework.api.transaction.CuratorTransactionFinal; import org.apache.curator.framework.imps.CuratorFrameworkState; import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex; import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode; @@ -49,6 +51,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj private final static Log log = LogFactory.getLog(ZookeeperStateMachineEnsemble.class); private final String uuid = UUID.randomUUID().toString(); + private final static int DEFAULT_LOGSIZE = 32; private final static String PATH_CURRENT = "current"; private final static String PATH_LOG = "log"; private final static String PATH_MEMBERS = "members"; @@ -57,6 +60,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj private final String baseDataPath; private final String statePath; private final String logPath; + private final int logSize; private final String memberPath; private final String mutexPath; private final boolean cleanState; @@ -72,7 +76,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj * @param basePath the base zookeeper path */ public ZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath) { - this(curatorClient, basePath, true); + this(curatorClient, basePath, true, DEFAULT_LOGSIZE); } /** @@ -81,16 +85,18 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj * @param curatorClient the curator client * @param basePath the base zookeeper path * @param cleanState if true clean existing state + * @param logSize the log size */ - public ZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath, boolean cleanState) { + public ZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath, boolean cleanState, int logSize) { this.curatorClient = curatorClient; this.cleanState = cleanState; + this.logSize = logSize; this.baseDataPath = basePath + "/data"; this.statePath = baseDataPath + "/" + PATH_CURRENT; this.logPath = baseDataPath + "/" + PATH_LOG; this.memberPath = basePath + "/" + PATH_MEMBERS; this.mutexPath = basePath + "/" + PATH_MUTEX; - this.persist = new ZookeeperStateMachinePersist(curatorClient, statePath); + this.persist = new ZookeeperStateMachinePersist(curatorClient, statePath, logPath, logSize); } @Override @@ -145,6 +151,10 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj public void setState(StateMachineContext context) { try { Stat stat = new Stat(); + StateWrapper stateWrapper = stateRef.get(); + if (stateWrapper != null) { + stat.setVersion(stateWrapper.version); + } persist.write(context, stat); stateRef.set(new StateWrapper(context, stat.getVersion())); } catch (Exception e) { @@ -155,6 +165,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj private StateWrapper readCurrentContext() { try { Stat stat = new Stat(); + // TODO: not nice that we need to set watcher here when persister is reading data curatorClient.getData().usingWatcher(watcher).forPath(statePath); StateMachineContext context = persist.read(stat); @@ -168,7 +179,13 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(curatorClient, mutexPath); try { + if (log.isTraceEnabled()) { + log.trace("About to acquire mutex"); + } mutex.acquire(); + if (log.isTraceEnabled()) { + log.trace("Mutex acquired"); + } if (cleanState) { if (curatorClient.checkExists().forPath(memberPath) != null) { @@ -184,14 +201,14 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj node.waitForInitialCreate(60, TimeUnit.SECONDS); if (curatorClient.checkExists().forPath(baseDataPath) == null) { - curatorClient.inTransaction() - .create().forPath(baseDataPath) - .and() - .create().forPath(statePath) - .and() - .create().forPath(logPath) - .and() - .commit(); + CuratorTransaction tx = curatorClient.inTransaction(); + CuratorTransactionFinal tt = tx.create().forPath(baseDataPath).and(); + tt = tt.create().forPath(statePath).and(); + tt = tt.create().forPath(logPath).and(); + for (int i = 0; i extends StateMachineEnsembleObj } finally { try { mutex.release(); + if (log.isTraceEnabled()) { + log.trace("Mutex released"); + } } catch (Exception e) { } } @@ -218,12 +238,17 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj if (log.isTraceEnabled()) { log.trace("NodeDataChanged currentStateWrapper=" + currentStateWrapper + " newStateWrapper=" + newStateWrapper); } + + // if we don't have a previous version, we've missed an update. + // we're going to replay those from log paths. if (currentStateWrapper.version + 1 == newStateWrapper.version && stateRef.compareAndSet(currentStateWrapper, newStateWrapper)) { if (log.isTraceEnabled()) { log.trace("Notify state change with new context"); } notifyStateChanged(newStateWrapper.context); + } else { + } break; default: 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 1f0d9612..27cf0116 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 @@ -26,6 +26,7 @@ import java.util.UUID; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.api.transaction.CuratorTransaction; +import org.apache.curator.framework.api.transaction.CuratorTransactionFinal; import org.apache.curator.framework.api.transaction.CuratorTransactionResult; import org.apache.zookeeper.data.Stat; import org.springframework.messaging.MessageHeaders; @@ -34,6 +35,7 @@ import org.springframework.statemachine.StateMachineException; import org.springframework.statemachine.ensemble.StateMachinePersist; import org.springframework.statemachine.support.DefaultExtendedState; import org.springframework.statemachine.support.DefaultStateMachineContext; +import org.springframework.util.Assert; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.Serializer; @@ -68,6 +70,8 @@ public class ZookeeperStateMachinePersist implements StateMachinePersist implements StateMachinePersist 0 && ((logSize & -logSize) == logSize), "Log size must be positive and power of two"); + } this.curatorClient = curatorClient; this.path = path; + this.logPath = logPath; + this.logSize = logSize; } @Override - public void write(org.springframework.statemachine.StateMachineContext context, Stat stat) { + public void write(StateMachineContext context, Stat stat) { byte[] data = serialize(context); CuratorTransaction tx = curatorClient.inTransaction(); try { - Collection results = tx.setData().forPath(path, data).and().commit(); + CuratorTransactionFinal tt = tx.setData().withVersion(stat.getVersion()).forPath(path, data).and(); + if (logPath != null) { + tt = tt.setData().forPath(logPath + "/" + stat.getVersion() % logSize, data).and(); + } + Collection results = tt.commit(); int version = results.iterator().next().getResultStat().getVersion(); stat.setVersion(version); } catch (Exception 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 b4edfe64..c53f9fb4 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 @@ -15,6 +15,7 @@ */ package org.springframework.statemachine.zookeeper; +import static org.hamcrest.Matchers.greaterThan; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; import static org.junit.Assert.assertThat; @@ -74,10 +75,12 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { ensemble.afterPropertiesSet(); assertThat(curatorClient.checkExists().forPath("/foo/data/current"), notNullValue()); + assertThat(curatorClient.getData().forPath("/foo/data/current").length, is(0)); ensemble.setState(new DefaultStateMachineContext("S1","E1", new HashMap(), new DefaultExtendedState())); - ensemble.setState(new DefaultStateMachineContext("S2","E1", new HashMap(), new DefaultExtendedState())); + assertThat(curatorClient.getData().forPath("/foo/data/current").length, greaterThan(0)); + ensemble.setState(new DefaultStateMachineContext("S2","E1", new HashMap(), new DefaultExtendedState())); } @Test @@ -136,6 +139,8 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { assertThat(curatorClient.getData().forPath("/foo/data/log").length, is(0)); } + // + @Override protected AnnotationConfigApplicationContext buildContext() { return new AnnotationConfigApplicationContext(); 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 bb5a734b..94736065 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 @@ -15,8 +15,8 @@ */ package org.springframework.statemachine.zookeeper; -import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.is; import static org.junit.Assert.assertThat; import java.util.concurrent.CountDownLatch; @@ -24,14 +24,19 @@ import java.util.concurrent.TimeUnit; import org.apache.curator.framework.CuratorFramework; import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +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.statemachine.StateMachine; import org.springframework.statemachine.config.EnableStateMachine; 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.listener.StateMachineListenerAdapter; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -45,7 +50,7 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { @Test @SuppressWarnings("unchecked") - public void testStateChanges() throws Exception { + public void testStateChangesManualSetup() throws Exception { context.register(ZkServerConfig.class, BaseConfig.class, Config1.class, Config2.class); context.refresh(); @@ -104,6 +109,67 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { assertThat(machine2.getState().getIds(), containsInAnyOrder("S2")); } + @Test + @SuppressWarnings("unchecked") + public void testLifecycle() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class, Config3.class, Config4.class); + context.refresh(); + + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + assertThat(((SmartLifecycle)machine1).isAutoStartup(), is(true)); + assertThat(((SmartLifecycle)machine1).isRunning(), is(true)); + assertThat(((SmartLifecycle)machine2).isAutoStartup(), is(true)); + assertThat(((SmartLifecycle)machine2).isRunning(), is(true)); + } + + @Test + @SuppressWarnings("unchecked") + public void testStateChangesConfigSetup() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class, Config3.class, Config4.class); + context.refresh(); + + StateMachine machine1 = + context.getBean("sm1", StateMachine.class); + StateMachine machine2 = + context.getBean("sm2", StateMachine.class); + + TestListener listener1 = + context.getBean("listener1", TestListener.class); + TestListener listener2 = + context.getBean("listener2", TestListener.class); + + assertThat(listener1.stateMachineStartedLatch.await(1, TimeUnit.SECONDS), is(true)); + assertThat(listener2.stateMachineStartedLatch.await(1, TimeUnit.SECONDS), is(true)); + + assertThat(machine1.getState().getIds(), containsInAnyOrder("SI")); + assertThat(machine2.getState().getIds(), containsInAnyOrder("SI")); + + listener1.reset(1); + listener2.reset(1); + machine1.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); + machine1.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 testJoinLaterShouldSyncState() throws Exception { @@ -155,15 +221,73 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { @Configuration @EnableStateMachine(name = "sm1") - static class Config1 extends SharedConfig { + static class Config1 extends SharedConfig1 { } @Configuration @EnableStateMachine(name = "sm2") - static class Config2 extends SharedConfig { + static class Config2 extends SharedConfig1 { } - static class SharedConfig extends StateMachineConfigurerAdapter { + @Configuration + @EnableStateMachine(name = "sm1") + static class Config3 extends SharedConfig2 { + + @Autowired + private CuratorFramework curatorClient; + + @Override + @Bean(name = "listener1") + public TestListener stateMachineListener() { + return new TestListener(); + } + + @Override + @Bean(name = "ensemble1") + public StateMachineEnsemble stateMachineEnsemble() throws Exception { + return new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + } + } + + @Configuration + @EnableStateMachine(name = "sm2") + static class Config4 extends SharedConfig2 { + + @Autowired + private CuratorFramework curatorClient; + + @Override + @Bean(name = "listener2") + public TestListener stateMachineListener() { + return new TestListener(); + } + + @Override + @Bean(name = "ensemble2") + public StateMachineEnsemble stateMachineEnsemble() throws Exception { + return new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + } + } + + abstract static class SharedConfig2 extends SharedConfig1 { + + @Override + public void configure(StateMachineConfigurationConfigurer config) throws Exception { + config + .withDistributed() + .ensemble(stateMachineEnsemble()) + .and() + .withConfiguration() + .listener(stateMachineListener()) + .autoStartup(true); + } + + public abstract StateMachineEnsemble stateMachineEnsemble() throws Exception; + public abstract TestListener stateMachineListener(); + + } + + static class SharedConfig1 extends StateMachineConfigurerAdapter { @Override public void configure(StateMachineStateConfigurer states) throws Exception { @@ -195,6 +319,12 @@ public class ZookeeperStateMachineTests extends AbstractZookeeperTests { volatile CountDownLatch stateChangedLatch = new CountDownLatch(1); volatile CountDownLatch transitionLatch = new CountDownLatch(0); volatile int stateChangedCount = 0; + volatile CountDownLatch stateMachineStartedLatch = new CountDownLatch(1); + + @Override + public void stateMachineStarted(StateMachine stateMachine) { + stateMachineStartedLatch.countDown(); + } @Override public void stateChanged(State from, State to) {