diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index c4bb09a55..5a883be4f 100644 --- a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -449,7 +449,6 @@ public abstract class AbstractBinderTests partitionKeyExtractorClass; + /** + * The name of the bean that implements {@link PartitionKeyExtractorStrategy}\. + * Used to extract a key used to compute the partition id (see 'partitionSelector*') + *
+ * Mutually exclusive with 'partitionKeyExpression'. + */ + private String partitionKeyExtractorName; + + /** + * @deprecated in favor of 'partitionSelectorName' + */ + @Deprecated private Class partitionSelectorClass; + /** + * The name of the bean that implements {@link PartitionSelectorStrategy}\. + * Used to determine partition id based on partition key (see 'partitionKeyExtractor*'). + *
+ * Mutually exclusive with 'partitionSelectorExpression'. + */ + private String partitionSelectorName; + @JsonSerialize(using = ExpressionSerializer.class) private Expression partitionSelectorExpression; @@ -63,22 +88,27 @@ public class ProducerProperties { this.partitionKeyExpression = partitionKeyExpression; } + @Deprecated public Class getPartitionKeyExtractorClass() { return partitionKeyExtractorClass; } + @Deprecated public void setPartitionKeyExtractorClass(Class partitionKeyExtractorClass) { this.partitionKeyExtractorClass = partitionKeyExtractorClass; } public boolean isPartitioned() { - return this.partitionKeyExpression != null || partitionKeyExtractorClass != null; + return this.partitionCount > 1 || this.partitionKeyExpression != null + || this.partitionKeyExtractorName != null || this.partitionKeyExtractorClass != null; } + @Deprecated public Class getPartitionSelectorClass() { return partitionSelectorClass; } + @Deprecated public void setPartitionSelectorClass(Class partitionSelectorClass) { this.partitionSelectorClass = partitionSelectorClass; } @@ -142,4 +172,20 @@ public class ProducerProperties { this.errorChannelEnabled = errorChannelEnabled; } + public String getPartitionKeyExtractorName() { + return partitionKeyExtractorName; + } + + public void setPartitionKeyExtractorName(String partitionKeyExtractorName) { + this.partitionKeyExtractorName = partitionKeyExtractorName; + } + + public String getPartitionSelectorName() { + return partitionSelectorName; + } + + public void setPartitionSelectorName(String partitionSelectorName) { + this.partitionSelectorName = partitionSelectorName; + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index d894f0109..3ff06ebda 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -17,11 +17,15 @@ package org.springframework.cloud.stream.binding; import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; @@ -53,7 +57,7 @@ import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; +import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; @@ -73,8 +77,9 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Oleg Zhurakousky */ -public class MessageConverterConfigurer - implements MessageChannelAndSourceConfigurer, BeanFactoryAware, InitializingBean { +public class MessageConverterConfigurer implements MessageChannelAndSourceConfigurer, BeanFactoryAware { + + private final Log logger = LogFactory.getLog(getClass()); private final MessageBuilderFactory messageBuilderFactory = new MutableMessageBuilderFactory(); @@ -84,12 +89,25 @@ public class MessageConverterConfigurer private ConfigurableListableBeanFactory beanFactory; + private final Map partitionKeyExtractors; + + private final Map partitionSelectors; + public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties, CompositeMessageConverterFactory compositeMessageConverterFactory) { + this(bindingServiceProperties, compositeMessageConverterFactory, Collections.emptyMap(), Collections.emptyMap()); + } + + public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties, + CompositeMessageConverterFactory compositeMessageConverterFactory, + Map partitionKeyExtractors, + Map 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; } @Override @@ -97,11 +115,6 @@ public class MessageConverterConfigurer this.beanFactory = (ConfigurableListableBeanFactory) beanFactory; } - @Override - public void afterPropertiesSet() throws Exception { - Assert.notNull(this.beanFactory, "Bean factory cannot be empty"); - } - @Override public void configureInputChannel(MessageChannel messageChannel, String channelName) { configureMessageChannel(messageChannel, channelName, true); @@ -165,50 +178,67 @@ public class MessageConverterConfigurer } } - private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy( - ProducerProperties producerProperties) { + @SuppressWarnings("deprecation") + private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy(ProducerProperties producerProperties) { + PartitionKeyExtractorStrategy partitionKeyExtractor; if (producerProperties.getPartitionKeyExtractorClass() != null) { - return getBean(producerProperties.getPartitionKeyExtractorClass().getName(), - PartitionKeyExtractorStrategy.class); + logger.warn("'partitionKeyExtractorClass' option is deprecated as of v2.0. Please configure partition " + + "key extractor as a @Bean that implements 'PartitionKeyExtractorStrategy'. Additionally you can " + + "specify 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' to specify which " + + "bean to use in the event there are more then one."); + partitionKeyExtractor = instantiate(producerProperties.getPartitionKeyExtractorClass(), PartitionKeyExtractorStrategy.class); } - return null; + else if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) { + partitionKeyExtractor = this.partitionKeyExtractors.get(producerProperties.getPartitionKeyExtractorName()); + 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 " + + "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(); + } + return partitionKeyExtractor; } - private PartitionSelectorStrategy getPartitionSelectorStrategy( - ProducerProperties producerProperties) { + @SuppressWarnings("deprecation") + private PartitionSelectorStrategy getPartitionSelectorStrategy(ProducerProperties producerProperties) { + PartitionSelectorStrategy partitionSelector; if (producerProperties.getPartitionSelectorClass() != null) { - return getBean(producerProperties.getPartitionSelectorClass().getName(), + logger.warn("'partitionSelectorClass' option is deprecated as of v2.0. Please configure partition " + + "selector as a @Bean that implements 'PartitionSelectorStrategy'. Additionally you can " + + "specify 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' to specify which " + + "bean to use in the event there are more then one."); + partitionSelector = instantiate(producerProperties.getPartitionSelectorClass(), PartitionSelectorStrategy.class); } - return new DefaultPartitionSelector(); + else if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) { + partitionSelector = this.partitionSelectors.get(producerProperties.getPartitionSelectorName()); + 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 " + + "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(); + } + return partitionSelector; } @SuppressWarnings("unchecked") - private T getBean(String className, Class type) { - if (this.beanFactory.containsBean(className)) { - return this.beanFactory.getBean(className, type); + private T instantiate(Class implClass, Class type) { + try { + return (T) implClass.newInstance(); } - else { - synchronized (this) { - T bean; - Class clazz; - try { - clazz = ClassUtils.forName(className, this.beanFactory.getBeanClassLoader()); - } - catch (Exception e) { - throw new BinderException("Failed to load class: " + className, e); - } - try { - bean = (T) clazz.newInstance(); - Assert.isInstanceOf(type, bean); - this.beanFactory.registerSingleton(className, bean); - this.beanFactory.initializeBean(bean, className); - } - catch (Exception e) { - throw new BinderException("Failed to instantiate class: " + className, e); - } - return bean; - } + catch (Exception e) { + throw new BinderException("Failed to instantiate class: " + implClass.getName(), e); } } @@ -253,22 +283,22 @@ public class MessageConverterConfigurer if (message instanceof ErrorMessage) { return message; } - + Message postProcessedMessage = message; MimeType contentType = this.mimeType; if (message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { Object ct = message.getHeaders().get(MessageHeaders.CONTENT_TYPE); contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (MimeType)ct; } - - boolean deserializationRequired = message.getPayload() instanceof byte[] && + + boolean deserializationRequired = message.getPayload() instanceof byte[] && ("text".equalsIgnoreCase(contentType.getType()) || equalTypeAndSubType(MimeTypeUtils.APPLICATION_JSON, contentType) || equalTypeAndSubType(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, contentType) || equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType)); - + Object payload = deserializationRequired ? this.deserializePayload(message, contentType) : message.getPayload(); - + if (payload != null) { Object ct = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE); contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (ct == null ? contentType : (MimeType)ct); @@ -312,7 +342,7 @@ public class MessageConverterConfigurer + message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE), e); } } - + Assert.isTrue(!(equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentTypeToUse) && targetClass == null), "Cannot deserialize into message since 'contentType` is not " + "encoded with the actual target type." diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index dfd0db0d5..805e915a9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -28,6 +28,8 @@ 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; @@ -123,8 +125,10 @@ public class BindingServiceConfiguration { @Bean public MessageConverterConfigurer messageConverterConfigurer(BindingServiceProperties bindingServiceProperties, - CompositeMessageConverterFactory compositeMessageConverterFactory) { - return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory); + CompositeMessageConverterFactory compositeMessageConverterFactory, + @Nullable Map partitionKeyExtractors, + @Nullable Map partitionSelectors) { + return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory, partitionKeyExtractors, partitionSelectors); } @Bean diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java index 16d075c80..59b21a345 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-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. @@ -20,11 +20,9 @@ import java.lang.reflect.Field; import org.junit.Assert; import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.PartitionHandler; import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; @@ -33,6 +31,7 @@ import org.springframework.cloud.stream.messaging.Source; import org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass; import org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; +import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.PropertySource; @@ -43,22 +42,23 @@ import org.springframework.integration.core.MessageSource; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ChannelInterceptor; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ReflectionUtils; /** * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ -@RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = CustomPartitionedProducerTest.TestSource.class) public class CustomPartitionedProducerTest { - @Autowired - private Source testSource; - @Test public void testCustomPartitionedProducer() { - DirectChannel messageChannel = (DirectChannel) this.testSource.output(); + ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, + "--spring.jmx.enabled=false", + "--spring.main.web-application-type=none", + "--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass", + "--spring.cloud.stream.bindings.output.producer.partitionSelectorClass=org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass"); + Source testSource = context.getBean(Source.class); + DirectChannel messageChannel = (DirectChannel) testSource.output(); for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) { if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { Field partitionHandlerField = ReflectionUtils @@ -80,6 +80,103 @@ public class CustomPartitionedProducerTest { } } } + + @Test + public void testCustomPartitionedProducerByName() { + ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, + "--spring.jmx.enabled=false", + "--spring.main.web-application-type=none", + "--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractor", + "--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelector"); + Source testSource = context.getBean(Source.class); + DirectChannel messageChannel = (DirectChannel) testSource.output(); + for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils + .findField(MessageConverterConfigurer.PartitioningInterceptor.class, "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils.getField(partitionHandlerField, + channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField(PartitionHandler.class, + "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils.findField(PartitionHandler.class, + "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + Assert.assertTrue(((PartitionKeyExtractorStrategy) ReflectionUtils.getField(partitonKeyExtractorField, + partitionHandler)).getClass().equals(CustomPartitionKeyExtractorClass.class)); + Assert.assertTrue( + ((PartitionSelectorStrategy) ReflectionUtils.getField(partitonSelectorField, partitionHandler)) + .getClass().equals(CustomPartitionSelectorClass.class)); + } + } + } + + @Test + public void testCustomPartitionedProducerAsSingletons() { + ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, + "--spring.jmx.enabled=false", "--spring.main.web-application-type=none"); + Source testSource = context.getBean(Source.class); + DirectChannel messageChannel = (DirectChannel) testSource.output(); + for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils + .findField(MessageConverterConfigurer.PartitioningInterceptor.class, "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils.getField(partitionHandlerField, + channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField(PartitionHandler.class, + "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils.findField(PartitionHandler.class, + "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + Assert.assertTrue(((PartitionKeyExtractorStrategy) ReflectionUtils.getField(partitonKeyExtractorField, + partitionHandler)).getClass().equals(CustomPartitionKeyExtractorClass.class)); + Assert.assertTrue( + ((PartitionSelectorStrategy) ReflectionUtils.getField(partitonSelectorField, partitionHandler)) + .getClass().equals(CustomPartitionSelectorClass.class)); + } + } + } + + public void testCustomPartitionedProducerMultipleInstances() { + ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSourceMultipleStrategies.class, + "--spring.jmx.enabled=false", + "--spring.main.web-application-type=none", + "--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractorOne", + "--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelectorTwo"); + Source testSource = context.getBean(Source.class); + DirectChannel messageChannel = (DirectChannel) testSource.output(); + for (ChannelInterceptor channelInterceptor : messageChannel.getChannelInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils + .findField(MessageConverterConfigurer.PartitioningInterceptor.class, "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils.getField(partitionHandlerField, + channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField(PartitionHandler.class, + "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils.findField(PartitionHandler.class, + "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + Assert.assertTrue(((PartitionKeyExtractorStrategy) ReflectionUtils.getField(partitonKeyExtractorField, + partitionHandler)).getClass().equals(CustomPartitionKeyExtractorClass.class)); + Assert.assertTrue( + ((PartitionSelectorStrategy) ReflectionUtils.getField(partitonSelectorField, partitionHandler)) + .getClass().equals(CustomPartitionSelectorClass.class)); + } + } + } + + @Test(expected=Exception.class) + // It actually throws UnsatisfiedDependencyException, but it is confusing when it comes to test + // But for the purposes of the test all we care about is that it fails + public void testCustomPartitionedProducerMultipleInstancesFailNoFilter() { + SpringApplication.run(CustomPartitionedProducerTest.TestSourceMultipleStrategies.class, + "--spring.jmx.enabled=false", "--spring.main.web-application-type=none"); + } @EnableBinding(Source.class) @EnableAutoConfiguration @@ -87,6 +184,54 @@ public class CustomPartitionedProducerTest { @PropertySource("classpath:/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties") public static class TestSource { + @Bean + public CustomPartitionSelectorClass customPartitionSelector() { + return new CustomPartitionSelectorClass(); + } + + @Bean + public CustomPartitionKeyExtractorClass customPartitionKeyExtractor() { + return new CustomPartitionKeyExtractorClass(); + } + + @Bean + @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) + public MessageSource timerMessageSource() { + return new MessageSource() { + @Override + public Message receive() { + throw new MessagingException("test"); + } + }; + } + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @Import(MockBinderRegistryConfiguration.class) + @PropertySource("classpath:/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties") + public static class TestSourceMultipleStrategies { + + @Bean + public CustomPartitionSelectorClass customPartitionSelectorOne() { + return new CustomPartitionSelectorClass(); + } + + @Bean + public CustomPartitionSelectorClass customPartitionSelectorTwo() { + return new CustomPartitionSelectorClass(); + } + + @Bean + public CustomPartitionKeyExtractorClass customPartitionKeyExtractorOne() { + return new CustomPartitionKeyExtractorClass(); + } + + @Bean + public CustomPartitionKeyExtractorClass customPartitionKeyExtractorTwo() { + return new CustomPartitionKeyExtractorClass(); + } + @Bean @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) public MessageSource timerMessageSource() { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java index af0d1a6cf..058a7caf6 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-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. @@ -24,7 +24,7 @@ import org.junit.Test; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.boot.actuate.context.properties.ConfigurationPropertiesReportEndpoint; -import org.springframework.boot.actuate.context.properties.ConfigurationPropertiesReportEndpoint.ConfigurationPropertiesDescriptor; +import org.springframework.boot.actuate.context.properties.ConfigurationPropertiesReportEndpoint.ContextConfigurationProperties; import org.springframework.context.support.StaticApplicationContext; import static org.junit.Assert.assertFalse; @@ -61,7 +61,9 @@ public class BinderPropertiesTests { ConfigurationPropertiesReportEndpoint endpoint = new ConfigurationPropertiesReportEndpoint(); endpoint.setApplicationContext(context); - ConfigurationPropertiesDescriptor configurationProperties = endpoint.configurationProperties(); + + ContextConfigurationProperties configurationProperties = endpoint.configurationProperties().getContexts().values().iterator().next(); + Map properties = configurationProperties.getBeans().get("bindingServiceProperties").getProperties(); assertFalse(properties.containsKey("error")); assertTrue(properties.containsKey("binders")); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java deleted file mode 100644 index 4e332d99f..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * 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. - * 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.partitioning; - -import org.junit.Assert; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.ArgumentCaptor; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.BinderFactory; -import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; -import org.springframework.context.annotation.Import; -import org.springframework.context.annotation.PropertySource; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.hamcrest.Matchers.equalTo; -import static org.mockito.Matchers.eq; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - */ -@RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = PartitionedProducerTest.TestSource.class) -public class PartitionedProducerTest { - - @Autowired - private BinderFactory binderFactory; - - @Autowired - private Source testSource; - - @Test - @SuppressWarnings("rawtypes") - public void testBindingPartitionedProducer() { - Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(ProducerProperties.class); - verify(binder).bindProducer(eq("partOut"), eq(this.testSource.output()), argumentCaptor.capture()); - Assert.assertThat(argumentCaptor.getValue().getPartitionCount(), equalTo(3)); - Assert.assertThat(argumentCaptor.getValue().getPartitionKeyExpression().getExpressionString(), - equalTo("payload")); - verifyNoMoreInteractions(binder); - } - - @EnableBinding(Source.class) - @EnableAutoConfiguration - @Import(MockBinderRegistryConfiguration.class) - @PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-producer-test.properties") - public static class TestSource { - - } -} diff --git a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties index 99efd2692..a96bf5076 100644 --- a/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties +++ b/spring-cloud-stream/src/test/resources/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties @@ -1,4 +1,2 @@ spring.cloud.stream.bindings.output.destination=partOut spring.cloud.stream.bindings.output.producer.partitionCount=3 -spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass -spring.cloud.stream.bindings.output.producer.partitionSelectorClass=org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass