Make use of kryo more configurable

- Move away from thread local in favour of using
  kryo pooling.
- New interface StateMachineSerialisationService
  with KryoStateMachineSerialisationService.
- Try to work via constructors for instead of full blown
  configuration as it looks like this may give enough
  for users to customise.
- Relates to #437

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 <S> the type of state
+ * @param <E> the type of event
+ */
+public interface StateMachineSerialisationService<S, E> {
+
+	/**
+	 * 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<S, E> 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<S, E> deserialiseStateMachineContext(byte[] data) throws Exception;
+}

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<S, E> 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<S, E> serialisationService) {
+		super(serialisationService);
+		this.jpaStateMachineRepository = jpaStateMachineRepository;
+	}
+
 	@Override
 	protected StateMachineRepository<JpaRepositoryStateMachine> getRepository() {
 		return jpaStateMachineRepository;

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<M extends RepositoryStateMachine, S, E> implements StateMachinePersist<S, E, Object> {
 
 	private final Log log = LogFactory.getLog(RepositoryStateMachinePersist.class);
-	private final KryoSerialisationService serialisationService = new KryoSerialisationService();
+	private final StateMachineSerialisationService<S, E> serialisationService;
+
+	/**
+	 * Instantiates a new repository state machine persist.
+	 */
+	protected RepositoryStateMachinePersist() {
+		this.serialisationService = new KryoStateMachineSerialisationService<S, E>();
+	}
+
+	/**
+	 * Instantiates a new repository state machine persist.
+	 *
+	 * @param serialisationService the serialisation service
+	 */
+	protected RepositoryStateMachinePersist(StateMachineSerialisationService<S, E> serialisationService) {
+		Assert.notNull(serialisationService, "'serialisationService' must be set");
+		this.serialisationService = serialisationService;
+	}
 
 	@Override
 	public void write(StateMachineContext<S, E> 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<M extends RepositoryStateMac
 	public StateMachineContext<S, E> 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;
 	}

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 <S> the type of state
+ * @param <E> the type of event
+ */
+public abstract class AbstractKryoStateMachineSerialisationService<S, E> implements StateMachineSerialisationService<S, E> {
+
+	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<S, E> context) throws Exception {
+		return encode(context);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public StateMachineContext<S, E> 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 <T> the type for decoded object
+	 * @return the decoded object
+	 */
+	protected abstract <T> T doDecode(Kryo kryo, Input input, Class<T> 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<Void>() {
+
+			@Override
+			public Void execute(Kryo kryo) {
+				doEncode(kryo, object, output);
+				return null;
+			}
+		});
+		output.close();
+	}
+
+	private <T> T decode(byte[] bytes, Class<T> 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> T decode(InputStream inputStream, final Class<T> 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<T>(){
+
+				@Override
+				public T execute(Kryo kryo) {
+					return doDecode(kryo, input, type);
+				}
+			});
+		}
+		finally {
+			input.close();
+		}
+		return result;
+	}
+}

spring-statemachine-kryo/src/main/java/org/springframework/statemachine/kryo/KryoSerialisationService.java

@@ -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<Kryo> kryoThreadLocal = new ThreadLocal<Kryo>() {
-
-		@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 <S> the generic type
-	 * @param <E> the element type
-	 * @param context the context
-	 * @return the byte[]
-	 */
-	public <S, E> byte[] serializeStateMachineContext(StateMachineContext<S, E> 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 <S> the generic type
-	 * @param <E> the element type
-	 * @param data the data
-	 * @return the state machine context
-	 */
-	@SuppressWarnings("unchecked")
-	public <S, E> StateMachineContext<S, E> 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);
-	}
-}

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 <S> the type of state
+ * @param <E> the type of event
+ */
+public class KryoStateMachineSerialisationService<S, E> extends AbstractKryoStateMachineSerialisationService<S, E> {
+
+	@Override
+	protected void doEncode(Kryo kryo, Object object, Output output) {
+		kryo.writeObject(output, object);
+	}
+
+	@Override
+	protected <T> T doDecode(Kryo kryo, Input input, Class<T> type) {
+		return kryo.readObject(input, type);
+	}
+
+	@Override
+	protected void configureKryoInstance(Kryo kryo) {
+		kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer<S, E>());
+		kryo.addDefaultSerializer(MessageHeaders.class, new MessageHeadersSerializer());
+		kryo.addDefaultSerializer(UUID.class, new UUIDSerializer());
+	}
+}