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

GH-1855 Ensure propper application or PartitioningInterceptor

Browse Source 
Ensure propper application or PartitioningInterceptor regardless of the functional or annotation-based programming model used

Resolves #1855
This commit is contained in:
Oleg Zhurakousky
2019-12-02 13:34:12 +01:00
parent a2dc34a160
commit 74aee81028
3 changed files with 32 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,6 +35,7 @@ import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.MessageConverterUtils;
import org.springframework.cloud.stream.function.StreamFunctionProperties;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.MessageBuilderFactory;
@@ -79,10 +80,12 @@ public class MessageConverterConfigurer
private final Field headersField;
private final StreamFunctionProperties streamFunctionProperties;
private ConfigurableListableBeanFactory beanFactory;
public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverter compositeMessageConverter) {
CompositeMessageConverter compositeMessageConverter, StreamFunctionProperties streamFunctionProperties) {
Assert.notNull(compositeMessageConverter,
"The message converter factory cannot be null");
this.bindingServiceProperties = bindingServiceProperties;
@@ -90,6 +93,12 @@ public class MessageConverterConfigurer
this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
this.headersField.setAccessible(true);
this.streamFunctionProperties = streamFunctionProperties;
}
public MessageConverterConfigurer(BindingServiceProperties bindingServiceProperties,
CompositeMessageConverter compositeMessageConverter) {
this(bindingServiceProperties, compositeMessageConverter, null);
}
@Override
@@ -135,24 +144,26 @@ public class MessageConverterConfigurer
.getBindingProperties(channelName);
String contentType = bindingProperties.getContentType();
ProducerProperties producerProperties = bindingProperties.getProducer();
if (!inbound && producerProperties != null
&& producerProperties.isPartitioned()) {
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)));
}
ConsumerProperties consumerProperties = bindingProperties.getConsumer();
if (this.isNativeEncodingNotSet(producerProperties, consumerProperties,
inbound)) {
if (inbound) {
messageChannel.addInterceptor(
new InboundContentTypeEnhancingInterceptor(contentType));
}
else {
messageChannel.addInterceptor(
new OutboundContentTypeConvertingInterceptor(contentType,
this.compositeMessageConverter));
if (this.isNativeEncodingNotSet(producerProperties, consumerProperties, inbound)) {
if (partitioned || !functional) {
if (inbound) {
messageChannel.addInterceptor(
new InboundContentTypeEnhancingInterceptor(contentType));
}
else {
messageChannel.addInterceptor(
new OutboundContentTypeConvertingInterceptor(contentType,
this.compositeMessageConverter));
}
}
}
}

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
View File

@@ -44,12 +44,12 @@ import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory;
import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
import org.springframework.cloud.stream.function.StreamFunctionProperties;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Role;
import org.springframework.core.env.Environment;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.core.io.support.PropertiesLoaderUtils;
@@ -211,13 +211,9 @@ public class BinderFactoryAutoConfiguration {
public MessageConverterConfigurer messageConverterConfigurer(
BindingServiceProperties bindingServiceProperties,
@Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) CompositeMessageConverter compositeMessageConverter,
Environment environment, BinderTypeRegistry binderTypeRegistry) {
@Nullable StreamFunctionProperties streamFunctionProperties) {
if (binderTypeRegistry.getAll().keySet().contains("kstream") || !environment.containsProperty("spring.cloud.stream.function.definition")) {
return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverter);
}
return null;
return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverter, streamFunctionProperties);
}

10
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
View File

@@ -70,7 +70,7 @@ public class GreenfieldFunctionEnableBindingTests {
TestChannelBinderConfiguration
.getCompleteConfiguration(SourceFromSupplier.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=date",
.run("--spring.cloud.function.definition=date",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
@@ -93,7 +93,7 @@ public class GreenfieldFunctionEnableBindingTests {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ProcessorFromFunction.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=toUpperCase",
"--spring.cloud.function.definition=toUpperCase",
"--spring.jmx.enabled=false")) {
InputDestination source = context.getBean(InputDestination.class);
@@ -110,7 +110,7 @@ public class GreenfieldFunctionEnableBindingTests {
TestChannelBinderConfiguration
.getCompleteConfiguration(SinkFromConsumer.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=sink",
.run("--spring.cloud.function.definition=sink",
"--spring.jmx.enabled=false")) {
InputDestination source = context.getBean(InputDestination.class);
@@ -126,7 +126,7 @@ public class GreenfieldFunctionEnableBindingTests {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
HttpInboundEndpoint.class)).web(WebApplicationType.SERVLET).run(
"--spring.cloud.stream.function.definition=upperCase",
"--spring.cloud.function.definition=upperCase",
"--spring.jmx.enabled=false", "--server.port=0")) {
TestRestTemplate restTemplate = new TestRestTemplate();
restTemplate.postForLocation(
@@ -146,7 +146,7 @@ public class GreenfieldFunctionEnableBindingTests {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FooTransform.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=fooFunction",
"--spring.cloud.function.definition=fooFunction",
"--spring.jmx" + ".enabled=false",
"--logging.level.org.springframework.integration=TRACE")) {
MessageChannel input = context.getBean("input", MessageChannel.class);