From 10c1f5051cd427ddd76a6968b0aa81eecf26e859 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Thu, 24 Sep 2015 17:53:11 -0400 Subject: [PATCH] Typesafe BindingProperties - Create BindingProperties and Converter to replace the current Map-based logic --- .../stream/binding/ChannelBindingService.java | 20 +-- .../stream/config/BindingProperties.java | 95 +++++++++++++++ .../config/BindingPropertiesConverter.java | 39 ++++++ .../ChannelBindingServiceConfiguration.java | 14 ++- ...a => ChannelBindingServiceProperties.java} | 115 ++++++------------ 5 files changed, 190 insertions(+), 93 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingPropertiesConverter.java rename spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/{ChannelBindingProperties.java => ChannelBindingServiceProperties.java} (57%) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index c3b83e0a5..9f8a6bcef 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/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 binder; - private ChannelBindingProperties channelBindingProperties; + private ChannelBindingServiceProperties channelBindingServiceProperties; - public ChannelBindingService(ChannelBindingProperties channelBindingProperties, Binder binder) { - this.channelBindingProperties = channelBindingProperties; + public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, Binder 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)); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java new file mode 100644 index 000000000..0315d9ded --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingPropertiesConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingPropertiesConverter.java new file mode 100644 index 000000000..83a056fc1 --- /dev/null +++ b/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 { + + 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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index 22bfa7519..08b200722 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/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 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 bindingPropertiesConverter() { + return new BindingPropertiesConverter(); + } + // IMPORTANT: Nested class to avoid instantiating all of the above early @Configuration protected static class PostProcessorConfiguration { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java similarity index 57% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java index a0453982a..4d2c8d10f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java +++ b/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 bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + private Map 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 getBindings() { + public Map getBindings() { return bindings; } - public void setBindings(Map bindings) { + public void setBindings(Map 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 channelBindingProperties = (Map) 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); }