Use Jackson ObjectMapper from AutoConfiguration

- For the message converters that use Jackson ObjectMapper, autowire the ObjectMapper from JacksonAutoConfiguration
   - This will allow configuration of ObjectMapper using boot configuration properties `JacksonProperties`

 - Add test

This resolves #466

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java

@@ -16,6 +16,7 @@
 package org.springframework.cloud.stream.config;
 
 import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
@@ -39,6 +40,9 @@ import org.springframework.messaging.MessagingException;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.tuple.Tuple;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
 /**
  * @author Ilayaperumal Gopinathan
  */
@@ -49,6 +53,9 @@ public class MessageChannelConfigurerTests {
 	@Autowired @Bindings(TestSink.class)
 	private Sink testSink;
 
+	@Autowired
+	private ObjectMapper objectMapper;
+
 	@Test
 	public void testMessageConverterConfigurer() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -66,6 +73,12 @@ public class MessageChannelConfigurerTests {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		testSink.input().unsubscribe(messageHandler);
 	}
+
+	@Test
+	public void testObjectMapperConfig() {
+		assertNotNull( "ObjectMapper should exist", objectMapper);
+		assertTrue("SerializationFeature 'WRITE_DATES_AS_TIMESTAMPS' should be disabled", !objectMapper.getSerializationConfig().isEnabled(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS));
+	}
 	
 	@EnableBinding(Sink.class)
 	@EnableAutoConfiguration

spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties

@@ -1,2 +1,3 @@
 spring.cloud.stream.bindings.input.destination=configure1
 spring.cloud.stream.bindings.input.contentType=application/x-spring-tuple
+spring.jackson.serialization.WRITE_DATES_AS_TIMESTAMPS:false

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

@@ -65,6 +65,8 @@ import org.springframework.messaging.handler.annotation.support.DefaultMessageHa
 import org.springframework.tuple.spel.TuplePropertyAccessor;
 import org.springframework.util.CollectionUtils;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 /**
  * Configuration class that provides necessary beans for {@link MessageChannel} binding.
  *
@@ -83,6 +85,9 @@ public class ChannelBindingServiceConfiguration {
 	@Autowired
 	private MessageBuilderFactory messageBuilderFactory;
 
+	@Lazy @Autowired(required = false)
+	private ObjectMapper objectMapper;
+
 	/**
 	 * User defined custom message converters
 	 */
@@ -165,7 +170,7 @@ public class ChannelBindingServiceConfiguration {
 		if (!CollectionUtils.isEmpty(customMessageConverters)) {
 			messageConverters.addAll(Collections.unmodifiableCollection(customMessageConverters));
 		}
-		return new CompositeMessageConverterFactory(messageConverters);
+		return new CompositeMessageConverterFactory(messageConverters, objectMapper);
 	}
 
 	// IMPORTANT: Nested class to avoid instantiating all of the above early
@@ -230,7 +235,8 @@ public class ChannelBindingServiceConfiguration {
 	}
 
 	@Bean
-	public static StreamListenerAnnotationBeanPostProcessor bindToAnnotationBeanPostProcessor(@Lazy BinderAwareChannelResolver binderAwareChannelResolver, @Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
+	public static StreamListenerAnnotationBeanPostProcessor bindToAnnotationBeanPostProcessor(@Lazy BinderAwareChannelResolver binderAwareChannelResolver,
+			@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
 		DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
 		messageHandlerMethodFactory.setMessageConverter(compositeMessageConverterFactory.getMessageConverterForAllRegistered());
 		messageHandlerMethodFactory.afterPropertiesSet();

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java

@@ -29,6 +29,8 @@ import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.ObjectUtils;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 
 /**
  * A factory for creating an instance of {@link CompositeMessageConverter} for a given target MIME type
@@ -38,16 +40,20 @@ import org.springframework.util.ObjectUtils;
  */
 public class CompositeMessageConverterFactory {
 
+	private final ObjectMapper objectMapper;
+
 	private final List<AbstractFromMessageConverter> converters;
 
 	public CompositeMessageConverterFactory() {
-		this(Collections.<AbstractFromMessageConverter>emptyList());
+		this(Collections.<AbstractFromMessageConverter>emptyList(), new ObjectMapper());
 	}
 
 	/**
 	 * @param customConverters a list of {@link AbstractFromMessageConverter}
 	 */
-	public CompositeMessageConverterFactory(List<? extends AbstractFromMessageConverter> customConverters) {
+	public CompositeMessageConverterFactory(List<? extends AbstractFromMessageConverter> customConverters,
+			ObjectMapper objectMapper) {
+		this.objectMapper = objectMapper;
 		if (!CollectionUtils.isEmpty(customConverters)) {
 			this.converters = new ArrayList<>(customConverters);
 		}
@@ -58,12 +64,11 @@ public class CompositeMessageConverterFactory {
 	}
 
 
-
 	private void initDefaultConverters() {
 		this.converters.add(new JsonToTupleMessageConverter());
-		this.converters.add(new TupleToJsonMessageConverter());
-		this.converters.add(new JsonToPojoMessageConverter());
-		this.converters.add(new PojoToJsonMessageConverter());
+		this.converters.add(new TupleToJsonMessageConverter(objectMapper));
+		this.converters.add(new JsonToPojoMessageConverter(objectMapper));
+		this.converters.add(new PojoToJsonMessageConverter(objectMapper));
 		this.converters.add(new ByteArrayToStringMessageConverter());
 		this.converters.add(new StringToByteArrayMessageConverter());
 		this.converters.add(new PojoToStringMessageConverter());

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 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.
@@ -16,23 +16,23 @@
 
 package org.springframework.cloud.stream.converter;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializationFeature;
-
 import org.springframework.messaging.Message;
 import org.springframework.util.MimeTypeUtils;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 /**
  *
  * @author David Turanski
+ * @author Ilayaperumal Gopinathan
  */
 public class JsonToPojoMessageConverter extends AbstractFromMessageConverter {
 
-	private final ObjectMapper mapper = new ObjectMapper();
+	private final ObjectMapper objectMapper;
 
-	public JsonToPojoMessageConverter() {
+	public JsonToPojoMessageConverter(ObjectMapper objectMapper) {
 		super(MimeTypeUtils.APPLICATION_JSON, MessageConverterUtils.X_JAVA_OBJECT);
-		mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+		this.objectMapper = (objectMapper != null) ? objectMapper : new ObjectMapper();
 	}
 
 	@Override
@@ -52,10 +52,10 @@ public class JsonToPojoMessageConverter extends AbstractFromMessageConverter {
 			Object payload = message.getPayload();
 
 			if (payload instanceof byte[]) {
-				result = mapper.readValue((byte[]) payload, targetClass);
+				result = objectMapper.readValue((byte[]) payload, targetClass);
 			}
 			else if (payload instanceof String) {
-				result = mapper.readValue((String) payload, targetClass);
+				result = objectMapper.readValue((String) payload, targetClass);
 			}
 		}
 		catch (Exception e) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 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.
@@ -16,14 +16,13 @@
 
 package org.springframework.cloud.stream.converter;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializationFeature;
-
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.messaging.Message;
 import org.springframework.util.MimeTypeUtils;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 
 /**
  * A {@link org.springframework.messaging.converter.MessageConverter}
@@ -31,17 +30,18 @@ import org.springframework.util.MimeTypeUtils;
  *
  * @author David Turanski
  * @author David Liu
+ * @author Ilayaperumal Gopinathan
  */
 public class PojoToJsonMessageConverter extends AbstractFromMessageConverter {
 
-	private final ObjectMapper mapper = new ObjectMapper();
+	private final ObjectMapper objectMapper;
 
 	@Value("${typeconversion.json.prettyPrint:false}")
 	private volatile boolean prettyPrint;
 
-	public PojoToJsonMessageConverter() {
+	public PojoToJsonMessageConverter(ObjectMapper objectMapper) {
 		super(MimeTypeUtils.APPLICATION_JSON);
-		mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+		this.objectMapper = (objectMapper != null) ? objectMapper : new ObjectMapper();
 	}
 
 	@Override
@@ -64,10 +64,10 @@ public class PojoToJsonMessageConverter extends AbstractFromMessageConverter {
 		Object result;
 		try {
 			if (prettyPrint) {
-				result = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(message.getPayload());
+				result = objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(message.getPayload());
 			}
 			else {
-				result = mapper.writeValueAsString(message.getPayload());
+				result = objectMapper.writeValueAsString(message.getPayload());
 			}
 		}
 		catch (JsonProcessingException e) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 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.
@@ -18,32 +18,35 @@ package org.springframework.cloud.stream.converter;
 
 import java.io.IOException;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializationFeature;
-
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.messaging.Message;
 import org.springframework.tuple.Tuple;
 import org.springframework.util.MimeTypeUtils;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 
 /**
  * A {@link org.springframework.messaging.converter.MessageConverter}
  * to convert a {@link Tuple} to a JSON String
  *
  * @author David Turanski
+ * @author Ilayaperumal Gopinathan
  */
 public class TupleToJsonMessageConverter extends AbstractFromMessageConverter {
 
 	@Value("${typeconversion.json.prettyPrint:false}")
 	private volatile boolean prettyPrint;
 
+	private final ObjectMapper objectMapper;
+
 	public void setPrettyPrint(boolean prettyPrint) {
 		this.prettyPrint = prettyPrint;
 	}
 
-	public TupleToJsonMessageConverter() {
+	public TupleToJsonMessageConverter(ObjectMapper objectMapper) {
 		super(MimeTypeUtils.APPLICATION_JSON);
+		this.objectMapper = (objectMapper != null) ? objectMapper : new ObjectMapper();
 	}
 
 	@Override
@@ -61,11 +64,9 @@ public class TupleToJsonMessageConverter extends AbstractFromMessageConverter {
 		Tuple t = (Tuple) message.getPayload();
 		String json;
 		if (prettyPrint) {
-			ObjectMapper mapper = new ObjectMapper();
-			mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
 			try {
-				Object tmp = mapper.readValue(t.toString(), Object.class);
-				json = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(tmp);
+				Object tmp = objectMapper.readValue(t.toString(), Object.class);
+				json = objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(tmp);
 			}
 			catch (IOException e) {
 				logger.error(e.getMessage(), e);