diff --git a/spring-statemachine-core/src/main/java/org/springframework/statemachine/service/StateMachineSerialisationService.java b/spring-statemachine-core/src/main/java/org/springframework/statemachine/service/StateMachineSerialisationService.java new file mode 100644 index 00000000..dd0bb8bd --- /dev/null +++ b/spring-statemachine-core/src/main/java/org/springframework/statemachine/service/StateMachineSerialisationService.java @@ -0,0 +1,47 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.statemachine.service; + +import org.springframework.statemachine.StateMachineContext; + +/** + * Generic interface to handle serialisation in a state machine. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public interface StateMachineSerialisationService { + + /** + * Serialise state machine context into byte array. + * + * @param context the context + * @return the data as byte[] + * @throws Exception the exception when serialisation fails + */ + byte[] serialiseStateMachineContext(StateMachineContext context) throws Exception; + + /** + * Deserialise state machine context from byte array. + * + * @param data the data + * @return the state machine context + * @throws Exception the exception when deserialisation fails + */ + StateMachineContext deserialiseStateMachineContext(byte[] data) throws Exception; +} diff --git a/spring-statemachine-data/jpa/src/main/java/org/springframework/statemachine/data/jpa/JpaRepositoryStateMachinePersist.java b/spring-statemachine-data/jpa/src/main/java/org/springframework/statemachine/data/jpa/JpaRepositoryStateMachinePersist.java index a4e7271d..30cb6cc5 100644 --- a/spring-statemachine-data/jpa/src/main/java/org/springframework/statemachine/data/jpa/JpaRepositoryStateMachinePersist.java +++ b/spring-statemachine-data/jpa/src/main/java/org/springframework/statemachine/data/jpa/JpaRepositoryStateMachinePersist.java @@ -18,6 +18,7 @@ package org.springframework.statemachine.data.jpa; import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.data.RepositoryStateMachinePersist; import org.springframework.statemachine.data.StateMachineRepository; +import org.springframework.statemachine.service.StateMachineSerialisationService; /** * {@code JPA} based implementation of a {@link RepositoryStateMachinePersist}. @@ -41,6 +42,18 @@ public class JpaRepositoryStateMachinePersist extends RepositoryStateMachi this.jpaStateMachineRepository = jpaStateMachineRepository; } + /** + * Instantiates a new jpa repository state machine persist. + * + * @param jpaStateMachineRepository the jpa state machine repository + * @param serialisationService the serialisation service + */ + public JpaRepositoryStateMachinePersist(JpaStateMachineRepository jpaStateMachineRepository, + StateMachineSerialisationService serialisationService) { + super(serialisationService); + this.jpaStateMachineRepository = jpaStateMachineRepository; + } + @Override protected StateMachineRepository getRepository() { return jpaStateMachineRepository; diff --git a/spring-statemachine-data/src/main/java/org/springframework/statemachine/data/RepositoryStateMachinePersist.java b/spring-statemachine-data/src/main/java/org/springframework/statemachine/data/RepositoryStateMachinePersist.java index fe1005a4..f8abeb49 100644 --- a/spring-statemachine-data/src/main/java/org/springframework/statemachine/data/RepositoryStateMachinePersist.java +++ b/spring-statemachine-data/src/main/java/org/springframework/statemachine/data/RepositoryStateMachinePersist.java @@ -19,7 +19,9 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.statemachine.StateMachineContext; import org.springframework.statemachine.StateMachinePersist; -import org.springframework.statemachine.kryo.KryoSerialisationService; +import org.springframework.statemachine.kryo.KryoStateMachineSerialisationService; +import org.springframework.statemachine.service.StateMachineSerialisationService; +import org.springframework.util.Assert; /** * Base implementation of a {@link StateMachinePersist} using Spring Data Repositories. @@ -33,14 +35,31 @@ import org.springframework.statemachine.kryo.KryoSerialisationService; public abstract class RepositoryStateMachinePersist implements StateMachinePersist { private final Log log = LogFactory.getLog(RepositoryStateMachinePersist.class); - private final KryoSerialisationService serialisationService = new KryoSerialisationService(); + private final StateMachineSerialisationService serialisationService; + + /** + * Instantiates a new repository state machine persist. + */ + protected RepositoryStateMachinePersist() { + this.serialisationService = new KryoStateMachineSerialisationService(); + } + + /** + * Instantiates a new repository state machine persist. + * + * @param serialisationService the serialisation service + */ + protected RepositoryStateMachinePersist(StateMachineSerialisationService serialisationService) { + Assert.notNull(serialisationService, "'serialisationService' must be set"); + this.serialisationService = serialisationService; + } @Override public void write(StateMachineContext context, Object contextObj) throws Exception { if (log.isDebugEnabled()) { log.debug("Persisting context " + context + " using contextObj " + contextObj); } - M build = build(context, serialisationService.serializeStateMachineContext(context)); + M build = build(context, serialisationService.serialiseStateMachineContext(context)); getRepository().save(build); } @@ -48,7 +67,7 @@ public abstract class RepositoryStateMachinePersist read(Object contextObj) throws Exception { M repositoryStateMachine = getRepository().findById(contextObj.toString()).orElse(null); if (repositoryStateMachine != null) { - return serialisationService.deserializeStateMachineContext(repositoryStateMachine.getStateMachineContext()); + return serialisationService.deserialiseStateMachineContext(repositoryStateMachine.getStateMachineContext()); } return null; } diff --git a/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/AbstractKryoStateMachineSerialisationService.java b/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/AbstractKryoStateMachineSerialisationService.java new file mode 100644 index 00000000..c3a8e669 --- /dev/null +++ b/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/AbstractKryoStateMachineSerialisationService.java @@ -0,0 +1,151 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.statemachine.kryo; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.service.StateMachineSerialisationService; +import org.springframework.util.Assert; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; +import com.esotericsoftware.kryo.pool.KryoCallback; +import com.esotericsoftware.kryo.pool.KryoFactory; +import com.esotericsoftware.kryo.pool.KryoPool; + +/** + * Abstract base implementation for {@link StateMachineSerialisationService} using kryo. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public abstract class AbstractKryoStateMachineSerialisationService implements StateMachineSerialisationService { + + protected final KryoPool pool; + + protected AbstractKryoStateMachineSerialisationService() { + KryoFactory factory = new KryoFactory() { + + @Override + public Kryo create() { + Kryo kryo = new Kryo(); + configureKryoInstance(kryo); + return kryo; + } + }; + this.pool = new KryoPool.Builder(factory).softReferences().build(); + } + + @Override + public byte[] serialiseStateMachineContext(StateMachineContext context) throws Exception { + return encode(context); + } + + @SuppressWarnings("unchecked") + @Override + public StateMachineContext deserialiseStateMachineContext(byte[] data) throws Exception { + return decode(data, StateMachineContext.class); + } + + /** + * Subclasses implement this method to encode with Kryo. + * + * @param kryo the Kryo instance + * @param object the object to encode + * @param output the Kryo Output instance + */ + protected abstract void doEncode(Kryo kryo, Object object, Output output); + + /** + * Subclasses implement this method to decode with Kryo. + * + * @param kryo the Kryo instance + * @param input the Kryo Input instance + * @param type the class of the decoded object + * @param the type for decoded object + * @return the decoded object + */ + protected abstract T doDecode(Kryo kryo, Input input, Class type); + + /** + * Subclasses implement this to configure the kryo instance. + * This is invoked on each new Kryo instance when it is created. + * + * @param kryo the kryo instance + */ + protected abstract void configureKryoInstance(Kryo kryo); + + private byte[] encode(Object object) throws IOException { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + encode(object, bos); + byte[] bytes = bos.toByteArray(); + bos.close(); + return bytes; + } + + private void encode(final Object object, OutputStream outputStream) throws IOException { + Assert.notNull(object, "cannot encode a null object"); + Assert.notNull(outputStream, "'outputSteam' cannot be null"); + final Output output = (outputStream instanceof Output ? (Output) outputStream : new Output(outputStream)); + this.pool.run(new KryoCallback() { + + @Override + public Void execute(Kryo kryo) { + doEncode(kryo, object, output); + return null; + } + }); + output.close(); + } + + private T decode(byte[] bytes, Class type) throws IOException { + Assert.notNull(bytes, "'bytes' cannot be null"); + final Input input = new Input(bytes); + try { + return decode(input, type); + } + finally { + input.close(); + } + } + + private T decode(InputStream inputStream, final Class type) throws IOException { + Assert.notNull(inputStream, "'inputStream' cannot be null"); + Assert.notNull(type, "'type' cannot be null"); + final Input input = (inputStream instanceof Input ? (Input) inputStream : new Input(inputStream)); + T result = null; + try { + result = this.pool.run(new KryoCallback(){ + + @Override + public T execute(Kryo kryo) { + return doDecode(kryo, input, type); + } + }); + } + finally { + input.close(); + } + return result; + } +} diff --git a/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoSerialisationService.java b/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoSerialisationService.java deleted file mode 100644 index 80cc29ab..00000000 --- a/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoSerialisationService.java +++ /dev/null @@ -1,87 +0,0 @@ -/* - * Copyright 2017 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.statemachine.kryo; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.util.UUID; - -import org.springframework.messaging.MessageHeaders; -import org.springframework.statemachine.StateMachineContext; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Input; -import com.esotericsoftware.kryo.io.Output; - -/** - * Simple service and utility class helping with kryo serialisation. - * - * @author Janne Valkealahti - * - */ -public class KryoSerialisationService { - - // 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") - @Override - 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; - } - }; - - /** - * Serialize state machine context into byte array. - * - * @param the generic type - * @param the element type - * @param context the context - * @return the byte[] - */ - public byte[] serializeStateMachineContext(StateMachineContext context) { - Kryo kryo = kryoThreadLocal.get(); - ByteArrayOutputStream out = new ByteArrayOutputStream(); - Output output = new Output(out); - kryo.writeObject(output, context); - output.close(); - return out.toByteArray(); - } - - /** - * Deserialize state machine context from byte array. - * - * @param the generic type - * @param the element type - * @param data the data - * @return the state machine context - */ - @SuppressWarnings("unchecked") - public StateMachineContext deserializeStateMachineContext(byte[] data) { - if (data == null || data.length == 0) { - return null; - } - Kryo kryo = kryoThreadLocal.get(); - ByteArrayInputStream in = new ByteArrayInputStream(data); - Input input = new Input(in); - return kryo.readObject(input, StateMachineContext.class); - } -} diff --git a/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoStateMachineSerialisationService.java b/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoStateMachineSerialisationService.java new file mode 100644 index 00000000..7c4b221c --- /dev/null +++ b/spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoStateMachineSerialisationService.java @@ -0,0 +1,54 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.statemachine.kryo; + +import java.util.UUID; + +import org.springframework.messaging.MessageHeaders; +import org.springframework.statemachine.StateMachineContext; +import org.springframework.statemachine.service.StateMachineSerialisationService; + +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.io.Input; +import com.esotericsoftware.kryo.io.Output; + +/** + * Implementation for {@link StateMachineSerialisationService} using kryo. + * + * @author Janne Valkealahti + * + * @param the type of state + * @param the type of event + */ +public class KryoStateMachineSerialisationService extends AbstractKryoStateMachineSerialisationService { + + @Override + protected void doEncode(Kryo kryo, Object object, Output output) { + kryo.writeObject(output, object); + } + + @Override + protected T doDecode(Kryo kryo, Input input, Class type) { + return kryo.readObject(input, type); + } + + @Override + protected void configureKryoInstance(Kryo kryo) { + kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer()); + kryo.addDefaultSerializer(MessageHeaders.class, new MessageHeadersSerializer()); + kryo.addDefaultSerializer(UUID.class, new UUIDSerializer()); + } +}