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

Fix concepts around internal transitions

Browse Source 
- Fixes #86
- Fix internal transition changes when extended
  state variables has been modified. Also internal
  transition is now correctly passed into other
  distributed state machines.
This commit is contained in:
Janne Valkealahti
2015-07-24 09:07:03 +01:00
parent 550c17f5e4
commit 74f2f72ed6
9 changed files with 91 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java
View File

@@ -17,6 +17,7 @@ package org.springframework.statemachine.ensemble;
import java.util.Iterator;
import org.springframework.statemachine.StateMachine;
import org.springframework.statemachine.StateMachineContext;
import org.springframework.statemachine.listener.AbstractCompositeListener;
@@ -32,18 +33,18 @@ public class CompositeEnsembleListener<S, E> extends AbstractCompositeListener<E
EnsembleListeger<S, E> {
@Override
public void stateMachineJoined(StateMachineContext<S, E> context) {
public void stateMachineJoined(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context) {
for (Iterator<EnsembleListeger<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
EnsembleListeger<S, E> listener = iterator.next();
listener.stateMachineJoined(context);
listener.stateMachineJoined(stateMachine, context);
}
}
@Override
public void stateMachineLeft(StateMachineContext<S, E> context) {
public void stateMachineLeft(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context) {
for (Iterator<EnsembleListeger<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
EnsembleListeger<S, E> listener = iterator.next();
listener.stateMachineLeft(context);
listener.stateMachineLeft(stateMachine, context);
}
}

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

@@ -35,6 +35,7 @@ import org.springframework.statemachine.support.DefaultStateMachineContext;
import org.springframework.statemachine.support.LifecycleObjectSupport;
import org.springframework.statemachine.support.StateMachineInterceptor;
import org.springframework.statemachine.transition.Transition;
import org.springframework.statemachine.transition.TransitionKind;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -175,6 +176,7 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
@Override
public void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
StateMachine<S, E> stateMachine) {
// only handle if state change originates from this dist machine
if (message != null
&& ObjectUtils.nullSafeEquals(delegate.getId(),
message.getHeaders().get(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER))) {
@@ -195,6 +197,16 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
@Override
public StateContext<S, E> postTransition(StateContext<S, E> stateContext) {
// only handle if state change originates from this dist machine
if (stateContext.getTransition() != null
&& stateContext.getTransition().getKind() == TransitionKind.INTERNAL
&& ObjectUtils.nullSafeEquals(delegate.getId(),
stateContext.getMessageHeader(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER))) {
StateMachineContext<S, E> xxx = ensemble.getState();
ensemble.setState(new DefaultStateMachineContext<S, E>(
xxx.getState(), stateContext.getEvent(), stateContext
.getMessageHeaders(), stateContext.getStateMachine().getExtendedState()));
}
return stateContext;
}
@@ -207,37 +219,47 @@ public class DistributedStateMachine<S, E> extends LifecycleObjectSupport implem
private class LocalEnsembleListener implements EnsembleListeger<S, E> {
@Override
public void stateMachineJoined(final StateMachineContext<S, E> context) {
if (context != null) {
// I'm now successfully joined, so set delegating
// sm to current known state by a context.
public void stateMachineJoined(final StateMachine<S, E> stateMachine, final StateMachineContext<S, E> context) {
if (stateMachine != null && stateMachine == DistributedStateMachine.this) {
if (context != null) {
// 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.resetStateMachine(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.resetStateMachine(context);
}
});
}
log.info("Requesting to start delegating state machine " + delegate);
log.info("Delegating machine id " + delegate.getId());
delegate.start();
}
log.info("Requesting to start delegating state machine " + delegate);
delegate.start();
}
@Override
public void stateMachineLeft(StateMachineContext<S, E> context) {
log.info("Requesting to stop delegating state machine " + delegate);
delegate.stop();
public void stateMachineLeft(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context) {
if (stateMachine != null && stateMachine == DistributedStateMachine.this) {
log.info("Requesting to stop delegating state machine " + delegate);
delegate.stop();
}
}
@Override
public void stateChanged(StateMachineContext<S, E> context) {
delegate.sendEvent(MessageBuilder.withPayload(context.getEvent()).copyHeaders(context.getEventHeaders()).build());
// do not pass if state change was originated from this dist machine
if (!ObjectUtils.nullSafeEquals(delegate.getId(),
context.getEventHeaders().get(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER))) {
delegate.sendEvent(MessageBuilder.withPayload(context.getEvent())
.copyHeaders(context.getEventHeaders()).build());
}
}
}

7
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.statemachine.ensemble;
import org.springframework.statemachine.StateMachine;
import org.springframework.statemachine.StateMachineContext;
/**
@@ -30,16 +31,18 @@ public interface EnsembleListeger<S, E> {
/**
* Called when state machine joined an ensemble.
*
* @param stateMachine the state machine
* @param context the state machine context
*/
void stateMachineJoined(StateMachineContext<S, E> context);
void stateMachineJoined(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context);
/**
* Called when state machine left an ensemble.
*
* @param stateMachine the state machine
* @param context the state machine context
*/
void stateMachineLeft(StateMachineContext<S, E> context);
void stateMachineLeft(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context);
/**
* Called when ensemble is discovering a state change.

5
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java
View File

@@ -64,6 +64,11 @@ public interface StateMachineEnsemble<S, E> {
*/
void setState(StateMachineContext<S, E> context);
/**
* Gets the state.
*
* @return the state
*/
StateMachineContext<S, E> getState();
}

8
spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java
View File

@@ -51,12 +51,12 @@ public abstract class StateMachineEnsembleObjectSupport<S, E> extends LifecycleO
ensembleListener.unregister(listener);
}
protected void notifyJoined(StateMachineContext<S, E> context) {
ensembleListener.stateMachineJoined(context);
protected void notifyJoined(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context) {
ensembleListener.stateMachineJoined(stateMachine, context);
}
protected void notifyLeft(StateMachineContext<S, E> context) {
ensembleListener.stateMachineLeft(context);
protected void notifyLeft(StateMachine<S, E> stateMachine, StateMachineContext<S, E> context) {
ensembleListener.stateMachineLeft(stateMachine, context);
}
protected void notifyStateChanged(StateMachineContext<S, E> context) {

6
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java
View File

@@ -344,7 +344,11 @@ public class DefaultStateMachineExecutor<S, E> extends LifecycleObjectSupport im
MessageHeaders messageHeaders = message != null ? message.getHeaders() : new MessageHeaders(
new HashMap<String, Object>());
Map<String, Object> map = new HashMap<String, Object>(messageHeaders);
map.put(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER, stateMachine.getId());
if (!map.containsKey(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER)) {
// don't set sm id if it's already present because
// we want to keep the originating sm id
map.put(StateMachineSystemConstants.STATEMACHINE_IDENTIFIER, stateMachine.getId());
}
return new DefaultStateContext<S, E>(event, new MessageHeaders(map), extendedState, transition, stateMachine);
}

4
spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java
View File

@@ -31,14 +31,14 @@ public class InMemoryStateMachineEnsemble<S, E> extends StateMachineEnsembleObje
public void join(StateMachine<S, E> stateMachine) {
if (!joined.contains(stateMachine)) {
joined.add(stateMachine);
notifyJoined(current);
notifyJoined(stateMachine, current);
}
}
@Override
public void leave(StateMachine<S, E> stateMachine) {
if (joined.remove(stateMachine)) {
notifyLeft(current);
notifyLeft(stateMachine, current);
}
}

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

@@ -16,7 +16,9 @@
package org.springframework.statemachine.zookeeper;
import java.io.IOException;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -69,6 +71,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
private final AtomicReference<StateWrapper> notifyRef = new AtomicReference<StateWrapper>();
private final CuratorWatcher watcher = new StateWatcher();
private PersistentEphemeralNode node;
private final Queue<StateMachine<S, E>> joinQueue = new ConcurrentLinkedQueue<StateMachine<S, E>>();
/**
* Instantiates a new zookeeper state machine ensemble.
@@ -116,12 +119,13 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
if (stateWrapper == null) {
try {
StateWrapper currentStateWrapper = readCurrentContext();
stateRef.set(new StateWrapper(currentStateWrapper.context, currentStateWrapper.version));
stateWrapper = stateRef.get();
stateRef.set(currentStateWrapper);
notifyRef.set(currentStateWrapper);
} catch (Exception e) {
log.error("Error reading current state during start", e);
}
}
joinQueued();
}
@Override
@@ -138,8 +142,20 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
@Override
public void join(StateMachine<S, E> stateMachine) {
if (!isRunning()) {
joinQueue.add(stateMachine);
} else {
StateWrapper stateWrapper = stateRef.get();
notifyJoined(stateMachine, stateWrapper != null ? stateWrapper.context : null);
}
}
private void joinQueued() {
StateWrapper stateWrapper = stateRef.get();
notifyJoined(stateWrapper != null ? stateWrapper.context : null);
StateMachine<S, E> stateMachine = null;
while ((stateMachine = joinQueue.poll()) != null) {
notifyJoined(stateMachine, stateWrapper != null ? stateWrapper.context : null);
}
}
@Override
@@ -151,7 +167,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
}
}
StateWrapper stateWrapper = stateRef.get();
notifyLeft(stateWrapper != null ? stateWrapper.context : null);
notifyLeft(stateMachine, stateWrapper != null ? stateWrapper.context : null);
}
@Override
@@ -314,6 +330,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
int ver = (stat.getVersion() - 1) * logSize + (i + 1);
if (log.isDebugEnabled()) {
log.debug("Replay position " + i + " with version " + ver);
log.debug("Context in position " + i + " " + context);
}
StateWrapper wrapper = new StateWrapper(context, ver);
mayNotifyStateChanged(wrapper);

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

@@ -272,12 +272,12 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests {
volatile List<StateMachineContext<String, String>> events = new ArrayList<StateMachineContext<String,String>>();
@Override
public void stateMachineJoined(StateMachineContext<String, String> context) {
public void stateMachineJoined(StateMachine<String, String> stateMachine, StateMachineContext<String, String> context) {
joinedLatch.countDown();
}
@Override
public void stateMachineLeft(StateMachineContext<String, String> context) {
public void stateMachineLeft(StateMachine<String, String> stateMachine, StateMachineContext<String, String> context) {
}
@Override