diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java index 9cc088ae..22ac3f1f 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/CompositeEnsembleListener.java @@ -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 extends AbstractCompositeListener { @Override - public void stateMachineJoined(StateMachineContext context) { + public void stateMachineJoined(StateMachine stateMachine, StateMachineContext context) { for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { EnsembleListeger listener = iterator.next(); - listener.stateMachineJoined(context); + listener.stateMachineJoined(stateMachine, context); } } @Override - public void stateMachineLeft(StateMachineContext context) { + public void stateMachineLeft(StateMachine stateMachine, StateMachineContext context) { for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { EnsembleListeger listener = iterator.next(); - listener.stateMachineLeft(context); + listener.stateMachineLeft(stateMachine, context); } } 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 452d0abe..48cc095f 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 @@ -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 extends LifecycleObjectSupport implem @Override public void preStateChange(State state, Message message, Transition transition, StateMachine 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 extends LifecycleObjectSupport implem @Override public StateContext postTransition(StateContext 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 xxx = ensemble.getState(); + ensemble.setState(new DefaultStateMachineContext( + xxx.getState(), stateContext.getEvent(), stateContext + .getMessageHeaders(), stateContext.getStateMachine().getExtendedState())); + } return stateContext; } @@ -207,37 +219,47 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem private class LocalEnsembleListener implements EnsembleListeger { @Override - public void stateMachineJoined(final StateMachineContext 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 stateMachine, final StateMachineContext 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>() { - - @Override - public void apply(StateMachineAccess function) { - function.resetStateMachine(context); + if (log.isDebugEnabled()) { + log.debug("Joining with context " + context); } - }); + delegate.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess 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 context) { - log.info("Requesting to stop delegating state machine " + delegate); - delegate.stop(); + public void stateMachineLeft(StateMachine stateMachine, StateMachineContext context) { + if (stateMachine != null && stateMachine == DistributedStateMachine.this) { + log.info("Requesting to stop delegating state machine " + delegate); + delegate.stop(); + } } @Override public void stateChanged(StateMachineContext 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()); + } } } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java index cd8adf8b..7b5b60d5 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/EnsembleListeger.java @@ -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 { /** * Called when state machine joined an ensemble. * + * @param stateMachine the state machine * @param context the state machine context */ - void stateMachineJoined(StateMachineContext context); + void stateMachineJoined(StateMachine stateMachine, StateMachineContext context); /** * Called when state machine left an ensemble. * + * @param stateMachine the state machine * @param context the state machine context */ - void stateMachineLeft(StateMachineContext context); + void stateMachineLeft(StateMachine stateMachine, StateMachineContext context); /** * Called when ensemble is discovering a state change. diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java index 913bae67..6bd8849b 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsemble.java @@ -64,6 +64,11 @@ public interface StateMachineEnsemble { */ void setState(StateMachineContext context); + /** + * Gets the state. + * + * @return the state + */ StateMachineContext getState(); } diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java index cc4fa758..cedfb99b 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachineEnsembleObjectSupport.java @@ -51,12 +51,12 @@ public abstract class StateMachineEnsembleObjectSupport extends LifecycleO ensembleListener.unregister(listener); } - protected void notifyJoined(StateMachineContext context) { - ensembleListener.stateMachineJoined(context); + protected void notifyJoined(StateMachine stateMachine, StateMachineContext context) { + ensembleListener.stateMachineJoined(stateMachine, context); } - protected void notifyLeft(StateMachineContext context) { - ensembleListener.stateMachineLeft(context); + protected void notifyLeft(StateMachine stateMachine, StateMachineContext context) { + ensembleListener.stateMachineLeft(stateMachine, context); } protected void notifyStateChanged(StateMachineContext context) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java index 154c3161..171a1320 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultStateMachineExecutor.java @@ -344,7 +344,11 @@ public class DefaultStateMachineExecutor extends LifecycleObjectSupport im MessageHeaders messageHeaders = message != null ? message.getHeaders() : new MessageHeaders( new HashMap()); Map map = new HashMap(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(event, new MessageHeaders(map), extendedState, transition, stateMachine); } diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java index 993fe9fe..2e278bf4 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/ensemble/InMemoryStateMachineEnsemble.java @@ -31,14 +31,14 @@ public class InMemoryStateMachineEnsemble extends StateMachineEnsembleObje public void join(StateMachine stateMachine) { if (!joined.contains(stateMachine)) { joined.add(stateMachine); - notifyJoined(current); + notifyJoined(stateMachine, current); } } @Override public void leave(StateMachine stateMachine) { if (joined.remove(stateMachine)) { - notifyLeft(current); + notifyLeft(stateMachine, current); } } 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 a70c3b20..104f3817 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 @@ -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 extends StateMachineEnsembleObj private final AtomicReference notifyRef = new AtomicReference(); private final CuratorWatcher watcher = new StateWatcher(); private PersistentEphemeralNode node; + private final Queue> joinQueue = new ConcurrentLinkedQueue>(); /** * Instantiates a new zookeeper state machine ensemble. @@ -116,12 +119,13 @@ public class ZookeeperStateMachineEnsemble 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 extends StateMachineEnsembleObj @Override public void join(StateMachine 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 stateMachine = null; + while ((stateMachine = joinQueue.poll()) != null) { + notifyJoined(stateMachine, stateWrapper != null ? stateWrapper.context : null); + } } @Override @@ -151,7 +167,7 @@ public class ZookeeperStateMachineEnsemble 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 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); 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 ef66f127..2600cec2 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 @@ -272,12 +272,12 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { volatile List> events = new ArrayList>(); @Override - public void stateMachineJoined(StateMachineContext context) { + public void stateMachineJoined(StateMachine stateMachine, StateMachineContext context) { joinedLatch.countDown(); } @Override - public void stateMachineLeft(StateMachineContext context) { + public void stateMachineLeft(StateMachine stateMachine, StateMachineContext context) { } @Override