Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4389: Default to SimpleSeqSizeReleaseStrategy (#2444)

Browse Source 
* INT-4389: Default to SimpleSeqSizeReleaseStrategy

JIRA: https://jira.spring.io/browse/INT-4389

An `AbstractCorrelatingMessageHandler` fallback to the
`SimpleSequenceSizeReleaseStrategy` when `releaseStrategy` isn't
provided.

* Make `ReleaseStrategyFactoryBean` to fallback to the
`SimpleSequenceSizeReleaseStrategy` for consistency.

* * Fix `ResequencerParser` to fallback to the null `releaseStrategy`.
This way a subsequent `releasePartialSequences` will set a
`SequenceSizeReleaseStrategy` as a default one

* * Fix `BarrierMessageHandler` to populate a default `CorrelationStrategy`

* * More polishing to `BarrierMessageHandler`
This commit is contained in:
Artem Bilan
2018-05-15 16:31:16 -04:00
committed by Gary Russell
parent 1b46224e41
commit 97b00a065b
7 changed files with 73 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -153,6 +153,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
: correlationStrategy);
this.releaseStrategy = releaseStrategy == null ? new SimpleSequenceSizeReleaseStrategy() : releaseStrategy;
this.releaseStrategySet = releaseStrategy != null;
this.sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
}

12
spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-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.
@@ -81,9 +81,7 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler
* @param outputProcessor the output {@link MessageGroupProcessor}.
*/
public BarrierMessageHandler(long timeout, MessageGroupProcessor outputProcessor) {
this(timeout, outputProcessor,
new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
);
this(timeout, outputProcessor, null);
}
/**
@@ -105,10 +103,12 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler
*/
public BarrierMessageHandler(long timeout, MessageGroupProcessor outputProcessor,
CorrelationStrategy correlationStrategy) {
Assert.notNull(outputProcessor, "'messageGroupProcessor' cannot be null");
Assert.notNull(correlationStrategy, "'correlationStrategy' cannot be null");
this.messageGroupProcessor = outputProcessor;
this.correlationStrategy = correlationStrategy;
this.correlationStrategy = (correlationStrategy == null
? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
: correlationStrategy);
this.timeout = timeout;
}

17
spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -25,7 +25,7 @@ import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy;
import org.springframework.integration.aggregator.SimpleSequenceSizeReleaseStrategy;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.util.StringUtils;
@@ -46,10 +46,7 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
private String methodName;
private ReleaseStrategy strategy = new SequenceSizeReleaseStrategy();
public ReleaseStrategyFactoryBean() {
}
private ReleaseStrategy strategy = new SimpleSequenceSizeReleaseStrategy();
public void setTarget(Object target) {
this.target = target;
@@ -79,16 +76,14 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
if (logger.isWarnEnabled()) {
logger.warn("No ReleaseStrategy annotated method found on "
+ this.target.getClass().getSimpleName()
+ "; falling back to SequenceSizeReleaseStrategy, target:"
+ this.target + ", methodName:" + this.methodName);
+ "; falling back to SimpleSequenceSizeReleaseStrategy, target: "
+ this.target + ", methodName: " + this.methodName);
}
}
}
}
else {
if (logger.isWarnEnabled()) {
logger.warn("No target supplied; falling back to SequenceSizeReleaseStrategy");
}
logger.warn("No target supplied; falling back to SimpleSequenceSizeReleaseStrategy");
}
}

61
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-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.
@@ -97,6 +97,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
String propertyName) {
setValueIfAttributeDefined(builder, element, attributeName, propertyName, false);
}
@@ -114,7 +115,9 @@ public abstract class IntegrationNamespaceUtils {
* @param element - the XML element where the attribute should be defined
* @param attributeName - the name of the attribute whose value will be set on the property
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName) {
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
setValueIfAttributeDefined(builder, element, attributeName, false);
}
@@ -131,6 +134,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
String propertyName, boolean emptyStringAllowed) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue) || (emptyStringAllowed && element.hasAttribute(attributeName))) {
builder.addPropertyValue(propertyName, new TypedStringValue(attributeValue));
@@ -156,6 +160,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element, String attributeName,
boolean emptyStringAllowed) {
setValueIfAttributeDefined(builder, element, attributeName,
Conventions.attributeNameToPropertyName(attributeName), emptyStringAllowed);
}
@@ -173,11 +178,13 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, String propertyName) {
setReferenceIfAttributeDefined(builder, element, attributeName, propertyName, false);
}
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, String propertyName, boolean emptyStringAllowed) {
if (element.hasAttribute(attributeName)) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
@@ -209,11 +216,13 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
setReferenceIfAttributeDefined(builder, element, attributeName, false);
}
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, boolean emptyStringAllowed) {
setReferenceIfAttributeDefined(builder, element, attributeName,
Conventions.attributeNameToPropertyName(attributeName), emptyStringAllowed);
}
@@ -245,6 +254,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void configurePollerMetadata(Element pollerElement, BeanDefinitionBuilder targetBuilder,
ParserContext parserContext) {
if (pollerElement.hasAttribute("ref")) {
int numberOfAttributes = pollerElement.getAttributes().getLength();
if (numberOfAttributes != 1) {
@@ -286,6 +296,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static String getTextFromAttributeOrNestedElement(Element element, String name,
ParserContext parserContext) {
String attr = element.getAttribute(name);
Element childElement = DomUtils.getChildElementByTagName(element, name);
if (StringUtils.hasText(attr) && childElement != null) {
@@ -334,6 +345,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void configureHeaderMapper(Element element, BeanDefinitionBuilder rootBuilder,
ParserContext parserContext, Class<?> headerMapperClass, String replyHeaderValue) {
configureHeaderMapper(element, rootBuilder, parserContext,
BeanDefinitionBuilder.genericBeanDefinition(headerMapperClass), replyHeaderValue);
}
@@ -349,6 +361,7 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void configureHeaderMapper(Element element, BeanDefinitionBuilder rootBuilder,
ParserContext parserContext, BeanDefinitionBuilder headerMapperBuilder, String replyHeaderValue) {
String defaultMappedReplyHeadersAttributeName = "mapped-reply-headers";
if (!StringUtils.hasText(replyHeaderValue)) {
replyHeaderValue = defaultMappedReplyHeadersAttributeName;
@@ -368,7 +381,8 @@ public abstract class IntegrationNamespaceUtils {
if (hasMappedRequestHeaders || hasMappedReplyHeaders) {
if (hasMappedRequestHeaders) {
headerMapperBuilder.addPropertyValue("requestHeaderNames", element.getAttribute("mapped-request-headers"));
headerMapperBuilder.addPropertyValue("requestHeaderNames",
element.getAttribute("mapped-request-headers"));
}
if (hasMappedReplyHeaders) {
headerMapperBuilder.addPropertyValue("replyHeaderNames", element.getAttribute(replyHeaderValue));
@@ -423,8 +437,10 @@ public abstract class IntegrationNamespaceUtils {
* @return The bean definition.
*/
public static BeanDefinition configureTransactionDefinition(Element txElement) {
BeanDefinitionBuilder txDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultTransactionAttribute.class);
txDefinitionBuilder.addPropertyValue("propagationBehaviorName", "PROPAGATION_" + txElement.getAttribute("propagation"));
BeanDefinitionBuilder txDefinitionBuilder =
BeanDefinitionBuilder.genericBeanDefinition(DefaultTransactionAttribute.class);
txDefinitionBuilder.addPropertyValue("propagationBehaviorName", "PROPAGATION_"
+ txElement.getAttribute("propagation"));
txDefinitionBuilder.addPropertyValue("isolationLevelName", "ISOLATION_" + txElement.getAttribute("isolation"));
txDefinitionBuilder.addPropertyValue("timeout", txElement.getAttribute("timeout"));
txDefinitionBuilder.addPropertyValue("readOnly", txElement.getAttribute("read-only"));
@@ -442,18 +458,21 @@ public abstract class IntegrationNamespaceUtils {
public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
BeanDefinition parentBeanDefinition, ParserContext parserContext) {
configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, false, parentBeanDefinition, parserContext);
}
public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement,
Element txElement, boolean handleMessageAdvice, BeanDefinition parentBeanDefinition,
ParserContext parserContext) {
configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, handleMessageAdvice,
parentBeanDefinition, parserContext, "adviceChain");
}
public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
BeanDefinition parentBeanDefinition, ParserContext parserContext, String propertyName) {
configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, false, parentBeanDefinition,
parserContext, propertyName);
}
@@ -462,6 +481,7 @@ public abstract class IntegrationNamespaceUtils {
public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
boolean handleMessageAdvice, BeanDefinition parentBeanDefinition, ParserContext parserContext,
String propertyName) {
ManagedList adviceChain = configureAdviceChain(adviceChainElement, txElement, handleMessageAdvice,
parentBeanDefinition, parserContext);
if (!CollectionUtils.isEmpty(adviceChain)) {
@@ -472,12 +492,14 @@ public abstract class IntegrationNamespaceUtils {
@SuppressWarnings("rawtypes")
public static ManagedList configureAdviceChain(Element adviceChainElement, Element txElement,
BeanDefinition parentBeanDefinition, ParserContext parserContext) {
return configureAdviceChain(adviceChainElement, txElement, false, parentBeanDefinition, parserContext);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
public static ManagedList configureAdviceChain(Element adviceChainElement, Element txElement,
boolean handleMessageAdvice, BeanDefinition parentBeanDefinition, ParserContext parserContext) {
ManagedList adviceChain = new ManagedList();
if (txElement != null) {
adviceChain.add(configureTransactionAttributes(txElement, handleMessageAdvice));
@@ -553,7 +575,8 @@ public abstract class IntegrationNamespaceUtils {
String expressionElementValue = element.getAttribute(expressionElementName);
if (StringUtils.hasText(expressionElementValue)) {
BeanDefinitionBuilder expressionDefBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
BeanDefinitionBuilder expressionDefBuilder =
BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
expressionDefBuilder.addConstructorArgValue(expressionElementValue);
return expressionDefBuilder.getRawBeanDefinition();
}
@@ -575,6 +598,7 @@ public abstract class IntegrationNamespaceUtils {
@SuppressWarnings("unchecked")
public static void checkAndConfigureFixedSubscriberChannel(Element element, ParserContext parserContext,
String channelName, String handlerBeanName) {
BeanDefinitionRegistry registry = parserContext.getRegistry();
if (registry.containsBeanDefinition(channelName)) {
BeanDefinition inputChannelDefinition = registry.getBeanDefinition(channelName);
@@ -592,17 +616,21 @@ public abstract class IntegrationNamespaceUtils {
}
else {
BeanDefinition bfppd;
if (!registry.containsBeanDefinition(IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME)) {
if (!registry.containsBeanDefinition(
IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME)) {
bfppd = new RootBeanDefinition(FixedSubscriberChannelBeanFactoryPostProcessor.class);
registry.registerBeanDefinition(IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME, bfppd);
registry.registerBeanDefinition(
IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME, bfppd);
}
else {
bfppd = registry.getBeanDefinition(IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME);
bfppd = registry.getBeanDefinition(
IntegrationContextUtils.INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME);
}
ManagedMap<String, String> candidates;
ValueHolder argumentValue = bfppd.getConstructorArgumentValues().getArgumentValue(0, Map.class);
if (argumentValue == null) {
candidates = new ManagedMap<String, String>();
candidates = new ManagedMap<>();
bfppd.getConstructorArgumentValues().addIndexedArgumentValue(0, candidates);
}
else {
@@ -615,14 +643,19 @@ public abstract class IntegrationNamespaceUtils {
public static void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute,
String expressionAttribute, String beanProperty, String adapterClass, Element element,
BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) {
BeanMetadataElement adapter = constructAdapter(beanRefAttribute, methodRefAttribute, expressionAttribute,
adapterClass, element, processor, parserContext);
builder.addPropertyValue(beanProperty, adapter);
if (adapter != null) {
builder.addPropertyValue(beanProperty, adapter);
}
}
public static void injectConstructorWithAdapter(String beanRefAttribute, String methodRefAttribute,
String expressionAttribute, String adapterClass, Element element,
BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) {
BeanMetadataElement adapter = constructAdapter(beanRefAttribute, methodRefAttribute, expressionAttribute,
adapterClass, element, processor, parserContext);
builder.addConstructorArgValue(adapter);
@@ -631,6 +664,7 @@ public abstract class IntegrationNamespaceUtils {
private static BeanMetadataElement constructAdapter(String beanRefAttribute, String methodRefAttribute,
String expressionAttribute, String adapterClass, Element element, BeanMetadataElement processor,
ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
final String expression = element.getAttribute(expressionAttribute);
@@ -657,14 +691,13 @@ public abstract class IntegrationNamespaceUtils {
else if (processor != null) {
adapter = createAdapter(processor, beanMethod, adapterClass);
}
else {
adapter = createAdapter(null, beanMethod, adapterClass);
}
return adapter;
}
private static BeanMetadataElement createAdapter(BeanMetadataElement ref, String method,
String unqualifiedClassName) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationConfigUtils.BASE_PACKAGE + ".config." + unqualifiedClassName
+ "FactoryBean");

6
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -30,6 +30,7 @@ import org.springframework.integration.aggregator.ResequencingMessageHandler;
* @author Dave Syer
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class ResequencerParser extends AbstractCorrelatingMessageHandlerParser {
@@ -43,10 +44,11 @@ public class ResequencerParser extends AbstractCorrelatingMessageHandlerParser {
builder.addConstructorArgValue(processorBuilder.getBeanDefinition());
this.doParse(builder, element, processorBuilder.getBeanDefinition(), parserContext);
this.doParse(builder, element, null, parserContext);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_PARTIAL_SEQUENCES_ATTRIBUTE);
return builder;
}
}

13
spring-integration-core/src/test/java/org/springframework/integration/config/ReleaseStrategyFactoryBeanTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2016 the original author or authors.
* Copyright 2014-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.
@@ -29,19 +29,20 @@ import org.junit.Test;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy;
import org.springframework.integration.aggregator.SimpleSequenceSizeReleaseStrategy;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.test.util.TestUtils;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0.2
*/
public class ReleaseStrategyFactoryBeanTests {
@Test
public void testRefWithNoMethod() throws Exception {
public void testRefWithNoMethod() {
Foo foo = new Foo();
ReleaseStrategyFactoryBean factory = new ReleaseStrategyFactoryBean();
factory.setTarget(foo);
@@ -53,7 +54,7 @@ public class ReleaseStrategyFactoryBeanTests {
catch (Exception e) {
assertThat(e, instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("Target object of type " +
"[class org.springframework.integration.config.ReleaseStrategyFactoryBeanTests$Foo] " +
"[class org.springframework.integration.config.ReleaseStrategyFactoryBeanTests$Foo] " +
"has no eligible methods for handling Messages."));
}
}
@@ -88,7 +89,7 @@ public class ReleaseStrategyFactoryBeanTests {
ReleaseStrategyFactoryBean factory = new ReleaseStrategyFactoryBean();
factory.afterPropertiesSet();
ReleaseStrategy delegate = factory.getObject();
assertThat(delegate, instanceOf(SequenceSizeReleaseStrategy.class));
assertThat(delegate, instanceOf(SimpleSequenceSizeReleaseStrategy.class));
}
@Test
@@ -98,7 +99,7 @@ public class ReleaseStrategyFactoryBeanTests {
factory.setTarget(foo);
factory.afterPropertiesSet();
ReleaseStrategy delegate = factory.getObject();
assertThat(delegate, instanceOf(SequenceSizeReleaseStrategy.class));
assertThat(delegate, instanceOf(SimpleSequenceSizeReleaseStrategy.class));
}
@Test

4
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-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.
@@ -74,7 +74,7 @@ public class ResequencerParserTests {
"The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag",
false, getPropertyValue(resequencer, "sendPartialResultOnExpiry"));
assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag",
false, getPropertyValue(getPropertyValue(resequencer, "releaseStrategy"), "releasePartialSequences"));
false, getPropertyValue(resequencer, "releasePartialSequences"));
}
@Test