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 854cdccf..400b8bea 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 @@ -23,17 +23,16 @@ import org.apache.commons.logging.LogFactory; import org.springframework.messaging.Message; 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; import org.springframework.statemachine.listener.StateMachineListener; -import org.springframework.statemachine.listener.StateMachineListenerAdapter; 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; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -54,8 +53,8 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem private final String uuid = UUID.randomUUID().toString(); private final StateMachineEnsemble ensemble; private final StateMachine delegate; - private final LocalEnsembleListener listener; - private final LocalStateMachineListener stateMachineListener; + private final LocalEnsembleListener listener = new LocalEnsembleListener(); + private final LocalStateChangeInterceptor interceptor = new LocalStateChangeInterceptor(); /** * Instantiates a new distributed state machine. @@ -68,26 +67,30 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem Assert.notNull(delegate, "State machine delegate must be set"); this.ensemble = ensemble; this.delegate = delegate; - this.listener = new LocalEnsembleListener(); - this.stateMachineListener = new LocalStateMachineListener(); } @Override protected void onInit() throws Exception { - super.onInit(); + delegate.getStateMachineAccessor().doWithAllRegions(new StateMachineFunction>() { + + @Override + public void apply(StateMachineAccess function) { + function.addStateChangeInterceptor(interceptor); + } + }); } @Override protected void doStart() { ensemble.addEnsembleListener(listener); ensemble.join(this); - delegate.addStateListener(stateMachineListener); super.doStart(); } @Override protected void doStop() { ensemble.removeEnsembleListener(listener); + ensemble.leave(this); super.doStop(); } @@ -146,17 +149,27 @@ public class DistributedStateMachine extends LifecycleObjectSupport implem return delegate.getStateMachineAccessor(); } - private class LocalStateMachineListener extends StateMachineListenerAdapter { + /** + * We intercept state changes order to attempt to update global + * distributed state. This attempt is sent to an ensemble which will + * tell us if that attempt was successful. + */ + private class LocalStateChangeInterceptor implements StateChangeInterceptor { @Override - public void stateChanged(StateContext context) { - if (ObjectUtils.nullSafeEquals(uuid, context.getMessageHeader("uuid"))) { - ensemble.setState(new DefaultStateMachineContext(delegate, context.getTransition().getTarget() - .getId(), context.getEvent(), context.getMessageHeaders(), context.getExtendedState())); + public void preStateChange(State state, Message message, Transition transition, + StateMachine stateMachine) { + if (message != null && ObjectUtils.nullSafeEquals(uuid, message.getHeaders().get("uuid"))) { + ensemble.setState(new DefaultStateMachineContext(delegate, transition.getTarget() + .getId(), message.getPayload(), message.getHeaders(), stateMachine.getExtendedState())); } } + } + /** + * + */ private class LocalEnsembleListener implements EnsembleListeger { @Override diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java index f937c297..b6e37bce 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/CompositeStateMachineListener.java @@ -17,7 +17,6 @@ package org.springframework.statemachine.listener; import java.util.Iterator; -import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -41,14 +40,6 @@ public class CompositeStateMachineListener extends AbstractCompositeListene } } - @Override - public void stateChanged(StateContext context) { - for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { - StateMachineListener listener = iterator.next(); - listener.stateChanged(context); - } - } - @Override public void stateEntered(State state) { for (Iterator> iterator = getListeners().reverse(); iterator.hasNext();) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java index e95a3559..74038adf 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListener.java @@ -15,7 +15,6 @@ */ package org.springframework.statemachine.listener; -import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -38,13 +37,6 @@ public interface StateMachineListener { */ void stateChanged(State from, State to); - /** - * Notified when state is changed. - * - * @param context the state context - */ - void stateChanged(StateContext context); - /** * Notified when state is entered. * diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java index ce73e248..ba49ba3b 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/listener/StateMachineListenerAdapter.java @@ -15,7 +15,6 @@ */ package org.springframework.statemachine.listener; -import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.state.State; import org.springframework.statemachine.transition.Transition; @@ -35,10 +34,6 @@ public class StateMachineListenerAdapter implements StateMachineListener from, State to) { } - @Override - public void stateChanged(StateContext context) { - } - @Override public void stateEntered(State state) { } 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 de1e6c9e..8e72733c 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 @@ -540,7 +540,6 @@ public abstract class AbstractStateMachine extends StateMachineObjectSuppo entryToState(state, message, transition, stateMachine); notifyStateChanged(notifyFrom, state); StateContext stateContext = buildStateContext(message, transition, stateMachine); - notifyStateChanged(stateContext); } else if (currentState != null) { if (findDeep != null) { if (exit) { diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java index 7f14a42b..a24e1a99 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/StateMachineObjectSupport.java @@ -24,7 +24,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.core.OrderComparator; import org.springframework.messaging.Message; -import org.springframework.statemachine.StateContext; import org.springframework.statemachine.StateMachine; import org.springframework.statemachine.event.StateMachineEventPublisher; import org.springframework.statemachine.listener.CompositeStateMachineListener; @@ -106,16 +105,6 @@ public abstract class StateMachineObjectSupport extends LifecycleObjectSup } } - protected void notifyStateChanged(StateContext context) { - stateListener.stateChanged(context); -// if (contextEventsEnabled) { -// StateMachineEventPublisher eventPublisher = getStateMachineEventPublisher(); -// if (eventPublisher != null) { -// eventPublisher.publishStateChanged(this, source, target); -// } -// } - } - protected void notifyStateEntered(State state) { stateListener.stateEntered(state); if (contextEventsEnabled) { @@ -218,11 +207,6 @@ public abstract class StateMachineObjectSupport extends LifecycleObjectSup stateChangedInRelay(); } - @Override - public void stateChanged(StateContext context) { - stateListener.stateChanged(context); - } - @Override public void stateEntered(State state) { stateListener.stateEntered(state); diff --git a/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java b/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java index 146cce2f..dcbcc752 100644 --- a/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java +++ b/spring-statemachine-core/src/test/java/org/springframework/statemachine/listener/ListenerTests.java @@ -131,10 +131,6 @@ public class ListenerTests extends AbstractStateMachineTests { states.add(new Holder(from, to)); } - @Override - public void stateChanged(StateContext context) { - } - @Override public void stateEntered(State state) { } 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 5b11be43..13511865 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 @@ -55,7 +55,9 @@ public class Application { return ensemble; } - @Bean(destroyMethod = "close") + // for now lets not close it here, we need to let + // some other framework, ie cloud, to create curator + @Bean//(destroyMethod = "close") public CuratorFramework curatorClient() throws Exception { CuratorFramework client = CuratorFrameworkFactory.builder().defaultData(new byte[0]) .retryPolicy(new ExponentialBackoffRetry(1000, 3)) 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 32282f29..ef9e9c2f 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 @@ -15,7 +15,10 @@ */ package org.springframework.statemachine.zookeeper; +import java.io.IOException; import java.util.Collection; +import java.util.UUID; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; @@ -24,7 +27,10 @@ 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.CuratorTransactionResult; -import org.apache.zookeeper.KeeperException; +import org.apache.curator.framework.imps.CuratorFrameworkState; +import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex; +import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode; +import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode.Mode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.data.Stat; import org.springframework.statemachine.StateMachine; @@ -45,13 +51,22 @@ import org.springframework.statemachine.ensemble.StateMachinePersist; public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObjectSupport { private final static Log log = LogFactory.getLog(ZookeeperStateMachineEnsemble.class); + private final String uuid = UUID.randomUUID().toString(); + private final static String PATH_CURRENT = "current"; + private final static String PATH_LOG = "log"; + private final static String PATH_MEMBERS = "members"; + private final static String PATH_MUTEX = "mutex"; private final CuratorFramework curatorClient; - private final String basePath; + private final String baseDataPath; private final String statePath; private final String logPath; + private final String memberPath; + private final String mutexPath; + private final boolean cleanState; private final StateMachinePersist persist = new KryoStateMachinePersist(); private final AtomicReference stateRef = new AtomicReference(); private final CuratorWatcher watcher = new StateWatcher(); + private PersistentEphemeralNode node; /** * Instantiates a new zookeeper state machine ensemble. @@ -60,10 +75,24 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj * @param basePath the base zookeeper path */ public ZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath) { + this(curatorClient, basePath, true); + } + + /** + * Instantiates a new zookeeper state machine ensemble. + * + * @param curatorClient the curator client + * @param basePath the base zookeeper path + * @param cleanState if true clean existing state + */ + public ZookeeperStateMachineEnsemble(CuratorFramework curatorClient, String basePath, boolean cleanState) { this.curatorClient = curatorClient; - this.basePath = basePath; - this.statePath = basePath + "/current"; - this.logPath = basePath + "/log"; + this.cleanState = cleanState; + this.baseDataPath = basePath + "/data"; + this.statePath = baseDataPath + "/" + PATH_CURRENT; + this.logPath = baseDataPath + "/" + PATH_LOG; + this.memberPath = basePath + "/" + PATH_MEMBERS; + this.mutexPath = basePath + "/" + PATH_MUTEX; } @Override @@ -75,6 +104,18 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj protected void doStart() { } + @Override + protected void doStop() { + if (node != null && curatorClient.getState() != CuratorFrameworkState.STOPPED) { + try { + node.close(); + } catch (IOException e) { + } finally { + node = null; + } + } + } + @Override public void join(StateMachine stateMachine) { StateWrapper stateWrapper = stateRef.get(); @@ -92,6 +133,14 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj @Override public void leave(StateMachine stateMachine) { + if (node != null) { + try { + node.close(); + } catch (IOException e) { + } + } + StateWrapper stateWrapper = stateRef.get(); + notifyLeft(stateWrapper != null ? stateWrapper.context : null); } @Override @@ -119,10 +168,27 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj } private void initPaths() { + + InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(curatorClient, mutexPath); try { - if (curatorClient.checkExists().forPath(statePath) == null) { + mutex.acquire(); + + if (cleanState) { + if (curatorClient.checkExists().forPath(memberPath) != null) { + if (curatorClient.getChildren().forPath(memberPath).size() == 0) { + log.info("Deleting from " + baseDataPath); + curatorClient.delete().deletingChildrenIfNeeded().forPath(baseDataPath); + } + } + } + + node = new PersistentEphemeralNode(curatorClient, Mode.EPHEMERAL, memberPath + "/" + uuid, new byte[0]); + node.start(); + node.waitForInitialCreate(60, TimeUnit.SECONDS); + + if (curatorClient.checkExists().forPath(baseDataPath) == null) { curatorClient.inTransaction() - .create().forPath(basePath) + .create().forPath(baseDataPath) .and() .create().forPath(statePath) .and() @@ -130,10 +196,14 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj .and() .commit(); } - } catch (KeeperException.NodeExistsException e) { - // ignore, already created + } catch (Exception e) { - throw new RuntimeException(e); + log.warn("Error in initPaths", e); + } finally { + try { + mutex.release(); + } 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 c6a3d1fa..e7e733cf 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 @@ -52,8 +52,8 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { ensemble.afterPropertiesSet(); - assertThat(curatorClient.checkExists().forPath("/foo/current"), notNullValue()); - assertThat(curatorClient.checkExists().forPath("/foo/log"), notNullValue()); + assertThat(curatorClient.checkExists().forPath("/foo/data/current"), notNullValue()); + assertThat(curatorClient.checkExists().forPath("/foo/data/log"), notNullValue()); ensemble.start(); } @@ -71,7 +71,7 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { ensemble.afterPropertiesSet(); - assertThat(curatorClient.checkExists().forPath("/foo/current"), notNullValue()); + assertThat(curatorClient.checkExists().forPath("/foo/data/current"), notNullValue()); ensemble.setState(new DefaultStateMachineContext(null, "S1","E1", null, null)); ensemble.setState(new DefaultStateMachineContext(null, "S2","E1", null, null)); @@ -113,6 +113,27 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { assertThat(listener2.eventLatch.await(2, TimeUnit.SECONDS), is(true)); } + @Test + public void testClearExistingStatePaths() throws Exception { + context.register(ZkServerConfig.class, BaseConfig.class); + context.refresh(); + + CuratorFramework curatorClient = + context.getBean("curatorClient", CuratorFramework.class); + + // members base path need to exist for delete to happen + curatorClient.create().creatingParentsIfNeeded().forPath("/foo/members"); + curatorClient.create().creatingParentsIfNeeded().forPath("/foo/data/log", new byte[10]); + + ZookeeperStateMachineEnsemble ensemble1 = + new ZookeeperStateMachineEnsemble(curatorClient, "/foo"); + ensemble1.afterPropertiesSet(); + ensemble1.start(); + + // we assume that if data is 0, it's re-created + 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/resources/log4j.properties b/spring-statemachine-zookeeper/src/test/resources/log4j.properties index 72283d00..4589b1e9 100644 --- a/spring-statemachine-zookeeper/src/test/resources/log4j.properties +++ b/spring-statemachine-zookeeper/src/test/resources/log4j.properties @@ -4,5 +4,5 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2} [%t] - %m%n -log4j.category.org.springframework.statemachine.zookeeper=TRACE +log4j.category.org.springframework.statemachine=TRACE