SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1330 Fixed inconsistency

Browse Source 
Resolves #1330
This commit is contained in:
Oleg Zhurakousky
2018-03-30 14:40:35 -04:00
parent 4bb54418df
commit f411442887
3 changed files with 18 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java
View File

@@ -92,13 +92,9 @@ public class PartitionHandler {
return Math.abs(partition % producerProperties.getPartitionCount());
}
@SuppressWarnings("deprecation")
private Object extractKey(Message<?> message) {
Object key = null;
if (this.producerProperties.getPartitionKeyExtractorClass() != null) {
key = invokeKeyExtractor(message);
}
else if (this.producerProperties.getPartitionKeyExpression() != null) {
Object key = invokeKeyExtractor(message);
if (key == null && this.producerProperties.getPartitionKeyExpression() != null) {
key = this.producerProperties.getPartitionKeyExpression().getValue(this.evaluationContext, message);
}
Assert.notNull(key, "Partition key cannot be null");
@@ -107,7 +103,10 @@ public class PartitionHandler {
}
private Object invokeKeyExtractor(Message<?> message) {
return this.partitionKeyExtractorStrategy.extractKey(message);
if (partitionKeyExtractorStrategy != null) {
return this.partitionKeyExtractorStrategy.extractKey(message);
}
return null;
}
}

34
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -17,7 +17,6 @@
package org.springframework.cloud.stream.binding;
import java.lang.reflect.Field;
import java.util.Collections;
import java.util.Map;
import org.apache.commons.logging.Log;
@@ -86,27 +85,14 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
private ConfigurableListableBeanFactory beanFactory;
private final Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors;
private final Map<String, PartitionSelectorStrategy> partitionSelectors;
private final Field headersField;
public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
this(bindingServiceProperties, compositeMessageConverterFactory, Collections.emptyMap(), Collections.emptyMap());
}
public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverterFactory compositeMessageConverterFactory,
Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors,
Map<String, PartitionSelectorStrategy> partitionSelectors) {
Assert.notNull(compositeMessageConverterFactory,
"The message converter factory cannot be null");
this.bindingServiceProperties = bindingServiceProperties;
this.compositeMessageConverterFactory = compositeMessageConverterFactory;
this.partitionKeyExtractors = partitionKeyExtractors == null ? Collections.emptyMap() : partitionKeyExtractors;
this.partitionSelectors = partitionSelectors == null ? Collections.emptyMap() : partitionSelectors;
this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
headersField.setAccessible(true);
@@ -190,17 +176,17 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
partitionKeyExtractor = instantiate(producerProperties.getPartitionKeyExtractorClass(), PartitionKeyExtractorStrategy.class);
}
else if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) {
partitionKeyExtractor = this.partitionKeyExtractors.get(producerProperties.getPartitionKeyExtractorName());
partitionKeyExtractor = this.beanFactory.getBean(producerProperties.getPartitionKeyExtractorName(), PartitionKeyExtractorStrategy.class);
Assert.notNull(partitionKeyExtractor, "PartitionKeyExtractorStrategy bean with the name '" + producerProperties.getPartitionKeyExtractorName()
+ "' can not be found. Has it been configured (e.g., @Bean)?");
}
else {
Assert.isTrue(this.partitionKeyExtractors.size() <= 1,
"Multiple beans of type 'PartitionKeyExtractorStrategy' found. " + this.partitionKeyExtractors + ". Please "
Map<String, PartitionKeyExtractorStrategy> extractors = this.beanFactory.getBeansOfType(PartitionKeyExtractorStrategy.class);
Assert.isTrue(extractors.size() <= 1,
"Multiple beans of type 'PartitionKeyExtractorStrategy' found. " + extractors + ". Please "
+ "use 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' property to specify "
+ "the name of the bean to be used.");
partitionKeyExtractor = CollectionUtils.isEmpty(this.partitionKeyExtractors) ?
null : this.partitionKeyExtractors.values().iterator().next();
partitionKeyExtractor = CollectionUtils.isEmpty(extractors) ? null : extractors.values().iterator().next();
}
return partitionKeyExtractor;
}
@@ -217,18 +203,18 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
PartitionSelectorStrategy.class);
}
else if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) {
partitionSelector = this.partitionSelectors.get(producerProperties.getPartitionSelectorName());
partitionSelector = this.beanFactory.getBean(producerProperties.getPartitionSelectorName(), PartitionSelectorStrategy.class);
Assert.notNull(partitionSelector,
"PartitionSelectorStrategy bean with the name '" + producerProperties.getPartitionSelectorName()
+ "' can not be found. Has it been configured (e.g., @Bean)?");
}
else {
Assert.isTrue(this.partitionSelectors.size() <= 1,
"Multiple beans of type 'PartitionSelectorStrategy' found. " + this.partitionSelectors + ". Please "
Map<String, PartitionSelectorStrategy> selectors = this.beanFactory.getBeansOfType(PartitionSelectorStrategy.class);
Assert.isTrue(selectors.size() <= 1,
"Multiple beans of type 'PartitionSelectorStrategy' found. " + selectors + ". Please "
+ "use 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' property to specify "
+ "the name of the bean to be used.");
partitionSelector = CollectionUtils.isEmpty(this.partitionSelectors)
? new DefaultPartitionSelector() : this.partitionSelectors.values().iterator().next();
partitionSelector = CollectionUtils.isEmpty(selectors) ? new DefaultPartitionSelector() : selectors.values().iterator().next();
}
return partitionSelector;
}

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -27,8 +27,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
import org.springframework.cloud.stream.binding.Bindable;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
@@ -128,10 +126,8 @@ public class BindingServiceConfiguration {
@Bean
public MessageConverterConfigurer messageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverterFactory compositeMessageConverterFactory,
@Nullable Map<String, PartitionKeyExtractorStrategy> partitionKeyExtractors,
@Nullable Map<String, PartitionSelectorStrategy> partitionSelectors) {
return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory, partitionKeyExtractors, partitionSelectors);
CompositeMessageConverterFactory compositeMessageConverterFactory) {
return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory);
}
@Bean