Change distributed machine to use interceptor

- Move away from using state machine listener if favor
  of interceptor which allows to skip state change.
- Remove previously added method in a
  StateMachineListener. (was a bad idea).
- More tweaks to zk example
- Relates to #35

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<S, E> extends LifecycleObjectSupport implem
 	private final String uuid = UUID.randomUUID().toString();
 	private final StateMachineEnsemble<S, E> ensemble;
 	private final StateMachine<S, E> 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<S, E> 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<StateMachineAccess<S, E>>() {
+
+			@Override
+			public void apply(StateMachineAccess<S, E> 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<S, E> extends LifecycleObjectSupport implem
 		return delegate.getStateMachineAccessor();
 	}
 
-	private class LocalStateMachineListener extends StateMachineListenerAdapter<S, E> {
+	/**
+	 * 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<S, E> {
 
 		@Override
-		public void stateChanged(StateContext<S, E> context) {
-			if (ObjectUtils.nullSafeEquals(uuid, context.getMessageHeader("uuid"))) {
-				ensemble.setState(new DefaultStateMachineContext<S, E>(delegate, context.getTransition().getTarget()
-						.getId(), context.getEvent(), context.getMessageHeaders(), context.getExtendedState()));
+		public void preStateChange(State<S, E> state, Message<E> message, Transition<S, E> transition,
+				StateMachine<S, E> stateMachine) {
+			if (message != null && ObjectUtils.nullSafeEquals(uuid, message.getHeaders().get("uuid"))) {
+				ensemble.setState(new DefaultStateMachineContext<S, E>(delegate, transition.getTarget()
+						.getId(), message.getPayload(), message.getHeaders(), stateMachine.getExtendedState()));
 			}
 		}
+
 	}
 
+	/**
+	 *
+	 */
 	private class LocalEnsembleListener implements EnsembleListeger<S, E> {
 
 		@Override

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<S,E> extends AbstractCompositeListene
 		}
 	}
 
-	@Override
-	public void stateChanged(StateContext<S, E> context) {
-		for (Iterator<StateMachineListener<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {
-			StateMachineListener<S, E> listener = iterator.next();
-			listener.stateChanged(context);
-		}
-	}
-
 	@Override
 	public void stateEntered(State<S, E> state) {
 		for (Iterator<StateMachineListener<S, E>> iterator = getListeners().reverse(); iterator.hasNext();) {

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<S,E> {
 	 */
 	void stateChanged(State<S,E> from, State<S,E> to);
 
-	/**
-	 * Notified when state is changed.
-	 *
-	 * @param context the state context
-	 */
-	void stateChanged(StateContext<S, E> context);
-
 	/**
 	 * Notified when state is entered.
 	 *

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<S, E> implements StateMachineListener<S
 	public void stateChanged(State<S, E> from, State<S, E> to) {
 	}
 
-	@Override
-	public void stateChanged(StateContext<S, E> context) {
-	}
-
 	@Override
 	public void stateEntered(State<S, E> state) {
 	}

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

@@ -540,7 +540,6 @@ public abstract class AbstractStateMachine<S, E> extends StateMachineObjectSuppo
 			entryToState(state, message, transition, stateMachine);
 			notifyStateChanged(notifyFrom, state);
 			StateContext<S, E> stateContext = buildStateContext(message, transition, stateMachine);
-			notifyStateChanged(stateContext);
 		} else if (currentState != null) {
 			if (findDeep != null) {
 				if (exit) {

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<S, E> extends LifecycleObjectSup
 		}
 	}
 
-	protected void notifyStateChanged(StateContext<S, E> context) {
-		stateListener.stateChanged(context);
-//		if (contextEventsEnabled) {
-//			StateMachineEventPublisher eventPublisher = getStateMachineEventPublisher();
-//			if (eventPublisher != null) {
-//				eventPublisher.publishStateChanged(this, source, target);
-//			}
-//		}
-	}
-
 	protected void notifyStateEntered(State<S,E> state) {
 		stateListener.stateEntered(state);
 		if (contextEventsEnabled) {
@@ -218,11 +207,6 @@ public abstract class StateMachineObjectSupport<S, E> extends LifecycleObjectSup
 			stateChangedInRelay();
 		}
 
-		@Override
-		public void stateChanged(StateContext<S, E> context) {
-			stateListener.stateChanged(context);
-		}
-
 		@Override
 		public void stateEntered(State<S, E> state) {
 			stateListener.stateEntered(state);

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<TestStates, TestEvents> context) {
-		}
-
 		@Override
 		public void stateEntered(State<TestStates, TestEvents> state) {
 		}

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))

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<S, E> extends StateMachineEnsembleObjectSupport<S, E> {
 
 	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<S, E> persist = new KryoStateMachinePersist<S, E>();
 	private final AtomicReference<StateWrapper> stateRef = new AtomicReference<StateWrapper>();
 	private final CuratorWatcher watcher = new StateWatcher();
+	private PersistentEphemeralNode node;
 
 	/**
 	 * Instantiates a new zookeeper state machine ensemble.
@@ -60,10 +75,24 @@ public class ZookeeperStateMachineEnsemble<S, E> 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<S, E> 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<S, E> stateMachine) {
 		StateWrapper stateWrapper = stateRef.get();
@@ -92,6 +133,14 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 
 	@Override
 	public void leave(StateMachine<S, E> 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<S, E> 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<S, E> 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) {
+			}
 		}
 	}
 

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<String, String>(null, "S1","E1", null, null));
 		ensemble.setState(new DefaultStateMachineContext<String, String>(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<String, String> ensemble1 =
+				new ZookeeperStateMachineEnsemble<String, String>(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();

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