Spring/spring-statemachine
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Tweak concepts around state reset

Browse Source 
- 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
This commit is contained in:
Janne Valkealahti
2015-07-15 11:17:43 +01:00
parent db5a1ab3f7
commit 15c6e507d8
14 changed files with 841 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-statemachine-core/src/main/java/org/springframework/statemachine/access/StateMachineAccess.java
View File

@@ -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<S, E> {
void setRelay(StateMachine<S, E> 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<S, E> stateMachineContext);
/**
* Adds the state change interceptor.

1
spring-statemachine-core/src/main/java/org/springframework/statemachine/config/configurers/ConfigurationConfigurer.java
View File

@@ -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;

13
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/DistributedStateMachine.java
View File

@@ -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 <S> the type of state
@@ -168,7 +171,8 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
}
/**
*
* Bridge for instructing delegating machine based on what
* is happening in an ensemble.
*/
private class LocalEnsembleListener implements EnsembleListeger<S, E> {
@@ -178,12 +182,15 @@ public class DistributedStateMachine<S, E> 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<StateMachineAccess<S, E>>() {
@Override
public void apply(StateMachineAccess<S, E> function) {
function.resetState(context.getState());
function.setExtendedState(context.getExtendedState());
function.resetStateMachine(context);
}
});

38
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/AbstractStateMachine.java
View File

@@ -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<S, E> extends StateMachineObjectSuppo
return extendedState;
}
@Override
public void setExtendedState(ExtendedState extendedState) {
this.extendedState = extendedState;
}
public void setHistoryState(PseudoState<S, E> history) {
this.history = history;
}
@@ -422,13 +418,30 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
}
@Override
public void resetState(S state) {
public void resetStateMachine(StateMachineContext<S, E> stateMachineContext) {
S state = stateMachineContext.getState();
boolean stateSet = false;
for (State<S, E> s : getStates()) {
if (s.getId().equals(state)) {
currentState = s;
for (State<S, E> 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<S, E> submachine = ((AbstractState<S, E>)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<S, E> extends StateMachineObjectSuppo
return true;
}
private boolean isInitialTransition(Transition<S,E> transition) {
return transition != null && transition.getKind() == TransitionKind.INITIAL;
}
private void switchToState(State<S,E> state, Message<E> message, Transition<S,E> transition, StateMachine<S, E> 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<S, E> extends StateMachineObjectSuppo
if (isTargetSubOf && currentState == transition.getTarget()) {
state = transition.getSource();
}
// else if (currentState == null && StateMachineUtils.isSubstate(findDeep, state)) {
// state = findDeep;
// }
}
boolean nonDeepStatePresent = false;

9
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineContext.java
View File

@@ -59,7 +59,8 @@ public class DefaultStateMachineContext<S, E> implements StateMachineContext<S,
* @param eventHeaders the event headers
* @param extendedState the extended state
*/
public DefaultStateMachineContext(List<StateMachineContext<S, E>> childs, S state, E event, Map<String, Object> eventHeaders, ExtendedState extendedState) {
public DefaultStateMachineContext(List<StateMachineContext<S, E>> childs, S state, E event,
Map<String, Object> eventHeaders, ExtendedState extendedState) {
this.childs = childs;
this.state = state;
this.event = event;
@@ -92,4 +93,10 @@ public class DefaultStateMachineContext<S, E> implements StateMachineContext<S,
return extendedState;
}
@Override
public String toString() {
return "DefaultStateMachineContext [state=" + state + ", event=" + event + ", eventHeaders=" + eventHeaders
+ ", extendedState=" + extendedState + "]";
}
}

289
spring-statemachine-core/src/test/java/org/springframework/statemachine/StateMachineResetTests.java Normal file
View File

@@ -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;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import java.util.HashMap;
import java.util.Map;
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.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.support.DefaultExtendedState;
import org.springframework.statemachine.support.DefaultStateMachineContext;
/**
* Tests for resetting a state machine state and extended variables using a
* {@link StateMachineContext}.
*
* @author Janne Valkealahti
*
*/
public class StateMachineResetTests extends AbstractStateMachineTests {
@Override
protected AnnotationConfigApplicationContext buildContext() {
return new AnnotationConfigApplicationContext();
}
@Test
public void testResetSubStates1() throws Exception {
context.register(Config1.class);
context.refresh();
@SuppressWarnings("unchecked")
StateMachine<States, Events> machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
Map<Object, Object> variables = new HashMap<Object, Object>();
variables.put("foo", 1);
ExtendedState extendedState = new DefaultExtendedState(variables);
DefaultStateMachineContext<States,Events> stateMachineContext = new DefaultStateMachineContext<States, Events>(States.S12, Events.I, null, extendedState);
machine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction<StateMachineAccess<States,Events>>() {
@Override
public void apply(StateMachineAccess<States, Events> 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<States, Events> machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
Map<Object, Object> variables = new HashMap<Object, Object>();
variables.put("foo", 1);
ExtendedState extendedState = new DefaultExtendedState(variables);
DefaultStateMachineContext<States,Events> stateMachineContext = new DefaultStateMachineContext<States, Events>(States.S211, Events.C, null, extendedState);
machine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction<StateMachineAccess<States,Events>>() {
@Override
public void apply(StateMachineAccess<States, Events> 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<States, Events> machine = context.getBean(StateMachineSystemConstants.DEFAULT_ID_STATEMACHINE, StateMachine.class);
Map<Object, Object> variables = new HashMap<Object, Object>();
variables.put("foo", 1);
ExtendedState extendedState = new DefaultExtendedState(variables);
DefaultStateMachineContext<States,Events> stateMachineContext = new DefaultStateMachineContext<States, Events>(States.S2, Events.C, null, extendedState);
machine.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction<StateMachineAccess<States,Events>>() {
@Override
public void apply(StateMachineAccess<States, Events> 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<States, Events> {
@Override
public void configure(StateMachineStateConfigurer<States, Events> 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<States, Events> 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<States, Events> {
@Override
public void execute(StateContext<States, Events> context) {
Map<Object, Object> 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<States, Events> {
private final int match;
public FooGuard(int match) {
this.match = match;
}
@Override
public boolean evaluate(StateContext<States, Events> context) {
Object foo = context.getExtendedState().getVariables().get("foo");
return !(foo == null || !foo.equals(match));
}
}
}

7
spring-statemachine-core/src/test/java/org/springframework/statemachine/access/StateMachineAccessTests.java
View File

@@ -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<String, String> stateMachineContext) {
}
@Override

289
spring-statemachine-core/src/test/java/org/springframework/statemachine/support/StateChangeInterceptorTests.java Normal file
View File

@@ -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<States, Events> 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<StateMachineAccess<States, Events>>() {
@Override
public void apply(StateMachineAccess<States, Events> 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<States, Events> {
@Override
public void configure(StateMachineStateConfigurer<States, Events> 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<States, Events> 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<States, Events> {
@Override
public void execute(StateContext<States, Events> context) {
Map<Object, Object> 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<States, Events> {
private final int match;
public FooGuard(int match) {
this.match = match;
}
@Override
public boolean evaluate(StateContext<States, Events> context) {
Object foo = context.getExtendedState().getVariables().get("foo");
return !(foo == null || !foo.equals(match));
}
}
private static class TestListener extends StateMachineListenerAdapter<States, Events> {
volatile CountDownLatch stateChangedLatch = new CountDownLatch(1);
volatile int stateChangedCount = 0;
@Override
public void stateChanged(State<States, Events> from, State<States, Events> to) {
stateChangedCount++;
stateChangedLatch.countDown();
}
public void reset(int c1) {
stateChangedLatch = new CountDownLatch(c1);
stateChangedCount = 0;
}
}
private static class TestStateChangeInterceptor implements StateChangeInterceptor<States, Events> {
volatile CountDownLatch preStateChangeLatch = new CountDownLatch(1);
volatile int preStateChangeCount = 0;
@Override
public void preStateChange(State<States, Events> state, Message<Events> message,
Transition<States, Events> transition, StateMachine<States, Events> stateMachine) {
preStateChangeCount++;
preStateChangeLatch.countDown();
}
public void reset(int c1) {
preStateChangeLatch = new CountDownLatch(c1);
preStateChangeCount = 0;
}
}
}

3
spring-statemachine-recipes/src/main/java/org/springframework/statemachine/recipes/persist/PersistStateMachineHandler.java
View File

@@ -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<StateMachineAccess<String, String>> withAllRegions = stateMachine.getStateMachineAccessor().withAllRegions();
for (StateMachineAccess<String, String> a : withAllRegions) {
a.resetState(state);
a.resetStateMachine(new DefaultStateMachineContext<String, String>(state, null, null, null));
}
stateMachine.start();
stateMachine.sendEvent(event);

8
spring-statemachine-samples/zookeeper/src/main/java/demo/zookeeper/Application.java
View File

@@ -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<String, String> config) throws Exception {
config
.withDistributed()
.ensemble(new ZookeeperStateMachineEnsemble<String, String>(curatorClient(), "/foo"));
.ensemble(stateMachineEnsemble());
}
@Override
@@ -68,6 +69,11 @@ public class Application {
.event("PUSH");
}
@Bean
public StateMachineEnsemble<String, String> stateMachineEnsemble() throws Exception {
return new ZookeeperStateMachineEnsemble<String, String>(curatorClient(), "/foo");
}
@Bean
public CuratorFramework curatorClient() throws Exception {
CuratorFramework client = CuratorFrameworkFactory.builder().defaultData(new byte[0])

47
spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsemble.java
View File

@@ -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<S, E> 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<S, E> 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<S, E> 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<S, E> 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<S, E>(curatorClient, statePath);
this.persist = new ZookeeperStateMachinePersist<S, E>(curatorClient, statePath, logPath, logSize);
}
@Override
@@ -145,6 +151,10 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
public void setState(StateMachineContext<S, E> 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<S, E> 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<S, E> context = persist.read(stat);
@@ -168,7 +179,13 @@ public class ZookeeperStateMachineEnsemble<S, E> 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<S, E> 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<logSize; i++) {
tt = tt.create().forPath(logPath + "/" + i).and();
}
tt.commit();
}
} catch (Exception e) {
@@ -199,6 +216,9 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
} finally {
try {
mutex.release();
if (log.isTraceEnabled()) {
log.trace("Mutex released");
}
} catch (Exception e) {
}
}
@@ -218,12 +238,17 @@ public class ZookeeperStateMachineEnsemble<S, E> 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:

29
spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java
View File

@@ -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<S, E> implements StateMachinePersist<S
private final CuratorFramework curatorClient;
private final String path;
private final String logPath;
private final int logSize;
/**
* Instantiates a new zookeeper state machine persist.
@@ -76,16 +80,37 @@ public class ZookeeperStateMachinePersist<S, E> implements StateMachinePersist<S
* @param path the path for persistent state
*/
public ZookeeperStateMachinePersist(CuratorFramework curatorClient, String path) {
this(curatorClient, path, null, 0);
}
/**
* Instantiates a new zookeeper state machine persist.
*
* @param curatorClient the curator client
* @param path the path
* @param logPath the log path
* @param logSize the log size
*/
public ZookeeperStateMachinePersist(CuratorFramework curatorClient, String path, String logPath, int logSize) {
if (logPath != null) {
Assert.state(logSize > 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<S,E> context, Stat stat) {
public void write(StateMachineContext<S,E> context, Stat stat) {
byte[] data = serialize(context);
CuratorTransaction tx = curatorClient.inTransaction();
try {
Collection<CuratorTransactionResult> 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<CuratorTransactionResult> results = tt.commit();
int version = results.iterator().next().getResultStat().getVersion();
stat.setVersion(version);
} catch (Exception e) {

7
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java
View File

@@ -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<String, String>("S1","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
ensemble.setState(new DefaultStateMachineContext<String, String>("S2","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
assertThat(curatorClient.getData().forPath("/foo/data/current").length, greaterThan(0));
ensemble.setState(new DefaultStateMachineContext<String, String>("S2","E1", new HashMap<String, Object>(), 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();

140
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineTests.java
View File

@@ -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<String, String> machine1 =
context.getBean("sm1", StateMachine.class);
StateMachine<String, String> 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<String, String> machine1 =
context.getBean("sm1", StateMachine.class);
StateMachine<String, String> 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<String, String> {
@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<String, String> stateMachineEnsemble() throws Exception {
return new ZookeeperStateMachineEnsemble<String, String>(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<String, String> stateMachineEnsemble() throws Exception {
return new ZookeeperStateMachineEnsemble<String, String>(curatorClient, "/foo");
}
}
abstract static class SharedConfig2 extends SharedConfig1 {
@Override
public void configure(StateMachineConfigurationConfigurer<String, String> config) throws Exception {
config
.withDistributed()
.ensemble(stateMachineEnsemble())
.and()
.withConfiguration()
.listener(stateMachineListener())
.autoStartup(true);
}
public abstract StateMachineEnsemble<String, String> stateMachineEnsemble() throws Exception;
public abstract TestListener stateMachineListener();
}
static class SharedConfig1 extends StateMachineConfigurerAdapter<String, String> {
@Override
public void configure(StateMachineStateConfigurer<String, String> 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<String, String> stateMachine) {
stateMachineStartedLatch.countDown();
}
@Override
public void stateChanged(State<String, String> from, State<String, String> to) {