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 946ee886..2bbcbf93 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 @@ -33,11 +33,11 @@ public interface StateMachinePersist { * Write a {@link StateMachineContext} into a persistent store * with a context object {@code T}. * - * @param contect the contect + * @param context the context * @param contextOjb the context ojb * @throws Exception the exception */ - void write(StateMachineContext contect, T contextOjb) throws Exception; + void write(StateMachineContext context, T contextOjb) throws Exception; /** * Read a {@link StateMachineContext} from a persistent store diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java index a072b416..5d3f1571 100644 --- a/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java @@ -37,6 +37,15 @@ public class DefaultExtendedState implements ExtendedState { this.variables = new HashMap(); } + /** + * Instantiates a new default extended state. + * + * @param variables the variables + */ + public DefaultExtendedState(Map variables) { + this.variables = variables; + } + @Override public Map getVariables() { return variables; diff --git a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java index de43665d..1f0d9612 100644 --- a/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java +++ b/spring-statemachine-zookeeper/src/main/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersist.java @@ -18,14 +18,21 @@ package org.springframework.statemachine.zookeeper; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.UUID; 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.messaging.MessageHeaders; import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.StateMachineException; import org.springframework.statemachine.ensemble.StateMachinePersist; +import org.springframework.statemachine.support.DefaultExtendedState; import org.springframework.statemachine.support.DefaultStateMachineContext; import com.esotericsoftware.kryo.Kryo; @@ -34,8 +41,8 @@ import com.esotericsoftware.kryo.io.Input; import com.esotericsoftware.kryo.io.Output; /** - * {@link StateMachinePersist} using kroy libraries as a backing - * serialization technique. + * {@link StateMachinePersist} using zookeeper as a storage and + * kroy libraries as a backing serialization technique. * * @author Janne Valkealahti * @@ -53,6 +60,8 @@ public class ZookeeperStateMachinePersist implements StateMachinePersist implements StateMachinePersist context) { kryo.writeClassAndObject(output, context.getEvent()); kryo.writeClassAndObject(output, context.getState()); + kryo.writeClassAndObject(output, context.getEventHeaders()); + kryo.writeClassAndObject(output, context.getExtendedState().getVariables()); + kryo.writeClassAndObject(output, context.getChilds()); } @SuppressWarnings("unchecked") @@ -125,9 +137,50 @@ public class ZookeeperStateMachinePersist implements StateMachinePersist read(Kryo kryo, Input input, Class> clazz) { E event = (E) kryo.readClassAndObject(input); S state = (S) kryo.readClassAndObject(input); - return new DefaultStateMachineContext(state, event, null, null); + Map eventHeaders = (Map) kryo.readClassAndObject(input); + Map variables = (Map) kryo.readClassAndObject(input); + List> childs = (List>) kryo.readClassAndObject(input); + return new DefaultStateMachineContext(childs, state, event, eventHeaders, new DefaultExtendedState(variables)); } } + private static class MessageHeadersSerializer extends Serializer { + + @Override + public void write(Kryo kryo, Output output, MessageHeaders object) { + HashMap map = new HashMap(); + for (Entry entry : object.entrySet()) { + map.put(entry.getKey(), entry.getValue()); + } + kryo.writeClassAndObject(output, map); + } + + @SuppressWarnings("unchecked") + @Override + public MessageHeaders read(Kryo kryo, Input input, Class type) { + Map eventHeaders = (Map) kryo.readClassAndObject(input); + return new MessageHeaders(eventHeaders); + } + + } + + private static class UUIDSerializer extends Serializer { + + public UUIDSerializer() { + setImmutable(true); + } + + @Override + public void write(final Kryo kryo, final Output output, final UUID uuid) { + output.writeLong(uuid.getMostSignificantBits()); + output.writeLong(uuid.getLeastSignificantBits()); + } + + @Override + public UUID read(final Kryo kryo, final Input input, final Class uuidClass) { + return new UUID(input.readLong(), input.readLong()); + } + } + } 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 f86e9a10..b4edfe64 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 @@ -20,6 +20,7 @@ import static org.hamcrest.Matchers.notNullValue; import static org.junit.Assert.assertThat; import java.util.Collection; +import java.util.HashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -34,6 +35,7 @@ import org.springframework.statemachine.access.StateMachineAccessor; import org.springframework.statemachine.ensemble.EnsembleListeger; import org.springframework.statemachine.listener.StateMachineListener; import org.springframework.statemachine.state.State; +import org.springframework.statemachine.support.DefaultExtendedState; import org.springframework.statemachine.support.DefaultStateMachineContext; import org.springframework.statemachine.transition.Transition; @@ -73,8 +75,8 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { assertThat(curatorClient.checkExists().forPath("/foo/data/current"), notNullValue()); - ensemble.setState(new DefaultStateMachineContext("S1","E1", null, null)); - ensemble.setState(new DefaultStateMachineContext("S2","E1", null, null)); + ensemble.setState(new DefaultStateMachineContext("S1","E1", new HashMap(), new DefaultExtendedState())); + ensemble.setState(new DefaultStateMachineContext("S2","E1", new HashMap(), new DefaultExtendedState())); } @@ -109,7 +111,7 @@ public class ZookeeperStateMachineEnsembleTests extends AbstractZookeeperTests { assertThat(listener1.joinedLatch.await(2, TimeUnit.SECONDS), is(true)); assertThat(listener2.joinedLatch.await(2, TimeUnit.SECONDS), is(true)); - ensemble1.setState(new DefaultStateMachineContext("S1", "E1", null, null)); + ensemble1.setState(new DefaultStateMachineContext("S1", "E1", new HashMap(), new DefaultExtendedState())); assertThat(listener2.eventLatch.await(2, TimeUnit.SECONDS), is(true)); } diff --git a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java index 7cfbd8d7..011c3e02 100644 --- a/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java +++ b/spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachinePersistTests.java @@ -18,12 +18,18 @@ package org.springframework.statemachine.zookeeper; import static org.hamcrest.Matchers.is; import static org.junit.Assert.assertThat; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; + 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.messaging.MessageHeaders; import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.ensemble.StateMachinePersist; +import org.springframework.statemachine.support.DefaultExtendedState; import org.springframework.statemachine.support.DefaultStateMachineContext; public class ZookeeperStateMachinePersistTests extends AbstractZookeeperTests { @@ -46,7 +52,7 @@ public class ZookeeperStateMachinePersistTests extends AbstractZookeeperTests { curatorClient, "/KryoStateMachinePersistTests"); StateMachineContext contextOut = - new DefaultStateMachineContext("S1", "E1", null, null); + new DefaultStateMachineContext("S1", "E1", new HashMap(), new DefaultExtendedState()); persist.write(contextOut, new Stat()); StateMachineContext contextIn = persist.read(new Stat()); @@ -54,4 +60,109 @@ public class ZookeeperStateMachinePersistTests extends AbstractZookeeperTests { assertThat(contextOut.getEvent(), is(contextIn.getEvent())); } + @Test + public void testEventHeaders() 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"); + + HashMap eventHeaders = new HashMap(); + eventHeaders.put("foo", "jee"); + + StateMachineContext contextOut = + new DefaultStateMachineContext("S1", "E1", eventHeaders, new DefaultExtendedState()); + persist.write(contextOut, new Stat()); + StateMachineContext contextIn = persist.read(new Stat()); + + assertThat(contextOut.getState(), is(contextIn.getState())); + assertThat(contextOut.getEvent(), is(contextIn.getEvent())); + assertThat(contextOut.getEventHeaders().get("foo"), is(contextIn.getEventHeaders().get("foo"))); + } + + @Test + public void testEventHeadersAsMessageHeaders() 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"); + + HashMap eventHeaders = new HashMap(); + eventHeaders.put("foo", "jee"); + MessageHeaders messageHeaders = new MessageHeaders(eventHeaders); + + StateMachineContext contextOut = + new DefaultStateMachineContext("S1", "E1", messageHeaders, new DefaultExtendedState()); + persist.write(contextOut, new Stat()); + StateMachineContext contextIn = persist.read(new Stat()); + + assertThat(contextOut.getState(), is(contextIn.getState())); + assertThat(contextOut.getEvent(), is(contextIn.getEvent())); + assertThat(contextOut.getEventHeaders().get("foo"), is(contextIn.getEventHeaders().get("foo"))); + } + + @Test + public void testExtendedState() 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"); + + HashMap eventHeaders = new HashMap(); + HashMap variables = new HashMap(); + variables.put("foo", "jee"); + + StateMachineContext contextOut = + new DefaultStateMachineContext("S1", "E1", eventHeaders, new DefaultExtendedState(variables)); + persist.write(contextOut, new Stat()); + StateMachineContext contextIn = persist.read(new Stat()); + + assertThat(contextOut.getState(), is(contextIn.getState())); + assertThat(contextOut.getEvent(), is(contextIn.getEvent())); + assertThat(contextOut.getExtendedState().getVariables().get("foo"), is(contextIn.getExtendedState().getVariables().get("foo"))); + } + + @Test + public void testChilds() 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 child = + new DefaultStateMachineContext("S2", "E2", new HashMap(), new DefaultExtendedState()); + List> childs = new ArrayList>(); + childs.add(child); + StateMachineContext contextOut = + new DefaultStateMachineContext(childs, "S1", "E1", new HashMap(), new DefaultExtendedState()); + persist.write(contextOut, new Stat()); + StateMachineContext contextIn = persist.read(new Stat()); + + assertThat(contextOut.getState(), is(contextIn.getState())); + assertThat(contextOut.getEvent(), is(contextIn.getEvent())); + + assertThat(contextIn.getChilds().size(), is(1)); + assertThat(contextIn.getChilds().get(0).getEvent(), is("E2")); + } + }