Spring/spring-statemachine
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add headers and events to serialization

Browse Source 
- Relates to #35
- Now event headers and extended state variables
  are serialized with kryo and zookeeper.
This commit is contained in:
Janne Valkealahti
2015-07-05 09:52:36 +01:00
parent cb3b811707
commit bba4bc1d28
5 changed files with 184 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -33,11 +33,11 @@ public interface StateMachinePersist<S, E, T> {
* 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<S, E> contect, T contextOjb) throws Exception;
void write(StateMachineContext<S, E> context, T contextOjb) throws Exception;
/**
* Read a {@link StateMachineContext} from a persistent store

9
spring-statemachine-core/src/main/java/org/springframework/statemachine/support/DefaultExtendedState.java
View File

@@ -37,6 +37,15 @@ public class DefaultExtendedState implements ExtendedState {
this.variables = new HashMap<Object, Object>();
}
/**
* Instantiates a new default extended state.
*
* @param variables the variables
*/
public DefaultExtendedState(Map<Object, Object> variables) {
this.variables = variables;
}
@Override
public Map<Object, Object> getVariables() {
return variables;

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

@@ -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<S, E> implements StateMachinePersist<S
protected Kryo initialValue() {
Kryo kryo = new Kryo();
kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer());
kryo.addDefaultSerializer(MessageHeaders.class, new MessageHeadersSerializer());
kryo.addDefaultSerializer(UUID.class, new UUIDSerializer());
return kryo;
}
};
@@ -118,6 +127,9 @@ public class ZookeeperStateMachinePersist<S, E> implements StateMachinePersist<S
public void write(Kryo kryo, Output output, StateMachineContext<S, E> 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<S, E> implements StateMachinePersist<S
public StateMachineContext<S, E> read(Kryo kryo, Input input, Class<StateMachineContext<S, E>> clazz) {
E event = (E) kryo.readClassAndObject(input);
S state = (S) kryo.readClassAndObject(input);
return new DefaultStateMachineContext<S, E>(state, event, null, null);
Map<String, Object> eventHeaders = (Map<String, Object>) kryo.readClassAndObject(input);
Map<Object, Object> variables = (Map<Object, Object>) kryo.readClassAndObject(input);
List<StateMachineContext<S, E>> childs = (List<StateMachineContext<S, E>>) kryo.readClassAndObject(input);
return new DefaultStateMachineContext<S, E>(childs, state, event, eventHeaders, new DefaultExtendedState(variables));
}
}
private static class MessageHeadersSerializer extends Serializer<MessageHeaders> {
@Override
public void write(Kryo kryo, Output output, MessageHeaders object) {
HashMap<String, Object> map = new HashMap<String, Object>();
for (Entry<String, Object> entry : object.entrySet()) {
map.put(entry.getKey(), entry.getValue());
}
kryo.writeClassAndObject(output, map);
}
@SuppressWarnings("unchecked")
@Override
public MessageHeaders read(Kryo kryo, Input input, Class<MessageHeaders> type) {
Map<String, Object> eventHeaders = (Map<String, Object>) kryo.readClassAndObject(input);
return new MessageHeaders(eventHeaders);
}
}
private static class UUIDSerializer extends Serializer<UUID> {
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<UUID> uuidClass) {
return new UUID(input.readLong(), input.readLong());
}
}
}

8
spring-statemachine-zookeeper/src/test/java/org/springframework/statemachine/zookeeper/ZookeeperStateMachineEnsembleTests.java
View File

@@ -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<String, String>("S1","E1", null, null));
ensemble.setState(new DefaultStateMachineContext<String, String>("S2","E1", null, null));
ensemble.setState(new DefaultStateMachineContext<String, String>("S1","E1", new HashMap<String, Object>(), new DefaultExtendedState()));
ensemble.setState(new DefaultStateMachineContext<String, String>("S2","E1", new HashMap<String, Object>(), 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<String, String>("S1", "E1", null, null));
ensemble1.setState(new DefaultStateMachineContext<String, String>("S1", "E1", new HashMap<String, Object>(), new DefaultExtendedState()));
assertThat(listener2.eventLatch.await(2, TimeUnit.SECONDS), is(true));
}

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

@@ -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<String, String> contextOut =
new DefaultStateMachineContext<String, String>("S1", "E1", null, null);
new DefaultStateMachineContext<String, String>("S1", "E1", new HashMap<String, Object>(), new DefaultExtendedState());
persist.write(contextOut, new Stat());
StateMachineContext<String, String> 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<String, String, Stat> persist = new ZookeeperStateMachinePersist<String, String>(
curatorClient, "/KryoStateMachinePersistTests");
HashMap<String, Object> eventHeaders = new HashMap<String, Object>();
eventHeaders.put("foo", "jee");
StateMachineContext<String, String> contextOut =
new DefaultStateMachineContext<String, String>("S1", "E1", eventHeaders, new DefaultExtendedState());
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()));
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<String, String, Stat> persist = new ZookeeperStateMachinePersist<String, String>(
curatorClient, "/KryoStateMachinePersistTests");
HashMap<String, Object> eventHeaders = new HashMap<String, Object>();
eventHeaders.put("foo", "jee");
MessageHeaders messageHeaders = new MessageHeaders(eventHeaders);
StateMachineContext<String, String> contextOut =
new DefaultStateMachineContext<String, String>("S1", "E1", messageHeaders, new DefaultExtendedState());
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()));
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<String, String, Stat> persist = new ZookeeperStateMachinePersist<String, String>(
curatorClient, "/KryoStateMachinePersistTests");
HashMap<String, Object> eventHeaders = new HashMap<String, Object>();
HashMap<Object, Object> variables = new HashMap<Object, Object>();
variables.put("foo", "jee");
StateMachineContext<String, String> contextOut =
new DefaultStateMachineContext<String, String>("S1", "E1", eventHeaders, new DefaultExtendedState(variables));
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()));
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<String, String, Stat> persist = new ZookeeperStateMachinePersist<String, String>(
curatorClient, "/KryoStateMachinePersistTests");
StateMachineContext<String, String> child =
new DefaultStateMachineContext<String, String>("S2", "E2", new HashMap<String, Object>(), new DefaultExtendedState());
List<StateMachineContext<String, String>> childs = new ArrayList<StateMachineContext<String, String>>();
childs.add(child);
StateMachineContext<String, String> contextOut =
new DefaultStateMachineContext<String, String>(childs, "S1", "E1", new HashMap<String, Object>(), new DefaultExtendedState());
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()));
assertThat(contextIn.getChilds().size(), is(1));
assertThat(contextIn.getChilds().get(0).getEvent(), is("E2"));
}
}