Change StateMachinePersist interface

- Remove away from plain serialize/deserialize to write/read
  concept which works better in a generic abstraction.
- Tweak related class in zookeeper package.
- Relates to #35

spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java

@@ -25,23 +25,28 @@ import org.springframework.statemachine.StateMachineContext;
  *
  * @param <S> the type of state
  * @param <E> the type of event
+ * @param <T> the type of context object
  */
-public interface StateMachinePersist<S, E> {
+public interface StateMachinePersist<S, E, T> {
 
 	/**
-	 * Serialize a {@link StateMachineContext}.
+	 * Write a {@link StateMachineContext} into a persistent store
+	 * with a context object {@code T}.
 	 *
-	 * @param context the state machine context
-	 * @return the serialized data
+	 * @param contect the contect
+	 * @param contextOjb the context ojb
+	 * @throws Exception the exception
 	 */
-	byte[] serialize(StateMachineContext<S, E> context);
+	void write(StateMachineContext<S, E> contect, T contextOjb) throws Exception;
 
 	/**
-	 * Deserialize a data into a {@link StateMachineContext}.
+	 * Read a {@link StateMachineContext} from a persistent store
+	 * with a context object {@code T}.
 	 *
-	 * @param data the data
+	 * @param contextOjb the context ojb
 	 * @return the state machine context
+	 * @throws Exception the exception
 	 */
-	StateMachineContext<S, E> deserialize(byte[] data);
+	StateMachineContext<S, E> read(T contextOjb) throws Exception;
 
 }

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

@@ -16,7 +16,6 @@
 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;
@@ -25,8 +24,6 @@ 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.CuratorTransactionResult;
 import org.apache.curator.framework.imps.CuratorFrameworkState;
 import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
 import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
@@ -63,7 +60,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	private final String memberPath;
 	private final String mutexPath;
 	private final boolean cleanState;
-	private final StateMachinePersist<S, E> persist = new KryoStateMachinePersist<S, E>();
+	private final StateMachinePersist<S, E, Stat> persist;
 	private final AtomicReference<StateWrapper> stateRef = new AtomicReference<StateWrapper>();
 	private final CuratorWatcher watcher = new StateWatcher();
 	private PersistentEphemeralNode node;
@@ -93,6 +90,7 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 		this.logPath = baseDataPath + "/" + PATH_LOG;
 		this.memberPath = basePath + "/" + PATH_MEMBERS;
 		this.mutexPath = basePath + "/" + PATH_MUTEX;
+		this.persist = new ZookeeperStateMachinePersist<S, E>(curatorClient, statePath);
 	}
 
 	@Override
@@ -145,12 +143,10 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 
 	@Override
 	public void setState(StateMachineContext<S, E> context) {
-		byte[] data = persist.serialize(context);
-		CuratorTransaction tx = curatorClient.inTransaction();
 		try {
-			Collection<CuratorTransactionResult> results = tx.setData().forPath(statePath, data).and().commit();
-			int version = results.iterator().next().getResultStat().getVersion();
-			stateRef.set(new StateWrapper(context, version));
+			Stat stat = new Stat();
+			persist.write(context, stat);
+			stateRef.set(new StateWrapper(context, stat.getVersion()));
 		} catch (Exception e) {
 			throw new StateMachineException("Error persisting data", e);
 		}
@@ -159,8 +155,9 @@ public class ZookeeperStateMachineEnsemble<S, E> extends StateMachineEnsembleObj
 	private StateWrapper readCurrentContext() {
 		try {
 			Stat stat = new Stat();
-			byte[] data = curatorClient.getData().storingStatIn(stat).usingWatcher(watcher).forPath(statePath);
-			StateMachineContext<S, E> context = persist.deserialize(data);
+			// 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);
 			return new StateWrapper(context, stat.getVersion());
 		} catch (Exception e) {
 			throw new StateMachineException("Error reading data", e);

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

@@ -17,8 +17,14 @@ package org.springframework.statemachine.zookeeper;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.util.Collection;
 
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.transaction.CuratorTransaction;
+import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
+import org.apache.zookeeper.data.Stat;
 import org.springframework.statemachine.StateMachineContext;
+import org.springframework.statemachine.StateMachineException;
 import org.springframework.statemachine.ensemble.StateMachinePersist;
 import org.springframework.statemachine.support.DefaultStateMachineContext;
 
@@ -36,8 +42,10 @@ import com.esotericsoftware.kryo.io.Output;
  * @param <S> the type of state
  * @param <E> the type of event
  */
-public class KryoStateMachinePersist<S, E> implements StateMachinePersist<S, E> {
+public class ZookeeperStateMachinePersist<S, E> implements StateMachinePersist<S, E, Stat> {
 
+	// kryo is not a thread safe so using thread local, also
+	// adding custom serializer for state machine context.
 	private static final ThreadLocal<Kryo> kryoThreadLocal = new ThreadLocal<Kryo>() {
 
 		@SuppressWarnings("rawtypes")
@@ -49,8 +57,42 @@ public class KryoStateMachinePersist<S, E> implements StateMachinePersist<S, E>
 		}
 	};
 
+	private final CuratorFramework curatorClient;
+	private final String path;
+
+	/**
+	 * Instantiates a new zookeeper state machine persist.
+	 *
+	 * @param curatorClient the curator client
+	 * @param path the path for persistent state
+	 */
+	public ZookeeperStateMachinePersist(CuratorFramework curatorClient, String path) {
+		this.curatorClient = curatorClient;
+		this.path = path;
+	}
+
 	@Override
-	public byte[] serialize(StateMachineContext<S, E> context) {
+	public void write(org.springframework.statemachine.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();
+			int version = results.iterator().next().getResultStat().getVersion();
+			stat.setVersion(version);
+		} catch (Exception e) {
+			throw new StateMachineException("Error persisting data", e);
+		}
+
+	}
+
+	@Override
+	public StateMachineContext<S, E> read(Stat stat) throws Exception {
+		byte[] data = curatorClient.getData().storingStatIn(stat).forPath(path);
+		StateMachineContext<S, E> context = deserialize(data);
+		return context;
+	}
+
+	private byte[] serialize(StateMachineContext<S, E> context) {
 		Kryo kryo = kryoThreadLocal.get();
 		ByteArrayOutputStream out = new ByteArrayOutputStream();
 		Output output = new Output(out);
@@ -60,8 +102,7 @@ public class KryoStateMachinePersist<S, E> implements StateMachinePersist<S, E>
 	}
 
 	@SuppressWarnings("unchecked")
-	@Override
-	public StateMachineContext<S, E> deserialize(byte[] data) {
+	private StateMachineContext<S, E> deserialize(byte[] data) {
 		if (data == null || data.length == 0) {
 			return null;
 		}

spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java

@@ -18,20 +18,37 @@ package org.springframework.statemachine.zookeeper;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.zookeeper.data.Stat;
 import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.statemachine.StateMachineContext;
 import org.springframework.statemachine.ensemble.StateMachinePersist;
 import org.springframework.statemachine.support.DefaultStateMachineContext;
 
-public class KryoStateMachinePersistTests {
+public class ZookeeperStateMachinePersistTests extends AbstractZookeeperTests {
+
+	@Override
+	protected AnnotationConfigApplicationContext buildContext() {
+		return new AnnotationConfigApplicationContext();
+	}
 
 	@Test
-	public void testStateEvent() {
-		StateMachinePersist<String, String> persist = new KryoStateMachinePersist<String, String>();
+	public void testStateEvent() throws Exception {
+		context.register(ZkServerConfig.class, BaseConfig.class);
+		context.refresh();
+
+		CuratorFramework curatorClient =
+				context.getBean("curatorClient", CuratorFramework.class);
+		curatorClient.create().forPath("/KryoStateMachinePersistTests");
+
+		StateMachinePersist<String, String, Stat> persist = new ZookeeperStateMachinePersist<String, String>(
+				curatorClient, "/KryoStateMachinePersistTests");
+
 		StateMachineContext<String, String> contextOut =
 				new DefaultStateMachineContext<String, String>(null, "S1", "E1", null, null);
-		byte[] data = persist.serialize(contextOut);
-		StateMachineContext<String, String> contextIn = persist.deserialize(data);
+		persist.write(contextOut, new Stat());
+		StateMachineContext<String, String> contextIn = persist.read(new Stat());
 
 		assertThat(contextOut.getState(), is(contextIn.getState()));
 		assertThat(contextOut.getEvent(), is(contextIn.getEvent()));