GH-1273 Added map flattening for binder configuration properties

- this ensures that properties are passed to child AC in an appropriate format (i.e., --foo.bar.baz=goo)
- changed BinderProperties.environment to be a Map<String, Object>

Resolves #1273
Resolves #1276

polish

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -26,12 +26,13 @@ import java.util.Properties;
  * systems of the same type.
  *
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
  */
 public class BinderConfiguration {
 
 	private final String binderType;
 
-	private final Map<Object, Object> properties;
+	private final Map<String, Object> properties;
 
 	private final boolean inheritEnvironment;
 
@@ -45,7 +46,7 @@ public class BinderConfiguration {
 	 * @param defaultCandidate whether the binder should be considered as a candidate when
 	 * determining a default
 	 */
-	public BinderConfiguration(String binderType, Map<Object, Object> properties, boolean inheritEnvironment,
+	public BinderConfiguration(String binderType, Map<String, Object> properties, boolean inheritEnvironment,
 			boolean defaultCandidate) {
 		this.binderType = binderType;
 		this.properties = properties;
@@ -57,7 +58,7 @@ public class BinderConfiguration {
 		return binderType;
 	}
 
-	public Map<Object, Object> getProperties() {
+	public Map<String, Object> getProperties() {
 		return properties;
 	}
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java

@@ -165,11 +165,14 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 			Assert.state(binderConfiguration != null, "Unknown binder configuration: " + configurationName);
 			BinderType binderType = this.binderTypeRegistry.get(binderConfiguration.getBinderType());
 			Assert.notNull(binderType, "Binder type " + binderConfiguration.getBinderType() + " is not defined");
-			Map<Object, Object> binderProperties = binderConfiguration.getProperties();
+			
+			Map<String, String> binderProperties = new HashMap<>();
+			this.flatten(null, binderConfiguration.getProperties(), binderProperties);
+
 			// Convert all properties to arguments, so that they receive maximum
 			// precedence
 			ArrayList<String> args = new ArrayList<>();
-			for (Map.Entry<Object, Object> property : binderProperties.entrySet()) {
+			for (Map.Entry<String, String> property : binderProperties.entrySet()) {
 				args.add(String.format("--%s=%s", property.getKey(), property.getValue()));
 			}
 			// Initialize the domain with a unique name based on the bootstrapping context
@@ -211,6 +214,20 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 		}
 		return (Binder<T, ?, ?>) this.binderInstanceCache.get(configurationName).getKey();
 	}
+	
+	/**
+	 * Ensures that nested properties are flattened (i.e., foo.bar=baz instead of foo={bar=baz})
+	 */
+	@SuppressWarnings("unchecked")
+	private void flatten(String propertyName, Object value, Map<String, String> flattenedProperties) {
+		if (value instanceof Map) {
+			((Map<Object, Object>) value)
+				.forEach((k, v) -> flatten((propertyName != null ? propertyName + "." : "") + k, v, flattenedProperties));
+		}
+		else {
+			flattenedProperties.put(propertyName, value.toString());
+		}
+	}
 
 	/**
 	 * A listener that can be registered with the {@link DefaultBinderFactory} that allows

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.config;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;
 
 /**
  * Contains the properties of a binder.
@@ -38,7 +39,7 @@ public class BinderProperties {
 	/**
 	 * Root for a set of properties that can be used to customize the environment of the binder. 
 	 */
-	private Map<Object, Object> environment = new HashMap<>();
+	private Map<String, Object> environment = new HashMap<>();
 
 	/**
 	 * Whether the configuration will inherit the environment of the application itself. Default: true
@@ -59,7 +60,7 @@ public class BinderProperties {
 		this.type = name;
 	}
 
-	public Map<Object, Object> getEnvironment() {
+	public Map<String, Object> getEnvironment() {
 		return environment;
 	}
 
@@ -69,10 +70,10 @@ public class BinderProperties {
 	@Deprecated
 	public void setEnvironment(Properties environment) {
 		this.environment.clear();
-		this.environment.putAll(environment);
+		this.environment.putAll(environment.entrySet().stream().collect(Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue())));
 	}
 
-	public void setEnvironment(Map<Object, Object> environment) {
+	public void setEnvironment(Map<String, Object> environment) {
 		this.environment = environment;
 	}