From 44d09f257e63166ba2c6c853aaea342145e6f123 Mon Sep 17 00:00:00 2001 From: Janne Valkealahti Date: Sat, 20 Jun 2015 15:41:35 +0100 Subject: [PATCH] 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 --- .../ensemble/StateMachinePersist.java | 21 +++++--- .../ZookeeperStateMachineEnsemble.java | 19 +++---- ...java => ZookeeperStateMachinePersist.java} | 49 +++++++++++++++++-- ...=> ZookeeperStateMachinePersistTests.java} | 27 ++++++++-- 4 files changed, 88 insertions(+), 28 deletions(-) rename spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/{KryoStateMachinePersist.java => ZookeeperStateMachinePersist.java} (61%) rename spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/{KryoStateMachinePersistTests.java => ZookeeperStateMachinePersistTests.java} (56%) diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java index e7e5f793..946ee886 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/ensemble/StateMachinePersist.java @@ -25,23 +25,28 @@ import org.springframework.statemachine.StateMachineContext; * * @param the type of state * @param the type of event + * @param the type of context object */ -public interface StateMachinePersist { +public interface StateMachinePersist { /** - * 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 context); + void write(StateMachineContext 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 deserialize(byte[] data); + StateMachineContext read(T contextOjb) throws Exception; } 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 ef9e9c2f..dfbb84ec 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,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 extends StateMachineEnsembleObj private final String memberPath; private final String mutexPath; private final boolean cleanState; - private final StateMachinePersist persist = new KryoStateMachinePersist(); + private final StateMachinePersist persist; private final AtomicReference stateRef = new AtomicReference(); private final CuratorWatcher watcher = new StateWatcher(); private PersistentEphemeralNode node; @@ -93,6 +90,7 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj this.logPath = baseDataPath + "/" + PATH_LOG; this.memberPath = basePath + "/" + PATH_MEMBERS; this.mutexPath = basePath + "/" + PATH_MUTEX; + this.persist = new ZookeeperStateMachinePersist(curatorClient, statePath); } @Override @@ -145,12 +143,10 @@ public class ZookeeperStateMachineEnsemble extends StateMachineEnsembleObj @Override public void setState(StateMachineContext context) { - byte[] data = persist.serialize(context); - CuratorTransaction tx = curatorClient.inTransaction(); try { - Collection 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 extends StateMachineEnsembleObj private StateWrapper readCurrentContext() { try { Stat stat = new Stat(); - byte[] data = curatorClient.getData().storingStatIn(stat).usingWatcher(watcher).forPath(statePath); - StateMachineContext 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 context = persist.read(stat); return new StateWrapper(context, stat.getVersion()); } catch (Exception e) { throw new StateMachineException("Error reading data", e); diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java similarity index 61% rename from spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java rename to spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java index 6d028dc3..0b1be583 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersist.java +++ b/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 the type of state * @param the type of event */ -public class KryoStateMachinePersist implements StateMachinePersist { +public class ZookeeperStateMachinePersist implements StateMachinePersist { + // kryo is not a thread safe so using thread local, also + // adding custom serializer for state machine context. private static final ThreadLocal kryoThreadLocal = new ThreadLocal() { @SuppressWarnings("rawtypes") @@ -49,8 +57,42 @@ public class KryoStateMachinePersist implements StateMachinePersist } }; + 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 context) { + public void write(org.springframework.statemachine.StateMachineContext context, Stat stat) { + byte[] data = serialize(context); + CuratorTransaction tx = curatorClient.inTransaction(); + try { + Collection 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 read(Stat stat) throws Exception { + byte[] data = curatorClient.getData().storingStatIn(stat).forPath(path); + StateMachineContext context = deserialize(data); + return context; + } + + private byte[] serialize(StateMachineContext context) { Kryo kryo = kryoThreadLocal.get(); ByteArrayOutputStream out = new ByteArrayOutputStream(); Output output = new Output(out); @@ -60,8 +102,7 @@ public class KryoStateMachinePersist implements StateMachinePersist } @SuppressWarnings("unchecked") - @Override - public StateMachineContext deserialize(byte[] data) { + private StateMachineContext deserialize(byte[] data) { if (data == null || data.length == 0) { return null; } diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java similarity index 56% rename from spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java rename to spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java index a52662a4..fa8cc719 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/KryoStateMachinePersistTests.java +++ b/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 persist = new KryoStateMachinePersist(); + 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 persist = new ZookeeperStateMachinePersist( + curatorClient, "/KryoStateMachinePersistTests"); + StateMachineContext contextOut = new DefaultStateMachineContext(null, "S1", "E1", null, null); - byte[] data = persist.serialize(contextOut); - StateMachineContext contextIn = persist.deserialize(data); + persist.write(contextOut, new Stat()); + StateMachineContext contextIn = persist.read(new Stat()); assertThat(contextOut.getState(), is(contextIn.getState())); assertThat(contextOut.getEvent(), is(contextIn.getEvent()));