Mirrors codec enhancements in XD 1.2.x

spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/CodecConfiguration.java

@@ -0,0 +1,59 @@
+/*
+ * 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.cloud.stream.config;
+
+import java.util.ArrayList;
+import java.util.Map;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
+import org.springframework.xd.dirt.integration.bus.serializer.kryo.FileKryoRegistrar;
+import org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoRegistrar;
+import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
+
+/**
+ * @author David Turanski
+ */
+@Configuration
+public class CodecConfiguration {
+
+
+	@Autowired
+	ApplicationContext applicationContext;
+
+	@ConditionalOnMissingBean(KryoCodecProperties.class)
+	@Bean(name = "spring.cloud.streams.codec.kryo.CONFIGURATION_PROPERTIES")
+	public KryoCodecProperties kryoCodecProperties() {
+		return new KryoCodecProperties();
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(name = "codec")
+	public MultiTypeCodec<?> codec() {
+		Map<String, KryoRegistrar> kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar
+				.class);
+		return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values()), kryoCodecProperties().isReferences());
+	}
+
+	@Bean
+	public KryoRegistrar fileRegistrar() {
+		return new FileKryoRegistrar();
+	}
+}

spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/KryoCodecProperties.java

@@ -0,0 +1,38 @@
+/*
+ * 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.cloud.stream.config;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author David Turanski
+ */
+@ConfigurationProperties("spring.cloud.codec.kryo")
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+public class KryoCodecProperties {
+	private boolean references = true;
+	
+	public boolean isReferences() {
+		return references;
+	}
+
+	public void setReferences(boolean references) {
+		this.references = references;
+	}
+	
+}

spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java

@@ -28,21 +28,18 @@ import com.esotericsoftware.kryo.pool.KryoFactory;
 import com.esotericsoftware.kryo.pool.KryoPool;
 
 import org.springframework.util.Assert;
-import org.springframework.xd.dirt.integration.bus.serializer.AbstractCodec;
+import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
 
 /**
  * Base class for Codecs using {@link com.esotericsoftware.kryo.Kryo}
- *
  * @author David Turanski
  */
-public abstract class AbstractKryoCodec<T> extends AbstractCodec<T> {
-
-	private final KryoFactory factory;
+public abstract class AbstractKryoCodec implements MultiTypeCodec<Object> {
 
 	protected final KryoPool pool;
 
 	protected AbstractKryoCodec() {
-		factory = new KryoFactory() {
+		KryoFactory factory = new KryoFactory() {
 			public Kryo create() {
 				Kryo kryo = new Kryo();
 				// configure kryo instance, customize settings
@@ -56,55 +53,65 @@ public abstract class AbstractKryoCodec<T> extends AbstractCodec<T> {
 
 	/**
 	 * Serialize an object using an existing output stream
-	 *
 	 * @param object the object to be serialized
 	 * @param outputStream the output stream, e.g. a FileOutputStream
 	 * @throws IOException
 	 */
-	@Override
-	public void serialize(final T object, OutputStream outputStream) throws IOException {
-		Assert.notNull(outputStream, "'outputSteam' cannot be null");
-		final Output output = new Output(outputStream);
-		try {
-			pool.run(new KryoCallback<Object>() {
-				@Override
-				public Object execute(Kryo kryo) {
-					doSerialize(kryo, object, output);
-					return Void.class;
-				}
-			});
-		} finally {
-			output.close();
-		}
+
+	public void serialize(final Object object, OutputStream outputStream) throws IOException {
+		Assert.notNull(outputStream, "\'outputSteam\' cannot be null");
+		final Output output = (outputStream instanceof Output ? (Output) outputStream : new Output(outputStream));
+		this.pool.run(new KryoCallback<Object>() {
+			@SuppressWarnings("unchecked")
+			public Object execute(Kryo kryo) {
+				doSerialize(kryo, object, output);
+				return Void.class;
+			}
+		});
+		output.close();
 	}
 
+	protected abstract void doSerialize(Kryo kryo, Object object, Output output);
+
+	protected abstract Object doDeserialize(Kryo kryo, Input input, Class<?> type);
+
+	protected abstract void configureKryoInstance(Kryo kryo);
+
 	/**
-	 * Deserialize an object when the type is known
-	 *
-	 * @param inputStream the input stream containing the serialized object
+	 * Deserialize an object of a given type given a byte array
+	 * @param bytes the byte array containing the serialized object
+	 * @param type the object's class
 	 * @return the object
 	 * @throws IOException
 	 */
 	@Override
-	public T deserialize(InputStream inputStream) throws IOException {
-		final Input input = new Input(inputStream);
+	public Object deserialize(byte[] bytes, Class<?> type) throws IOException {
+		final Input input = new Input(bytes);
 		try {
-			T result = pool.run(new KryoCallback<T>() {
-				@Override
-				public T execute(Kryo kryo) {
-					return doDeserialize(kryo, input);
-				}
-			});
-			return result;
-		} finally {
+			return deserialize(input, type);
+		}
+		finally {
 			input.close();
 		}
 	}
 
-	protected abstract void doSerialize(Kryo kryo, T object, Output output);
-
-	protected abstract T doDeserialize(Kryo kryo, Input input);
-
-	protected void configureKryoInstance(Kryo kryo) {
+	@Override
+	public Object deserialize(InputStream inputStream, final Class<?> type) throws IOException {
+		Assert.notNull(inputStream, "\'inputStream\' cannot be null");
+		final Input input = (inputStream instanceof Input ? (Input) inputStream : new Input(inputStream));
+		Object result = null;
+		try {
+			result = this.pool.run(new KryoCallback<Object>() {
+				@SuppressWarnings("unchecked")
+				public Object execute(Kryo kryo) {
+					return doDeserialize(kryo, input, type);
+				}
+			});
+		}
+		finally {
+			input.close();
+		}
+		return result;
 	}
+
 }

spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java

@@ -1,107 +0,0 @@
-/*
- * Copyright 2013-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.xd.dirt.integration.bus.serializer.kryo;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.lang.reflect.ParameterizedType;
-
-import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
-
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.pool.KryoCallback;
-
-/**
- * Base class for Codecs using {@link com.esotericsoftware.kryo.Kryo} to serialize arbitrary types
- *
- * @author David Turanski
- * @since 1.0
- */
-abstract class AbstractKryoMultiTypeCodec<T> extends AbstractKryoCodec<T> implements MultiTypeCodec<T> {
-
-	/**
-	 * Deserialize an object of a given type given a byte array
-	 *
-	 * @param bytes the byte array containing the serialized object
-	 * @param type the object's class
-	 * @return the object
-	 * @throws IOException
-	 */
-	@Override
-	public T deserialize(byte[] bytes, Class<? extends T> type) throws IOException {
-		final Input input = new Input(bytes);
-		try {
-			return deserialize(input, type);
-		} finally {
-			input.close();
-		}
-	}
-
-	/**
-	 * Deserialize an object of a given type given an InputStream
-	 *
-	 * @param inputStream the input stream containing the serialized object
-	 * @param type the object's class
-	 * @return the object
-	 * @throws IOException
-	 */
-	@Override
-	public T deserialize(InputStream inputStream, final Class<? extends T> type) throws IOException {
-		final Input input = new Input(inputStream);
-		try {
-			return deserialize(input, type);
-		} finally {
-			input.close();
-		}
-	}
-
-	/**
-	 * Deserialize an object of a given type given an Input.
-	 *
-	 * @param input the Kryo input stream containing the serialized object
-	 * @param type the object's class
-	 * @return the object
-	 * @throws IOException
-	 */
-	protected T deserialize(final Input input, final Class<? extends T> type) throws IOException {
-		return pool.run(new KryoCallback<T>() {
-
-			@Override
-			public T execute(Kryo kryo) {
-				return doDeserialize(kryo, input, type);
-			}
-		});
-	}
-
-	/**
-	 * Infers the type from this class's generic type argument
-	 * @param kryo the Kryo
-	 * @param input the input
-	 * @return the object
-	 */
-	@Override
-	@SuppressWarnings("unchecked")
-	protected T doDeserialize(Kryo kryo, Input input) {
-		Class<T> type = (Class<T>) (
-				(ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
-		return doDeserialize(kryo, input, type);
-	}
-
-	protected abstract T doDeserialize(Kryo kryo, Input input, Class<? extends T> type);
-
-}

spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java

@@ -29,32 +29,58 @@ import org.springframework.util.CollectionUtils;
 /**
  * Kryo Codec that can serialize and deserialize arbitrary types. Classes and associated
  * {@link com.esotericsoftware.kryo.Serializer}s may be registered via
- * {@link KryoRegistrar}s.
+ * {@link org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoRegistrar}s.
  * @author David Turanski
  * @since 1.0
  */
-public class PojoCodec extends AbstractKryoMultiTypeCodec<Object> {
+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
+	 * @param kryoRegistrar the registrar.
 	 */
 	public PojoCodec(KryoRegistrar kryoRegistrar) {
-		this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null);
+		this(kryoRegistrar != null ? Collections.singletonList(kryoRegistrar) : null, true);
 	}
 
 	/**
 	 * Create an instance with zero to many KryoRegistrars.
-	 * @param 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) {
 		kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null :
 				new CompositeKryoRegistrar(kryoRegistrars);
+		this.useReferences = useReferences;
 	}
 
 	@Override
@@ -64,15 +90,15 @@ public class PojoCodec extends AbstractKryoMultiTypeCodec<Object> {
 
 
 	@Override
-	protected Object doDeserialize(Kryo kryo, Input input, Class<? extends Object> type) {
+	protected Object doDeserialize(Kryo kryo, Input input, Class<?> type) {
 		return kryo.readObject(input, type);
 	}
 
 	@Override
 	protected void configureKryoInstance(Kryo kryo) {
-		super.configureKryoInstance(kryo);
 		if (kryoRegistrar != null) {
 			kryoRegistrar.registerTypes(kryo);
 		}
+		kryo.setReferences(useReferences);
 	}
 }