diff --git a/spring-cloud-stream-codec/pom.xml b/spring-cloud-stream-codec/pom.xml index 1844cc3a1..c36b94656 100644 --- a/spring-cloud-stream-codec/pom.xml +++ b/spring-cloud-stream-codec/pom.xml @@ -40,6 +40,14 @@ org.springframework.boot spring-boot-starter-logging + + org.springframework.boot + spring-boot-autoconfigure + + + com.fasterxml.jackson.core + jackson-annotations + org.springframework.boot spring-boot-starter-test diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/CodecConfiguration.java b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/CodecConfiguration.java new file mode 100644 index 000000000..94af0efea --- /dev/null +++ b/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 kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar + .class); + return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values()), kryoCodecProperties().isReferences()); + } + + @Bean + public KryoRegistrar fileRegistrar() { + return new FileKryoRegistrar(); + } +} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/KryoCodecProperties.java b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/KryoCodecProperties.java new file mode 100644 index 000000000..bd5d415ed --- /dev/null +++ b/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; + } + +} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java index 12249e157..a967f22b6 100644 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoCodec.java +++ b/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 extends AbstractCodec { - - private final KryoFactory factory; +public abstract class AbstractKryoCodec implements MultiTypeCodec { 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 extends AbstractCodec { /** * 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() { - @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() { + @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() { - @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() { + @SuppressWarnings("unchecked") + public Object execute(Kryo kryo) { + return doDeserialize(kryo, input, type); + } + }); + } + finally { + input.close(); + } + return result; } + } diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java deleted file mode 100644 index 89d9bf3bf..000000000 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/AbstractKryoMultiTypeCodec.java +++ /dev/null @@ -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 extends AbstractKryoCodec implements MultiTypeCodec { - - /** - * 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 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 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 type) throws IOException { - return pool.run(new KryoCallback() { - - @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 type = (Class) ( - (ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0]; - return doDeserialize(kryo, input, type); - } - - protected abstract T doDeserialize(Kryo kryo, Input input, Class type); - -} diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java b/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java index 71800c78a..a532ba8e3 100644 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/xd/dirt/integration/bus/serializer/kryo/PojoCodec.java +++ b/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 { +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 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 kryoRegistrars, boolean useReferences) { kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null : new CompositeKryoRegistrar(kryoRegistrars); + this.useReferences = useReferences; } @Override @@ -64,15 +90,15 @@ public class PojoCodec extends AbstractKryoMultiTypeCodec { @Override - protected Object doDeserialize(Kryo kryo, Input input, Class 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); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java index 1e2552c44..e00c651b9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java @@ -24,9 +24,10 @@ import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import org.springframework.cloud.stream.config.AggregateBuilderConfiguration; +import org.springframework.cloud.stream.config.ChannelBindingAdapterConfiguration; +import org.springframework.cloud.stream.config.CodecConfiguration; import org.springframework.cloud.stream.config.EnableModuleConfiguration; import org.springframework.cloud.stream.config.LifecycleConfiguration; -import org.springframework.cloud.stream.config.ChannelBindingAdapterConfiguration; import org.springframework.cloud.stream.config.RabbitServiceConfiguration; import org.springframework.cloud.stream.config.RedisServiceConfiguration; import org.springframework.context.annotation.Configuration; @@ -35,9 +36,9 @@ import org.springframework.integration.annotation.MessageEndpoint; /** * Annotation that identifies a class as a module. - * * @author Dave Syer * @author Marius Bogoevici + * @author David Turanski */ @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @@ -45,9 +46,9 @@ import org.springframework.integration.annotation.MessageEndpoint; @Inherited @Configuration @MessageEndpoint -@Import({ RedisServiceConfiguration.class, RabbitServiceConfiguration.class, - ChannelBindingAdapterConfiguration.class, LifecycleConfiguration.class, - AggregateBuilderConfiguration.class, EnableModuleConfiguration.class}) +@Import({RedisServiceConfiguration.class, RabbitServiceConfiguration.class, + ChannelBindingAdapterConfiguration.class, CodecConfiguration.class, LifecycleConfiguration.class, + AggregateBuilderConfiguration.class, EnableModuleConfiguration.class}) public @interface EnableModule { } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java index 9615685d8..b1af42fc1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java @@ -78,7 +78,7 @@ public class ChannelBindingAdapterConfiguration { ChannelBindingAdapter adapter = new ChannelBindingAdapter(this.module, this.messageBus); adapter.setOutputChannels(getOutputChannels()); adapter.setInputChannels(getInputChannels()); - if (this.channelLocator!=null) { + if (this.channelLocator != null) { adapter.setChannelLocator(this.channelLocator); } return adapter; @@ -102,7 +102,7 @@ public class ChannelBindingAdapterConfiguration { BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(name); // for now, just assume that the beans are at least AbstractBeanDefinition if (beanDefinition instanceof AbstractBeanDefinition - && ((AbstractBeanDefinition)beanDefinition).getQualifier(Output.class.getName()) != null) { + && ((AbstractBeanDefinition) beanDefinition).getQualifier(Output.class.getName()) != null) { channels.add(new OutputChannelBinding(name)); } } @@ -116,7 +116,7 @@ public class ChannelBindingAdapterConfiguration { BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(name); // for now, just assume that the beans are at least AbstractBeanDefinition if (beanDefinition instanceof AbstractBeanDefinition - && ((AbstractBeanDefinition)beanDefinition).getQualifier(Input.class.getName()) != null) { + && ((AbstractBeanDefinition) beanDefinition).getQualifier(Input.class.getName()) != null) { channels.add(new InputChannelBinding(name)); } } @@ -167,32 +167,5 @@ public class ChannelBindingAdapterConfiguration { } } - - } - - @ConditionalOnMissingBean(ChannelBindingProperties.class) - protected static class ModulePropertiesConfiguration { - @Bean(name = "spring.cloud.channels.CONFIGURATION_PROPERTIES") - public ChannelBindingProperties moduleProperties() { - return new ChannelBindingProperties(); - } - } - - protected static class CodecConfiguration { - @Autowired - ApplicationContext applicationContext; - - @Bean - @ConditionalOnMissingBean(name = "codec") - public MultiTypeCodec codec() { - Map kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar - .class); - return new PojoCodec(new ArrayList<>(kryoRegistrarMap.values())); - } - - @Bean - public KryoRegistrar fileRegistrar() { - return new FileKryoRegistrar(); - } } }