Use conversion service when binding default properties

- When binding the default BindingProperties, we use RelaxedDataBinder without any specific conversion service. This makes the `String` to SpEL `Expression` conversion failing when the producer properties such as `partitionKey/SelectorExpression` are bound.
 - Update the underlying conversion service (which includes the SpELConverter via EnableBinding) into RelaxedDataBinder
 - Update test

Resolves #1040

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2017 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.
@@ -33,10 +33,8 @@ import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.EnvironmentAware;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.env.ConfigurableEnvironment;
-import org.springframework.core.env.Environment;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.util.Assert;
 
@@ -48,7 +46,7 @@ import org.springframework.util.Assert;
  */
 @ConfigurationProperties("spring.cloud.stream")
 @JsonInclude(Include.NON_DEFAULT)
-public class BindingServiceProperties implements ApplicationContextAware, EnvironmentAware, InitializingBean {
+public class BindingServiceProperties implements ApplicationContextAware, InitializingBean {
 
 	private ConversionService conversionService;
 
@@ -120,18 +118,15 @@ public class BindingServiceProperties implements ApplicationContextAware, Enviro
 	public void setApplicationContext(ApplicationContext applicationContext)
 			throws BeansException {
 		this.applicationContext = (ConfigurableApplicationContext) applicationContext;
-	}
-
-	@Override
-	public void setEnvironment(Environment environment) {
-		if (environment instanceof ConfigurableEnvironment) {
+		if (this.applicationContext.getEnvironment() instanceof ConfigurableEnvironment) {
 			// override the bindings store with the environment-initializing version if in
 			// a Spring context
 			Map<String, BindingProperties> delegate = new TreeMap<String, BindingProperties>(
 					String.CASE_INSENSITIVE_ORDER);
 			delegate.putAll(this.bindings);
-			this.bindings = new EnvironmentEntryInitializingTreeMap<>((ConfigurableEnvironment) environment,
-					BindingProperties.class, "spring.cloud.stream.default", delegate);
+			this.bindings = new EnvironmentEntryInitializingTreeMap<>(this.applicationContext.getEnvironment(),
+					BindingProperties.class, "spring.cloud.stream.default", delegate,
+					IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()));
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -23,6 +23,7 @@ import java.util.Set;
 import org.springframework.beans.BeanUtils;
 import org.springframework.boot.bind.PropertySourcesPropertyValues;
 import org.springframework.boot.bind.RelaxedDataBinder;
+import org.springframework.core.convert.ConversionService;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.util.Assert;
 
@@ -38,6 +39,7 @@ import org.springframework.util.Assert;
  * This implementation is not thread safe.
  *
  * @author Marius Bogoevici
+ * @author Ilayaperumal Gopinathan
  */
 public class EnvironmentEntryInitializingTreeMap<T> extends AbstractMap<String, T> {
 
@@ -49,6 +51,8 @@ public class EnvironmentEntryInitializingTreeMap<T> extends AbstractMap<String,
 
 	private final Map<String, T> delegate;
 
+	private final ConversionService conversionService;
+
 	/**
 	 * Constructs the map.
 	 * 
@@ -56,9 +60,11 @@ public class EnvironmentEntryInitializingTreeMap<T> extends AbstractMap<String,
 	 * @param entryClass the entry class
 	 * @param defaultsPrefix the prefix for initializing the properties
 	 * @param delegate the actual map that stores the values
+	 * @param conversionService the conversion service to use when binding the default
+	 * property values.
 	 */
 	public EnvironmentEntryInitializingTreeMap(ConfigurableEnvironment environment, Class<T> entryClass,
-			String defaultsPrefix, Map<String, T> delegate) {
+			String defaultsPrefix, Map<String, T> delegate, ConversionService conversionService) {
 		Assert.notNull(environment, "The environment cannot be null");
 		Assert.notNull(entryClass, "The entry class cannot be null");
 		Assert.notNull(defaultsPrefix, "The prefix for the property defaults cannot be null");
@@ -67,6 +73,7 @@ public class EnvironmentEntryInitializingTreeMap<T> extends AbstractMap<String,
 		this.entryClass = entryClass;
 		this.defaultsPrefix = defaultsPrefix;
 		this.delegate = delegate;
+		this.conversionService = conversionService;
 	}
 
 	@Override
@@ -74,6 +81,7 @@ public class EnvironmentEntryInitializingTreeMap<T> extends AbstractMap<String,
 		if (!this.delegate.containsKey(key) && key instanceof String) {
 			T entry = BeanUtils.instantiate(entryClass);
 			RelaxedDataBinder defaultsDataBinder = new RelaxedDataBinder(entry, defaultsPrefix);
+			defaultsDataBinder.setConversionService(this.conversionService);
 			defaultsDataBinder.bind(new PropertySourcesPropertyValues(environment.getPropertySources()));
 			this.delegate.put((String) key, entry);
 		}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2017 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.
@@ -37,7 +37,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringBootTest(classes = SourceBindingWithGlobalPropertiesOnlyTest.TestSource.class, properties = {
-		"spring.cloud.stream.default.contentType=application/json" })
+		"spring.cloud.stream.default.contentType=application/json",
+		"spring.cloud.stream.default.producer.partitionKeyExpression=key" })
 public class SourceBindingWithGlobalPropertiesOnlyTest {
 
 	@Autowired
@@ -48,6 +49,8 @@ public class SourceBindingWithGlobalPropertiesOnlyTest {
 	public void testGlobalPropertiesSet() {
 		BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(Source.OUTPUT);
 		Assertions.assertThat(bindingProperties.getContentType()).isEqualTo("application/json");
+		Assertions.assertThat(bindingProperties.getProducer()).isNotNull();
+		Assertions.assertThat(bindingProperties.getProducer().getPartitionKeyExpression().getExpressionString()).isEqualTo("key");
 	}
 
 	@EnableBinding(Source.class)