diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java index 34010c2cd..a369fcef6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2020 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. @@ -16,9 +16,18 @@ package org.springframework.cloud.stream.binder; +import java.lang.reflect.Field; +import java.util.Map; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.context.expression.BeanFactoryResolver; import org.springframework.expression.EvaluationContext; import org.springframework.messaging.Message; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; /** * Utility class to determine if a binding is configured for partitioning (based on the @@ -31,6 +40,7 @@ import org.springframework.util.Assert; * @author Ilayaperumal Gopinathan * @author Mark Fisher * @author Marius Bogoevici + * @author Oleg Zhurakousky */ public class PartitionHandler { @@ -42,6 +52,8 @@ public class PartitionHandler { private final PartitionSelectorStrategy partitionSelectorStrategy; + private final ConfigurableListableBeanFactory beanFactory; + private volatile int partitionCount; /** @@ -50,16 +62,35 @@ public class PartitionHandler { * @param properties binder properties * @param partitionKeyExtractorStrategy PartitionKeyExtractor strategy * @param partitionSelectorStrategy PartitionSelector strategy + * + * @deprecated since 3.0.2. Please use another constructor which allows you to pass an instance of beanFactory */ + @Deprecated public PartitionHandler(EvaluationContext evaluationContext, ProducerProperties properties, PartitionKeyExtractorStrategy partitionKeyExtractorStrategy, PartitionSelectorStrategy partitionSelectorStrategy) { + + this(evaluationContext, properties, (ConfigurableListableBeanFactory) extractBeanFactoryFromEvaluationContext(evaluationContext)); + } + + /** + * Construct a {@code PartitionHandler}. + * @param evaluationContext evaluation context for binder + * @param properties binder properties + * @param beanFactory instance of ConfigurableListableBeanFactory + * + * @since 3.0.2 + */ + public PartitionHandler(EvaluationContext evaluationContext, + ProducerProperties properties, ConfigurableListableBeanFactory beanFactory) { + this.beanFactory = beanFactory; this.evaluationContext = evaluationContext; this.producerProperties = properties; - this.partitionKeyExtractorStrategy = partitionKeyExtractorStrategy; - this.partitionSelectorStrategy = partitionSelectorStrategy; + this.partitionKeyExtractorStrategy = this.getPartitionKeyExtractorStrategy(properties); + this.partitionSelectorStrategy = this.getPartitionSelectorStrategy(properties); this.partitionCount = this.producerProperties.getPartitionCount(); + } /** @@ -120,4 +151,87 @@ public class PartitionHandler { return null; } + private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy( + ProducerProperties producerProperties) { + PartitionKeyExtractorStrategy partitionKeyExtractor; + if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) { + partitionKeyExtractor = (PartitionKeyExtractorStrategy) 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 { + Map 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(extractors) ? null + : extractors.values().iterator().next(); + } + return partitionKeyExtractor; + } + + private PartitionSelectorStrategy getPartitionSelectorStrategy( + ProducerProperties producerProperties) { + PartitionSelectorStrategy partitionSelector; + if (StringUtils.hasText(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 { + Map 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(selectors) + ? new DefaultPartitionSelector() + : selectors.values().iterator().next(); + } + return partitionSelector; + } + + private static BeanFactory extractBeanFactoryFromEvaluationContext(EvaluationContext evaluationContext) { + try { + Field field = ReflectionUtils.findField(BeanFactoryResolver.class, "beanFactory"); + field.setAccessible(true); + return (BeanFactory) field.get(evaluationContext); + } + catch (Exception e) { + throw new RuntimeException("Failed to extract beanFactory from EvaluationContext. Please use different constructor" + + " which allows you to pass the instance of the beanFactory."); + } + } + + /** + * Default partition strategy; only works on keys with "real" hash codes, such as + * String. Caller now always applies modulo so no need to do so here. + */ + private static class DefaultPartitionSelector implements PartitionSelectorStrategy { + + @Override + public int selectPartition(Object key, int partitionCount) { + int hashCode = key.hashCode(); + if (hashCode == Integer.MIN_VALUE) { + hashCode = 0; + } + return Math.abs(hashCode); + } + } + + + } 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 e84acaea9..6ecd4f5c5 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -28,8 +28,6 @@ import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; import org.springframework.cloud.stream.binder.JavaClassMimeTypeUtils; import org.springframework.cloud.stream.binder.PartitionHandler; -import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; -import org.springframework.cloud.stream.binder.PartitionSelectorStrategy; import org.springframework.cloud.stream.binder.PollableMessageSource; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.BindingProperties; @@ -49,11 +47,11 @@ import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; +//import org.springframework.util.StringUtils; /** * A {@link MessageChannelConfigurer} that sets data types and message converters based on @@ -147,23 +145,21 @@ public class MessageConverterConfigurer boolean partitioned = !inbound && producerProperties != null && producerProperties.isPartitioned(); boolean functional = streamFunctionProperties != null && StringUtils.hasText(streamFunctionProperties.getDefinition()); if (partitioned) { - messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties, - getPartitionKeyExtractorStrategy(producerProperties), - getPartitionSelectorStrategy(producerProperties))); + if (inbound || !functional) { + messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties)); + } } ConsumerProperties consumerProperties = bindingProperties.getConsumer(); if (this.isNativeEncodingNotSet(producerProperties, consumerProperties, inbound)) { - if (partitioned || !functional) { - if (inbound) { - messageChannel.addInterceptor( - new InboundContentTypeEnhancingInterceptor(contentType)); - } - else { - messageChannel.addInterceptor( - new OutboundContentTypeConvertingInterceptor(contentType, - this.compositeMessageConverter)); - } + if (inbound) { + messageChannel.addInterceptor( + new InboundContentTypeEnhancingInterceptor(contentType)); + } + else { + messageChannel.addInterceptor( + new OutboundContentTypeConvertingInterceptor(contentType, + this.compositeMessageConverter)); } } } @@ -180,76 +176,6 @@ public class MessageConverterConfigurer } } - private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy( - ProducerProperties producerProperties) { - PartitionKeyExtractorStrategy partitionKeyExtractor; - if (StringUtils.hasText(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 { - Map 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(extractors) ? null - : extractors.values().iterator().next(); - } - return partitionKeyExtractor; - } - - private PartitionSelectorStrategy getPartitionSelectorStrategy( - ProducerProperties producerProperties) { - PartitionSelectorStrategy partitionSelector; - if (StringUtils.hasText(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 { - Map 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(selectors) - ? new DefaultPartitionSelector() - : selectors.values().iterator().next(); - } - return partitionSelector; - } - - /** - * Default partition strategy; only works on keys with "real" hash codes, such as - * String. Caller now always applies modulo so no need to do so here. - */ - private static class DefaultPartitionSelector implements PartitionSelectorStrategy { - - @Override - public int selectPartition(Object key, int partitionCount) { - int hashCode = key.hashCode(); - if (hashCode == Integer.MIN_VALUE) { - hashCode = 0; - } - return Math.abs(hashCode); - } - - } - /** * Primary purpose of this interceptor is to enhance/enrich Message that sent to the * *inbound* channel with 'contentType' header for cases where 'contentType' is not @@ -381,15 +307,12 @@ public class MessageConverterConfigurer private final PartitionHandler partitionHandler; - PartitioningInterceptor(BindingProperties bindingProperties, - PartitionKeyExtractorStrategy partitionKeyExtractorStrategy, - PartitionSelectorStrategy partitionSelectorStrategy) { + PartitioningInterceptor(BindingProperties bindingProperties) { this.bindingProperties = bindingProperties; this.partitionHandler = new PartitionHandler( ExpressionUtils.createStandardEvaluationContext( MessageConverterConfigurer.this.beanFactory), - this.bindingProperties.getProducer(), partitionKeyExtractorStrategy, - partitionSelectorStrategy); + this.bindingProperties.getProducer(), MessageConverterConfigurer.this.beanFactory); } public void setPartitionCount(int partitionCount) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java index d92500bb8..1d37024da 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ContentTypeConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.config; -import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; @@ -29,20 +28,21 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Role; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter; import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter; +import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.converter.MessageConverter; /** * @author Vinicius Carvalho * @author Artem Bilan + * @author Oleg Zhurakousky */ @Configuration @Role(BeanDefinition.ROLE_INFRASTRUCTURE) class ContentTypeConfiguration { @Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) - public ConfigurableCompositeMessageConverter configurableCompositeMessageConverter( + public CompositeMessageConverter configurableCompositeMessageConverter( ObjectProvider objectMapperObjectProvider, List customMessageConverters) { @@ -50,12 +50,9 @@ class ContentTypeConfiguration { .filter(c -> !(c instanceof DefaultDatatypeChannelMessageConverter)).collect(Collectors.toList()); CompositeMessageConverterFactory factory = - new CompositeMessageConverterFactory(new ArrayList<>(), objectMapperObjectProvider.getIfAvailable(ObjectMapper::new)); + new CompositeMessageConverterFactory(customMessageConverters, objectMapperObjectProvider.getIfAvailable(ObjectMapper::new)); - ArrayList messageConverters = new ArrayList<>(customMessageConverters); - messageConverters.addAll(factory.getMessageConverterForAllRegistered().getConverters()); - - return new ConfigurableCompositeMessageConverter(messageConverters); + return factory.getMessageConverterForAllRegistered(); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index c5d137622..d154ddf60 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -58,9 +58,11 @@ import org.springframework.cloud.function.context.config.ContextFunctionCatalogA import org.springframework.cloud.function.context.config.FunctionContextUtils; import org.springframework.cloud.function.context.config.RoutingFunction; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.PartitionHandler; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.BindableProxyFactory; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; @@ -81,11 +83,13 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.env.Environment; import org.springframework.core.type.MethodMetadata; +import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.support.MessageBuilder; import org.springframework.lang.Nullable; @@ -127,10 +131,6 @@ public class FunctionConfiguration { BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, FunctionBindingRegistrar bindingHolder, BinderAwareChannelResolver dynamicDestinationResolver) { -// boolean shouldCreateInitializer = bindableProxyFactories != null -// && (applicationContext.containsBean("output") // need this to compose to existing legacy message source -// || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class))); - boolean shouldCreateInitializer = applicationContext.containsBean("output") || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)); @@ -417,7 +417,8 @@ public class FunctionConfiguration { ProducerProperties producerProperties = StringUtils.hasText(outputChannelName) ? this.serviceProperties.getBindingProperties(outputChannelName).getProducer() : null; - ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties, producerProperties)) { + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties, + producerProperties, applicationContext)) { @Override protected void sendOutputs(Object result, Message requestMessage) { if (result instanceof Message && ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { @@ -512,7 +513,10 @@ public class FunctionConfiguration { private final Field headersField; - FunctionWrapper(Function function, ConsumerProperties consumerProperties, ProducerProperties producerProperties) { + private final ConfigurableApplicationContext applicationContext; + + FunctionWrapper(Function function, ConsumerProperties consumerProperties, + ProducerProperties producerProperties, ConfigurableApplicationContext applicationContext) { this.function = function; Type type = ((FunctionInvocationWrapper) function).getFunctionType(); if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(type, 0))) { @@ -522,7 +526,9 @@ public class FunctionConfiguration { this.producerProperties = producerProperties; this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers"); this.headersField.setAccessible(true); + this.applicationContext = applicationContext; } + @SuppressWarnings("unchecked") @Override public Object apply(Message message) { @@ -531,7 +537,21 @@ public class FunctionConfiguration { .getField(this.headersField, message.getHeaders()); headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding()); } - Object result = function.apply(message); + + Function outputMessageEnricher = null; + if (producerProperties != null && producerProperties.isPartitioned()) { + StandardEvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.applicationContext.getBeanFactory()); + PartitionHandler partitionHandler = new PartitionHandler(evaluationContext, producerProperties, this.applicationContext.getBeanFactory()); + + outputMessageEnricher = outputMessage -> { + int partitionId = partitionHandler.determinePartition(outputMessage); + return MessageBuilder + .fromMessage(outputMessage) + .setHeader(BinderHeaders.PARTITION_HEADER, partitionId).build(); + }; + } + + Object result = ((FunctionInvocationWrapper) function).apply(message, outputMessageEnricher); if (result instanceof Publisher && ((FunctionInvocationWrapper) this.function).getTarget() instanceof RoutingFunction) { throw new IllegalStateException("Routing to functions that return Publisher " + "is not supported in the context of Spring Cloud Stream."); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index 26645855a..decd81623 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -552,6 +552,29 @@ public class ImplicitFunctionBindingTests { } } + @Test + public void partitionOnOutputPayloadTest() { + System.clearProperty("spring.cloud.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration + .getCompleteConfiguration(PojoFunctionConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.cloud.function.definition=func", + "--spring.cloud.stream.bindings.func-out-0.producer.partitionKeyExpression=payload.id", + "--spring.cloud.stream.bindings.func-out-0.producer.partitionCount=5", + "--spring.jmx.enabled=false")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("Jim Lahey".getBytes()).build(); + + inputDestination.send(inputMessage, "func-in-0"); + + assertThat(outputDestination.receive(100, "func-out-0").getHeaders().get("scst_partition")).isEqualTo(3); + + assertThat(outputDestination.receive(100)).isNull(); + } + } + @EnableAutoConfiguration public static class NoEnableBindingConfiguration { @@ -737,4 +760,39 @@ public class ImplicitFunctionBindingTests { } } + @EnableAutoConfiguration + public static class PojoFunctionConfiguration { + + @Bean + public Function func() { + return x -> { + Person person = new Person(); + person.setName(x); + person.setId(3); + return person; + }; + } + } + + public static class Person { + private String name; + private int id; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + } + }