Typesafe BindingProperties

- Create BindingProperties and Converter to replace the current Map-based logic

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java

@@ -17,7 +17,7 @@
 package org.springframework.cloud.stream.binding;
 
 import org.springframework.cloud.stream.binder.Binder;
-import org.springframework.cloud.stream.config.ChannelBindingProperties;
+import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -34,34 +34,34 @@ public class ChannelBindingService {
 
 	private Binder<MessageChannel> binder;
 
-	private ChannelBindingProperties channelBindingProperties;
+	private ChannelBindingServiceProperties channelBindingServiceProperties;
 
-	public ChannelBindingService(ChannelBindingProperties channelBindingProperties, Binder<MessageChannel> binder) {
-		this.channelBindingProperties = channelBindingProperties;
+	public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, Binder<MessageChannel> binder) {
+		this.channelBindingServiceProperties = channelBindingServiceProperties;
 		this.binder = binder;
 	}
 
 	public void bindConsumer(MessageChannel inputChannel, String inputChannelName) {
-		String channelBindingTarget = this.channelBindingProperties.getBindingDestination(inputChannelName);
+		String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName);
 		if (isChannelPubSub(channelBindingTarget)) {
 			this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget),
-					inputChannel, this.channelBindingProperties.getConsumerProperties(inputChannelName));
+					inputChannel, this.channelBindingServiceProperties.getConsumerProperties(inputChannelName));
 		}
 		else {
 			this.binder.bindConsumer(channelBindingTarget, inputChannel,
-					this.channelBindingProperties.getConsumerProperties(inputChannelName));
+					this.channelBindingServiceProperties.getConsumerProperties(inputChannelName));
 		}
 	}
 
 	public void bindProducer(MessageChannel outputChannel, String outputChannelName) {
-		String channelBindingTarget = this.channelBindingProperties.getBindingDestination(outputChannelName);
+		String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(outputChannelName);
 		if (isChannelPubSub(channelBindingTarget)) {
 			this.binder.bindPubSubProducer(removePrefix(channelBindingTarget),
-					outputChannel, this.channelBindingProperties.getProducerProperties(outputChannelName));
+					outputChannel, this.channelBindingServiceProperties.getProducerProperties(outputChannelName));
 		}
 		else {
 			this.binder.bindProducer(channelBindingTarget, outputChannel,
-					this.channelBindingProperties.getProducerProperties(outputChannelName));
+					this.channelBindingServiceProperties.getProducerProperties(outputChannelName));
 		}
 	}
 

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.config;
+
+/**
+ * Contains the properties of a binding.
+ *
+ * @author Marius Bogoevici
+ */
+public class BindingProperties {
+
+	private String destination;
+
+	private boolean partitioned = false;
+
+	private int partitionCount = 1;
+
+	private String partitionKeyExpression;
+
+	private String partitionKeyExtractorClass;
+
+	private String partitionSelectorClass;
+
+	private String partitionSelectorExpression;
+
+	public String getDestination() {
+		return destination;
+	}
+
+	public void setDestination(String destination) {
+		this.destination = destination;
+	}
+
+	public boolean isPartitioned() {
+		return partitioned;
+	}
+
+	public void setPartitioned(boolean partitioned) {
+		this.partitioned = partitioned;
+	}
+
+	public int getPartitionCount() {
+		return partitionCount;
+	}
+
+	public void setPartitionCount(int partitionCount) {
+		this.partitionCount = partitionCount;
+	}
+
+	public String getPartitionKeyExpression() {
+		return partitionKeyExpression;
+	}
+
+	public void setPartitionKeyExpression(String partitionKeyExpression) {
+		this.partitionKeyExpression = partitionKeyExpression;
+	}
+
+	public String getPartitionKeyExtractorClass() {
+		return partitionKeyExtractorClass;
+	}
+
+	public void setPartitionKeyExtractorClass(String partitionKeyExtractorClass) {
+		this.partitionKeyExtractorClass = partitionKeyExtractorClass;
+	}
+
+	public String getPartitionSelectorClass() {
+		return partitionSelectorClass;
+	}
+
+	public void setPartitionSelectorClass(String partitionSelectorClass) {
+		this.partitionSelectorClass = partitionSelectorClass;
+	}
+
+	public String getPartitionSelectorExpression() {
+		return partitionSelectorExpression;
+	}
+
+	public void setPartitionSelectorExpression(String partitionSelectorExpression) {
+		this.partitionSelectorExpression = partitionSelectorExpression;
+	}
+}

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.config;
+
+import org.springframework.core.convert.converter.Converter;
+
+/**
+ * Converter that transforms {@link String} expressions into {@link BindingProperties}. Useful for shorthand
+ * binding property configuration.
+ *
+ * @author Marius Bogoevici
+ */
+public class BindingPropertiesConverter implements Converter<String,BindingProperties> {
+
+	public BindingPropertiesConverter() {
+	}
+
+	@Override
+	public BindingProperties convert(String bindingConfiguration) {
+		BindingProperties bindingProperties = new BindingProperties();
+		// for now, just configure the destination - in the future do some more advanced parsing
+		bindingProperties.setDestination(bindingConfiguration);
+		return bindingProperties;
+	}
+}

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

@@ -21,6 +21,7 @@ import java.util.Properties;
 import org.springframework.beans.factory.BeanFactoryUtils;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
@@ -30,6 +31,7 @@ import org.springframework.cloud.stream.binding.ChannelBindingService;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.DependsOn;
+import org.springframework.core.convert.converter.Converter;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.core.DestinationResolver;
@@ -43,7 +45,7 @@ import org.springframework.messaging.core.DestinationResolver;
  * @author Ilayaperumal Gopinathan
  */
 @Configuration
-@EnableConfigurationProperties(ChannelBindingProperties.class)
+@EnableConfigurationProperties(ChannelBindingServiceProperties.class)
 public class ChannelBindingServiceConfiguration {
 
 	@Bean
@@ -52,9 +54,9 @@ public class ChannelBindingServiceConfiguration {
 	// already exists).
 	@ConditionalOnMissingBean(ChannelBindingService.class)
 	public ChannelBindingService bindingService(
-			ChannelBindingProperties channelBindingProperties,
+			ChannelBindingServiceProperties channelBindingServiceProperties,
 			Binder<MessageChannel> binder) {
-		return new ChannelBindingService(channelBindingProperties, binder);
+		return new ChannelBindingService(channelBindingServiceProperties, binder);
 	}
 
 	@Bean
@@ -69,6 +71,12 @@ public class ChannelBindingServiceConfiguration {
 		return new BinderAwareChannelResolver(binder, new Properties());
 	}
 
+	@Bean
+	@ConfigurationPropertiesBinding
+	public Converter<String,BindingProperties> bindingPropertiesConverter() {
+		return new BindingPropertiesConverter();
+	}
+
 	// IMPORTANT: Nested class to avoid instantiating all of the above early
 	@Configuration
 	protected static class PostProcessorConfiguration {

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

@@ -23,6 +23,7 @@ import java.util.TreeMap;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.stream.binder.BinderProperties;
+import org.springframework.util.StringUtils;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
@@ -33,16 +34,10 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include;
  */
 @ConfigurationProperties("spring.cloud.stream")
 @JsonInclude(Include.NON_DEFAULT)
-public class ChannelBindingProperties {
-
-	public static final String DESTINATION = "destination";
-
-	public static final String PARTITION_COUNT = "partitionCount";
-
-	public static final String PARTITIONED = "partitioned";
+public class ChannelBindingServiceProperties {
 
 	@Value("${INSTANCE_INDEX:${CF_INSTANCE_INDEX:0}}")
-	private int instanceIndex;
+	private int instanceIndex = 0;
 
 	private int instanceCount = 1;
 
@@ -50,7 +45,7 @@ public class ChannelBindingProperties {
 
 	private Properties producerProperties = new Properties();
 
-	private Map<String,Object> bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+	private Map<String,BindingProperties> bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
 
 	private Properties getConsumerProperties() {
 		return this.consumerProperties;
@@ -68,11 +63,11 @@ public class ChannelBindingProperties {
 		this.producerProperties = producerProperties;
 	}
 
-	public Map<String, Object> getBindings() {
+	public Map<String, BindingProperties> getBindings() {
 		return bindings;
 	}
 
-	public void setBindings(Map<String, Object> bindings) {
+	public void setBindings(Map<String, BindingProperties> bindings) {
 		this.bindings = bindings;
 	}
 
@@ -94,70 +89,27 @@ public class ChannelBindingProperties {
 
 
 	public String getBindingDestination(String channelName) {
-		Object binding = bindings.get(channelName);
+		BindingProperties bindingProperties = bindings.get(channelName);
 		// we may shortcut directly to the path
-		if (binding != null) {
-			if (binding instanceof String) {
-				return (String) binding;
-			}
-			else if (binding instanceof Map) {
-				Map<?, ?> bindingProperties = (Map<?, ?>) binding;
-				Object bindingPath = bindingProperties.get(DESTINATION);
-				if (bindingPath != null) {
-					return bindingPath.toString();
-				}
-			}
-		}
 		// just return the channel name if not found
-		return channelName;
+		return bindingProperties != null && StringUtils.hasText(bindingProperties.getDestination()) ?
+				bindingProperties.getDestination() : channelName;
 	}
 
 	public int getPartitionCount(String channelName) {
-		Object binding = bindings.get(channelName);
-		// we may shortcut directly to the path
-		if (binding instanceof Map) {
-			try {
-				Map<?, ?> bindingProperties = (Map<?, ?>) binding;
-				Object bindingPath = bindingProperties.get(PARTITION_COUNT);
-				if (bindingPath != null) {
-					return Integer.parseInt(bindingPath.toString());
-				}
-			} catch (NumberFormatException e) {
-				// ignore and just return 1
-			}
-		}
-		return 1;
+		return bindings.containsKey(channelName) ? bindings.get(channelName).getPartitionCount() : 1;
 	}
 
 	public boolean isPartitionedConsumer(String channelName) {
-		Object binding = bindings.get(channelName);
-		// if the setting is just a target shortcut
-		if (binding == null || binding instanceof String) {
-			return false;
-		}
-		else if (binding instanceof Map) {
-			Map<?, ?> bindingProperties = (Map<?, ?>) binding;
-			Object bindingPath = bindingProperties.get(PARTITIONED);
-			if (bindingPath != null) {
-				return Boolean.valueOf(bindingPath.toString());
-			}
-		}
-		// just return the channel name if not found
-		return false;
+		return bindings.containsKey(channelName) && bindings.get(channelName).isPartitioned();
 	}
 
 	public boolean isPartitionedProducer(String channelName) {
-		Object binding = bindings.get(channelName);
-		// if the setting is just a target shortcut
-		if (binding == null || binding instanceof String) {
-			return false;
-		}
-		else if (binding instanceof Map) {
-			Map<?, ?> bindingProperties = (Map<?, ?>) binding;
-			return bindingProperties.get(BinderProperties.PARTITION_KEY_EXPRESSION) != null
-					|| bindingProperties.get(BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS) != null;
-		}
-		return false;
+		BindingProperties bindingProperties = bindings.get(channelName);
+		return bindingProperties != null &&
+					(StringUtils.hasText(bindingProperties.getPartitionKeyExpression())
+					|| StringUtils.hasText(bindingProperties.getPartitionKeyExtractorClass()));
+
 	}
 
 	/**
@@ -195,14 +147,25 @@ public class ChannelBindingProperties {
 			channelProducerProperties.putAll(this.producerProperties);
 			channelProducerProperties.setProperty(BinderProperties.NEXT_MODULE_COUNT,
 					Integer.toString(getPartitionCount(outputChannelName)));
-			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
-					BinderProperties.PARTITION_KEY_EXPRESSION);
-			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
-					BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS);
-			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
-					BinderProperties.PARTITION_SELECTOR_CLASS);
-			copyChannelBindingProperty(outputChannelName, channelProducerProperties,
-					BinderProperties.PARTITION_SELECTOR_EXPRESSION);
+			BindingProperties bindingProperties = bindings.get(outputChannelName);
+			if (bindingProperties != null) {
+				if (bindingProperties.getPartitionKeyExpression() != null) {
+					channelProducerProperties.setProperty(BinderProperties.PARTITION_KEY_EXPRESSION,
+							bindingProperties.getPartitionKeyExpression());
+				}
+				if (bindingProperties.getPartitionKeyExtractorClass() != null) {
+					channelProducerProperties.setProperty(BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS,
+							bindingProperties.getPartitionKeyExtractorClass());
+				}
+				if (bindingProperties.getPartitionSelectorClass() != null) {
+					channelProducerProperties.setProperty(BinderProperties.PARTITION_SELECTOR_CLASS,
+							bindingProperties.getPartitionSelectorClass());
+				}
+				if (bindingProperties.getPartitionSelectorExpression() != null) {
+					channelProducerProperties.setProperty(BinderProperties.PARTITION_SELECTOR_EXPRESSION,
+							bindingProperties.getPartitionSelectorExpression());
+				}
+			}
 			return channelProducerProperties;
 		}
 		else {
@@ -210,14 +173,6 @@ public class ChannelBindingProperties {
 		}
 	}
 
-	@SuppressWarnings("unchecked")
-	private void copyChannelBindingProperty(String outputChannelName, Properties targetProperties, String propertyName) {
-		Map<String, Object> channelBindingProperties = (Map<String, Object>) bindings.get(outputChannelName);
-		if (null != channelBindingProperties && channelBindingProperties.containsKey(propertyName)) {
-			targetProperties.setProperty(propertyName, (String) channelBindingProperties.get(propertyName));
-		}
-	}
-
 	public String getTapChannelName(String channelName) {
 		return "tap:" + getBindingDestination(channelName);
 	}