diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java index 3e7a43422..6f7dd06e9 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java @@ -272,13 +272,12 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag Schema schema; schema = extractSchemaForWriting(payload); - ParsedSchema parsedSchema = this.getCache(REFERENCE_CACHE_NAME) - .get(schema, ParsedSchema.class); + ParsedSchema parsedSchema = this.getCache(REFERENCE_CACHE_NAME).get(schema, + ParsedSchema.class); if (parsedSchema == null) { parsedSchema = new ParsedSchema(schema); - this.getCache(REFERENCE_CACHE_NAME).putIfAbsent(schema, - parsedSchema); + this.getCache(REFERENCE_CACHE_NAME).putIfAbsent(schema, parsedSchema); } if (parsedSchema.getRegistration() == null) { @@ -315,8 +314,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag if (schemaContent != null) { Schema schema = new Schema.Parser().parse(schemaContent); parsedSchema = new ParsedSchema(schema); - this.getCache(REFERENCE_CACHE_NAME) - .putIfAbsent(schemaReference, parsedSchema); + this.getCache(REFERENCE_CACHE_NAME).putIfAbsent(schemaReference, + parsedSchema); } } if (parsedSchema != null) { @@ -356,8 +355,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag else { schema = ReflectData.get().getSchema(payload.getClass()); } - this.getCache(REFLECTION_CACHE_NAME) - .put(payload.getClass().getName(), schema); + this.getCache(REFLECTION_CACHE_NAME).put(payload.getClass().getName(), + schema); } } return schema; @@ -392,9 +391,10 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag private Cache getCache(String name) { Cache cache = this.cacheManager.getCache(""); - Assert.notNull(cache, "Cache by the name '" + name + "' is not present in this CacheManager - '" - + this.cacheManager + "'. Typically caches are auto-created by the CacheManagers. " - + "Consider reporting it as an issue to the developer of this CacheManager."); + Assert.notNull(cache, "Cache by the name '" + name + + "' is not present in this CacheManager - '" + this.cacheManager + + "'. Typically caches are auto-created by the CacheManagers. " + + "Consider reporting it as an issue to the developer of this CacheManager."); return cache; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 6a91ed28f..650294df1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -16,13 +16,18 @@ package org.springframework.cloud.stream.binder; +import java.io.IOException; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializerProvider; +import com.fasterxml.jackson.databind.module.SimpleModule; +import com.fasterxml.jackson.databind.ser.std.StdSerializer; import org.apache.commons.logging.Log; import org.reactivestreams.Publisher; @@ -43,6 +48,7 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.Lifecycle; +import org.springframework.expression.Expression; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.AbstractSubscribableChannel; import org.springframework.integration.channel.DirectChannel; @@ -136,6 +142,11 @@ public abstract class AbstractMessageChannelBinder { } : containerCustomizer; + + SimpleModule module = new SimpleModule(); + module.addSerializer(Expression.class, + new ExpressionSerializer(Expression.class)); + objectMapper.registerModule(module); } protected ApplicationEventPublisher getApplicationEventPublisher() { @@ -443,7 +454,8 @@ public abstract class AbstractMessageChannelBinder messageSource = resources.getSource(); if (messageSource instanceof BeanFactoryAware) { - ((BeanFactoryAware)messageSource).setBeanFactory(getApplicationContext().getBeanFactory()); + ((BeanFactoryAware) messageSource) + .setBeanFactory(getApplicationContext().getBeanFactory()); } bindingTarget.setSource(messageSource); if (resources.getErrorInfrastructure() != null) { @@ -1133,4 +1145,19 @@ public abstract class AbstractMessageChannelBinder { + + protected ExpressionSerializer(Class t) { + super(t); + } + + @Override + public void serialize(Expression value, JsonGenerator gen, + SerializerProvider provider) throws IOException { + gen.writeString(value.getExpressionString()); + } + + } + }