GH-1614 Added Expression serializer to ObjectMapper

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;
 	}
 

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<C extends ConsumerProperties,
 		this.provisioningProvider = provisioningProvider;
 		this.containerCustomizer = containerCustomizer == null ? (c, q, g) -> {
 		} : 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<C extends ConsumerProperties,
 
 		MessageSource<?> 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<C extends ConsumerProperties,
 
 	}
 
+	@SuppressWarnings("serial")
+	private static class ExpressionSerializer extends StdSerializer<Expression> {
+
+		protected ExpressionSerializer(Class<Expression> t) {
+			super(t);
+		}
+
+		@Override
+		public void serialize(Expression value, JsonGenerator gen,
+				SerializerProvider provider) throws IOException {
+			gen.writeString(value.getExpressionString());
+		}
+
+	}
+
 }