INT-3776: Add Codec support

JIRA: https://jira.spring.io/browse/INT-3776

revert LoggingHandlerTests

Added byte[] serialize(Object obj)

Clean up java doc and some minor name changes

added null checks

Cleanup and rename methods to encode/decode

remove SerializationFaileException

clean up

Polishing

build.gradle

@@ -113,6 +113,7 @@ subprojects { subproject ->
 		jsonpathVersion = '2.0.0'
 		junitVersion = '4.11'
 		jythonVersion = '2.5.3'
+		kryoShadedVersion = '3.0.0'
 		log4jVersion = '1.2.17'
 		mockitoVersion = '1.9.5'
 		mysqlVersion = '5.1.34'
@@ -304,6 +305,7 @@ project('spring-integration-core') {
 		compile("com.fasterxml.jackson.core:jackson-databind:$jackson2Version", optional)
 		compile("com.jayway.jsonpath:json-path:$jsonpathVersion", optional)
 		compile("io.fastjson:boon:$boonVersion", optional)
+		compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional)
 
 		testCompile ("org.aspectj:aspectjweaver:$aspectjVersion")
 		testCompile ("net.openhft:chronicle:$chronicleVersion")

spring-integration-core/src/main/java/org/springframework/integration/codec/Codec.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2015 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.integration.codec;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * Interface for classes that perform both encode (serialize) and decode (deserialize) on multiple classes.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public interface Codec {
+	/**
+	 * Encode (encode) an object to an OutputStream
+	 * @param object the object to encode
+	 * @param outputStream the OutputStream
+	 * @throws IOException if the operation fails
+	 */
+	void encode(Object object, OutputStream outputStream) throws IOException;
+
+	/**
+	 * Encode an object to a byte array
+	 * @param object the object to encode
+	 * @return the bytes
+	 * @throws IOException if the operation fails
+	 */
+	byte[] encode(Object object) throws IOException;
+
+	/**
+	 * Decode an object of a given type
+	 * @param inputStream the input stream containing the encoded object
+	 * @param type the object's class
+	 * @param <T> the object's type
+	 * @return the object
+	 * @throws IOException if the operation fails
+	 */
+	<T> T decode(InputStream inputStream, Class<T> type) throws IOException;
+
+	/**
+	 * Decode an object of a given type
+	 * @param bytes the byte array containing the encoded object
+	 * @param type the object's class
+	 * @param <T> the object's type
+	 * @return the object
+	 * @throws IOException if the operation fails
+	 */
+	<T> T decode(byte[] bytes, Class<T> type) throws IOException;
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/CompositeCodec.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2015 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.integration.codec;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.integration.util.ClassUtils;
+import org.springframework.util.Assert;
+
+/**
+ * A Codec that can delegate to one out of many Codecs, each mapped to a class.
+ * @author David Turanski
+ * @since 4.2
+ */
+public class CompositeCodec implements Codec {
+
+	private final Codec defaultCodec;
+
+	private final Map<Class<?>, Codec> delegates;
+
+	public CompositeCodec(Map<Class<?>, Codec> delegates, Codec defaultCodec) {
+		Assert.notNull(defaultCodec, "'defaultCodec' cannot be null");
+		this.defaultCodec = defaultCodec;
+		this.delegates = new HashMap<Class<?>, Codec>(delegates);
+	}
+
+	public CompositeCodec(Codec defaultCodec) {
+		this(null, defaultCodec);
+	}
+
+	@Override
+	public void encode(Object object, OutputStream outputStream) throws IOException {
+		Assert.notNull(object, "cannot encode a null object");
+		Assert.notNull(outputStream, "'outputStream' cannot be null");
+		Codec codec = findDelegate(object.getClass());
+		if (codec != null) {
+			codec.encode(object, outputStream);
+		}
+		else {
+			this.defaultCodec.encode(object, outputStream);
+		}
+	}
+
+	@Override
+	public byte[] encode(Object object) throws IOException {
+		Assert.notNull(object, "cannot encode a null object");
+		Codec codec = findDelegate(object.getClass());
+		if (codec != null) {
+			return codec.encode(object);
+		}
+		else {
+			return this.defaultCodec.encode(object);
+		}
+	}
+
+	@Override
+	public <T> T decode(InputStream inputStream, Class<T> type) throws IOException {
+		Assert.notNull(inputStream, "'inputStream' cannot be null");
+		Assert.notNull(type, "'type' cannot be null");
+		Codec codec = findDelegate(type);
+		if (codec != null) {
+			return codec.decode(inputStream, type);
+		}
+		else {
+			return this.defaultCodec.decode(inputStream, type);
+		}
+	}
+
+	@Override
+	public <T> T decode(byte[] bytes, Class<T> type) throws IOException {
+		return decode(new ByteArrayInputStream(bytes), type);
+	}
+
+	private Codec findDelegate(Class<?> type) {
+		if (this.delegates == null) {
+			return null;
+		}
+
+		Class<?> clazz = ClassUtils.findClosestMatch(type, this.delegates.keySet(), false);
+		return this.delegates.get(clazz);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.springframework.integration.codec.Codec;
+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;
+
+/**
+ * Base class for {@link Codec}s using {@link Kryo}.
+ * Manages pooled {@link Kryo} instances.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public abstract class AbstractKryoCodec implements Codec {
+
+	protected final KryoPool pool;
+
+	protected AbstractKryoCodec() {
+		KryoFactory factory = new KryoFactory() {
+			public Kryo create() {
+				Kryo kryo = new Kryo();
+				// configure Kryo instance, customize settings
+				configureKryoInstance(kryo);
+				return kryo;
+			}
+		};
+		// Build pool with SoftReferences enabled (optional)
+		pool = new KryoPool.Builder(factory).softReferences().build();
+	}
+
+	@Override
+	public 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<Object>() {
+
+			public Object execute(Kryo kryo) {
+				doEncode(kryo, object, output);
+				return Void.class;
+			}
+
+		});
+		output.close();
+	}
+
+	@Override
+	public <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();
+		}
+	}
+
+	@Override
+	public <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>() {
+
+				public T execute(Kryo kryo) {
+					return doDecode(kryo, input, type);
+				}
+
+			});
+		}
+		finally {
+			input.close();
+		}
+		return result;
+	}
+
+	@Override
+	public byte[] encode(Object object) throws IOException {
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		encode(object, bos);
+		byte[] bytes = bos.toByteArray();
+		bos.close();
+		return bytes;
+	}
+
+	/**
+	 * 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);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoRegistrar.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * Base class for {@link KryoRegistrar} implementations.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public abstract class AbstractKryoRegistrar implements KryoRegistrar {
+
+	protected final static Kryo kryo = new Kryo();
+
+	protected final Log log = LogFactory.getLog(this.getClass());
+
+	@Override
+	public void registerTypes(Kryo kryo) {
+		for (Registration registration : getRegistrations()) {
+			register(kryo, registration);
+		}
+	}
+
+	/**
+	 * Subclasses implement this to get provided registrations.
+	 * @return a list of {@link Registration}
+	 */
+	public abstract List<Registration> getRegistrations();
+
+	private void register(Kryo kryo, Registration registration) {
+		int id = registration.getId();
+
+		Registration existing = kryo.getRegistration(id);
+
+		if (existing != null) {
+			throw new RuntimeException((String.format("registration already exists %s", existing)));
+		}
+
+		if (log.isInfoEnabled()) {
+			log.info(String.format("registering %s with serializer %s", registration,
+					registration.getSerializer().getClass().getName()));
+		}
+
+		kryo.register(registration);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/CompositeKryoRegistrar.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar} that delegates and validates registrations across all components.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class CompositeKryoRegistrar extends AbstractKryoRegistrar {
+
+	private final List<KryoRegistrar> delegates;
+
+	public CompositeKryoRegistrar(List<KryoRegistrar> delegates) {
+		this.delegates = new ArrayList<KryoRegistrar>(delegates);
+
+		if (!CollectionUtils.isEmpty(this.delegates)) {
+			validateRegistrations();
+		}
+	}
+
+	@Override
+	public List<Registration> getRegistrations() {
+		List<Registration> registrations = new ArrayList<Registration>();
+		for (KryoRegistrar registrar : delegates) {
+			registrations.addAll(registrar.getRegistrations());
+		}
+		return registrations;
+	}
+
+	private void validateRegistrations() {
+		List<Integer> ids = new ArrayList<Integer>();
+		List<Class<?>> types = new ArrayList<Class<?>>();
+
+		for (Registration registration : getRegistrations()) {
+			Assert.isTrue(registration.getId() >= MIN_REGISTRATION_VALUE,
+					"registration ID must be >= " + MIN_REGISTRATION_VALUE);
+			if (ids.contains(registration.getId())) {
+				throw new RuntimeException(String.format("Duplicate registration ID found: %d",
+						registration.getId()));
+			}
+			ids.add(registration.getId());
+
+			if (types.contains(registration.getType())) {
+				throw new RuntimeException(String.format("Duplicate registration found for type: %s",
+						registration.getType()));
+			}
+			types.add(registration.getType());
+
+			if (log.isInfoEnabled()) {
+				log.info(String.format("configured Kryo registration %s with serializer %s", registration,
+						registration.getSerializer().getClass().getName()));
+			}
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileKryoRegistrar.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar} used to validateRegistration a File serializer.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class FileKryoRegistrar extends AbstractKryoRegistrar {
+
+	private final static int DEFAULT_REGISTRATION_ID = 40;
+
+	private final int registrationId;
+
+	private final FileSerializer fileSerializer = new FileSerializer();
+
+	public FileKryoRegistrar() {
+		this.registrationId = DEFAULT_REGISTRATION_ID;
+	}
+
+	/**
+	 *
+	 * @param registrationId overrides the default registration ID.
+	 */
+	public FileKryoRegistrar(int registrationId) {
+		this.registrationId = registrationId;
+	}
+
+	@Override
+	public List<Registration> getRegistrations() {
+		return Collections.singletonList(new Registration(File.class, this.fileSerializer, this.registrationId));
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileSerializer.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.io.File;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ * A custom Kryo {@link Serializer} for serializing File payloads.
+ * It serializes the file path and creates a new File instance to preserve the original path.
+ * File does not preserve the absolute otherwise as <em>prefixLength</em>
+ * is declared transient.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class FileSerializer extends Serializer<File> {
+
+	@Override
+	public void write(Kryo kryo, Output output, File file) {
+		output.writeString(file.getPath());
+	}
+
+	@Override
+	public File read(Kryo kryo, Input input, Class<File> type) {
+		String path  = input.readString();
+		return new File(path);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassListRegistrar.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar} used to validateRegistration a
+ * list of Java classes. This assigns a sequential registration ID starting with an initial value (50 by default), but
+ * may be configured. This is easiest to set up but requires that every server node be configured with the identical
+ * list in the same order.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class KryoClassListRegistrar extends AbstractKryoRegistrar {
+
+	private final List<Class<?>> registeredClasses;
+
+	private int initialValue = 50;
+
+	/**
+	 * @param classes the list of classes to validateRegistration
+	 */
+	public KryoClassListRegistrar(List<Class<?>> classes) {
+		this.registeredClasses = new ArrayList<Class<?>>(classes);
+	}
+
+	/**
+	 * Set the inital ID value. Classes in the list will be sequentially assigned an ID starting with this value
+	 * (default is 50).
+	 * @param initialValue the initial value
+	 */
+	public void setInitialValue(int initialValue) {
+		Assert.isTrue(initialValue >= MIN_REGISTRATION_VALUE,
+				"'initialValue' must be >= " + MIN_REGISTRATION_VALUE);
+		this.initialValue = initialValue;
+	}
+
+
+	@Override
+	public List<Registration> getRegistrations() {
+		List<Registration> registrations = new ArrayList<Registration>();
+		if (!CollectionUtils.isEmpty(this.registeredClasses)) {
+			for (int i = 0; i < this.registeredClasses.size(); i++) {
+				registrations.add(new Registration(this.registeredClasses.get(i),
+						kryo.getSerializer(this.registeredClasses.get(i)), i + this.initialValue));
+			}
+		}
+		return registrations;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoClassMapRegistrar.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.util.CollectionUtils;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar} implementation backed by a Map
+ * used to explicitly set the registration ID for each class.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class KryoClassMapRegistrar extends AbstractKryoRegistrar {
+
+	private final Map<Integer, Class<?>> registeredClasses;
+
+	public KryoClassMapRegistrar(Map<Integer, Class<?>> kryoRegisteredClasses) {
+		this.registeredClasses = new HashMap<Integer, Class<?>>(kryoRegisteredClasses);
+	}
+
+	@Override
+	public List<Registration> getRegistrations() {
+		List<Registration> registrations = new ArrayList<Registration>();
+		if (!CollectionUtils.isEmpty(this.registeredClasses)) {
+			for (Map.Entry<Integer, Class<?>> entry : this.registeredClasses.entrySet()) {
+				registrations.add(
+						new Registration(entry.getValue(), kryo.getSerializer(entry.getValue()), entry.getKey()));
+			}
+		}
+		return registrations;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrar.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.List;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * Strategy interface used by {@link PojoCodec} to configure registrations
+ * classes consistently across {@link Kryo} instances.
+ * By default, user defined types are not registered to Kryo.
+ * Registration allows a unique ID (small positive integer is ideal) to represent the type
+ * in the byte stream. In a distributed environment, all Kryo instances must maintain a
+ * consistent registration configuration in order for serialization to function properly.
+ * Registrations can result in better performance in demanding situations,
+ * but requires some care to maintain. Use this feature only if you really need it.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public interface KryoRegistrar {
+
+	int MIN_REGISTRATION_VALUE = 10;
+
+	/**
+	 * This method is invoked by the {@link PojoCodec} and
+	 * applied to the {@link Kryo} instance whenever a new instance is created.
+	 * @param kryo the Kryo instance
+	 */
+	void registerTypes(Kryo kryo);
+
+	/**
+	 *
+	 * @return the list of {@link Registration} provided
+	 */
+	List<Registration> getRegistrations();
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/KryoRegistrationRegistrar.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * A {@link KryoRegistrar} implementation backed by a List of {@link Registration}.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class KryoRegistrationRegistrar extends AbstractKryoRegistrar {
+
+	private final List<Registration> registrations;
+
+	public KryoRegistrationRegistrar(List<Registration> registrations) {
+		this.registrations = registrations != null
+				? new ArrayList<Registration>(registrations)
+				: new ArrayList<Registration>();
+	}
+
+	@Override
+	public List<Registration> getRegistrations() {
+		return this.registrations;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.util.CollectionUtils;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ * Kryo Codec that can encode and decode arbitrary types. Classes and associated
+ * {@link com.esotericsoftware.kryo.Serializer}s may be registered via
+ * {@link KryoRegistrar}s.
+ *
+ * @author David Turanski
+ * @since 4.2
+ */
+public class PojoCodec extends AbstractKryoCodec {
+
+	private final CompositeKryoRegistrar kryoRegistrar;
+
+	private final boolean useReferences;
+
+	public PojoCodec() {
+		this.kryoRegistrar = null;
+		this.useReferences = true;
+	}
+
+	/**
+	 * Create an instance with a single KryoRegistrar.
+	 * @param kryoRegistrar the registrar.
+	 */
+	public PojoCodec(KryoRegistrar kryoRegistrar) {
+		this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, true);
+	}
+
+	/**
+	 * Create an instance with zero to many KryoRegistrars.
+	 * @param kryoRegistrars a list KryoRegistrars.
+	 */
+	public PojoCodec(List<KryoRegistrar> kryoRegistrars) {
+		this.kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null :
+				new CompositeKryoRegistrar(kryoRegistrars);
+		this.useReferences = true;
+	}
+
+	/**
+	 * Create an instance with a single KryoRegistrar.
+	 * @param kryoRegistrar the registrar.
+	 * @param useReferences set to false if references are not required (if the object graph is known to be acyclical).
+	 * The default is 'true' which is less performant but more flexible.
+	 */
+	public PojoCodec(KryoRegistrar kryoRegistrar, boolean useReferences) {
+		this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, useReferences);
+	}
+
+	/**
+	 * Create an instance with zero to many KryoRegistrars.
+	 * @param kryoRegistrars a list KryoRegistrars.
+	 * @param useReferences set to false if references are not required (if the object graph is known to be acyclical).
+	 * The default is 'true' which is less performant but more flexible.
+	 */
+	public PojoCodec(List<KryoRegistrar> kryoRegistrars, boolean useReferences) {
+		this.kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null :
+				new CompositeKryoRegistrar(kryoRegistrars);
+		this.useReferences = useReferences;
+	}
+
+	@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) {
+		if (this.kryoRegistrar != null) {
+			this.kryoRegistrar.registerTypes(kryo);
+		}
+		kryo.setReferences(this.useReferences);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * The <a href="https://code.google.com/p/kryo/">Kryo</a> specific {@code Codec} classes.
+ */
+package org.springframework.integration.codec.kryo;

spring-integration-core/src/main/java/org/springframework/integration/codec/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides base classes for the {@code Codec} abstraction.
+ */
+package org.springframework.integration.codec;

spring-integration-core/src/test/java/org/springframework/integration/codec/CompositeCodecTests.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2015 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.integration.codec;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.integration.codec.kryo.PojoCodec;
+
+/**
+ * @author David Turanski
+ * @since 4.2
+ */
+public class CompositeCodecTests {
+
+	private Codec codec;
+
+
+	@Before
+	public void setup() {
+		Map<Class<?>, Codec> codecs = new HashMap<Class<?>, Codec>();
+		this.codec = new CompositeCodec(codecs, new PojoCodec());
+	}
+
+	@Test
+	public void testPojoSerialization() throws IOException {
+		SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("hello", 123);
+		SomeClassWithNoDefaultConstructors foo2 = this.codec.decode(
+				this.codec.encode(foo),
+				SomeClassWithNoDefaultConstructors.class);
+		assertEquals(foo, foo2);
+	}
+
+	static class SomeClassWithNoDefaultConstructors {
+
+		private String val1;
+
+		private int val2;
+
+		public SomeClassWithNoDefaultConstructors(String val1, int val2) {
+			this.val1 = val1;
+			this.val2 = val2;
+		}
+
+		@Override
+		public boolean equals(Object other) {
+			if (!(other instanceof SomeClassWithNoDefaultConstructors)) {
+				return false;
+			}
+			SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other;
+			return (this.val1.equals(that.val1) && this.val2 == that.val2);
+		}
+
+		@Override
+		public int hashCode() {
+			int result = this.val1.hashCode();
+			result = 31 * result + this.val2;
+			return result;
+		}
+
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/FileKryoRegistrarTests.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.junit.Test;
+
+
+/**
+ * @author David Turanski
+ * @since 4.2
+ */
+public class FileKryoRegistrarTests {
+
+	@Test
+	public void test() throws IOException {
+		PojoCodec pc = new PojoCodec(new FileKryoRegistrar());
+		File file = new File("/foo/bar");
+		File file2 = pc.decode(pc.encode(file), File.class);
+		assertEquals(file, file2);
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/codec/kryo/KryoCodecTests.java

@@ -0,0 +1,174 @@
+/*
+ * Copyright 2015 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.integration.codec.kryo;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+/**
+ * @author David Turanski
+ * @since 4.2
+ */
+public class KryoCodecTests {
+
+	@Test
+	public void testStringSerialization() throws IOException {
+		String str = "hello";
+		PojoCodec codec = new PojoCodec();
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+
+		codec.encode(str, bos);
+
+		String s2 = codec.decode(bos.toByteArray(), String.class);
+		assertEquals(str, s2);
+	}
+
+	@Test
+	public void testSerializationWithStreams() throws IOException {
+		String str = "hello";
+		File file = new File("test.ser");
+		PojoCodec codec = new PojoCodec();
+		FileOutputStream fos = new FileOutputStream(file);
+		codec.encode(str, fos);
+		fos.close();
+
+		FileInputStream fis = new FileInputStream(file);
+		String s2 = codec.decode(fis, String.class);
+		file.delete();
+		assertEquals(str, s2);
+	}
+
+	@Test
+	public void testPojoSerialization() throws IOException {
+		PojoCodec codec = new PojoCodec();
+		SomeClassWithNoDefaultConstructors foo = new SomeClassWithNoDefaultConstructors("foo", 123);
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		codec.encode(foo, bos);
+		Object foo2 = codec.decode(bos.toByteArray(), SomeClassWithNoDefaultConstructors.class);
+		assertEquals(foo, foo2);
+	}
+
+	@Test
+	public void testPrimitiveSerialization() throws IOException {
+		PojoCodec codec = new PojoCodec();
+
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		codec.encode(true, bos);
+		boolean b = codec.decode(bos.toByteArray(), Boolean.class);
+		assertEquals(true, b);
+		b = codec.decode(bos.toByteArray(), boolean.class);
+		assertEquals(true, b);
+
+		bos = new ByteArrayOutputStream();
+		codec.encode(3.14159, bos);
+
+		double d = codec.decode(bos.toByteArray(), double.class);
+		assertEquals(3.14159, d, 0.00001);
+
+		bos = new ByteArrayOutputStream();
+		codec.encode(3.14159, bos);
+
+		d = codec.decode(bos.toByteArray(), Double.class);
+		assertEquals(3.14159, d, 0.00001);
+
+	}
+
+	@Test
+	public void testMapSerialization() throws IOException {
+		PojoCodec codec = new PojoCodec();
+		Map<String, Integer> map = new HashMap<String, Integer>();
+		map.put("one", 1);
+		map.put("two", 2);
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		codec.encode(map, bos);
+		Map<?, ?> m2 = (Map<?, ?>) codec.decode(bos.toByteArray(), HashMap.class);
+		assertEquals(2, m2.size());
+		assertEquals(1, m2.get("one"));
+		assertEquals(2, m2.get("two"));
+	}
+
+	@Test
+	public void testComplexObjectSerialization() throws IOException {
+		PojoCodec codec = new PojoCodec();
+		Foo foo = new Foo();
+		foo.put("one", 1);
+		foo.put("two", 2);
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		codec.encode(foo, bos);
+
+		Foo foo2 = codec.decode(bos.toByteArray(), Foo.class);
+		assertEquals(1, foo2.get("one"));
+		assertEquals(2, foo2.get("two"));
+	}
+
+	static class SomeClassWithNoDefaultConstructors {
+
+		private String val1;
+
+		private int val2;
+
+		public SomeClassWithNoDefaultConstructors(String val1, int val2) {
+			this.val1 = val1;
+			this.val2 = val2;
+		}
+
+		@Override
+		public boolean equals(Object other) {
+			if (!(other instanceof SomeClassWithNoDefaultConstructors)) {
+				return false;
+			}
+			SomeClassWithNoDefaultConstructors that = (SomeClassWithNoDefaultConstructors) other;
+			return (this.val1.equals(that.val1) && this.val2 == that.val2);
+		}
+
+		@Override
+		public int hashCode() {
+			int result = this.val1.hashCode();
+			result = 31 * result + this.val2;
+			return result;
+		}
+
+	}
+
+	static class Foo {
+
+		private Map<Object, Object> map;
+
+		public Foo() {
+			map = new HashMap<Object, Object>();
+		}
+
+		public void put(Object key, Object value) {
+			this.map.put(key, value);
+		}
+
+		public Object get(Object key) {
+			return this.map.get(key);
+		}
+
+	}
+
+}