From d5f1bb65168b10d4b0ef846b2a93bafab747e2fb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 12 Sep 2014 19:29:41 +0300 Subject: [PATCH] INT-3512: Add `` to the Aggregator JIRA: https://jira.spring.io/browse/INT-3512 INT-3512: Fix typos INT-3512: add `expire-` prefix to the advice sub-elements INT-3512: Mark `AggregatorWithCustomReleaseStrategyTests` as `LONG_RUNNING_TEST` Doc Polishing --- .../AbstractCorrelatingMessageHandler.java | 41 ++++++- ...stractCorrelatingMessageHandlerParser.java | 22 ++-- .../integration/handler/DelayHandler.java | 3 +- .../config/xml/spring-integration-4.1.xsd | 9 ++ ...regatorWithCustomReleaseStrategyTests.java | 39 +++++++ .../integration/jdbc/JdbcMessageStore.java | 11 +- .../AggregatorIntegrationTests-context.xml | 47 ++++++++ .../jdbc/AggregatorIntegrationTests.java | 102 ++++++++++++++++++ src/reference/docbook/aggregator.xml | 25 ++++- src/reference/docbook/whats-new.xml | 9 ++ 10 files changed, 291 insertions(+), 17 deletions(-) create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests-context.xml create mode 100644 spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 19a4233158..b42e949c18 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -24,9 +24,11 @@ import java.util.UUID; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.Lock; +import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -116,6 +118,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private volatile Expression groupTimeoutExpression; + private volatile List forceReleaseAdviceChain; + + private MessageGroupProcessor forceReleaseProcessor = new ForceReleaseMessageGroupProcessor(); + private EvaluationContext evaluationContext; private volatile ApplicationEventPublisher applicationEventPublisher; @@ -157,7 +163,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { - forceComplete(group); + forceReleaseProcessor.processMessageGroup(group); } }); } @@ -177,6 +183,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.groupTimeoutExpression = groupTimeoutExpression; } + public void setForceReleaseAdviceChain(List forceReleaseAdviceChain) { + Assert.notNull(forceReleaseAdviceChain, "forceReleaseAdviceChain must not be null"); + this.forceReleaseAdviceChain = forceReleaseAdviceChain; + } + @Override public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { this.evaluationContext = evaluationContext; @@ -226,6 +237,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP * (checked in the setter). */ this.lockRegistrySet = true; + this.forceReleaseProcessor = createGroupTimeoutProcessor(); + } + + private MessageGroupProcessor createGroupTimeoutProcessor() { + MessageGroupProcessor processor = new ForceReleaseMessageGroupProcessor(); + + if (this.groupTimeoutExpression != null && !CollectionUtils.isEmpty(this.forceReleaseAdviceChain)) { + ProxyFactory proxyFactory = new ProxyFactory(processor); + for (Advice advice : this.forceReleaseAdviceChain) { + proxyFactory.addAdvice(advice); + } + return (MessageGroupProcessor) proxyFactory.getProxy(getApplicationContext().getClassLoader()); + } + return processor; } public void setDiscardChannel(MessageChannel discardChannel) { @@ -409,7 +434,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override public void run() { try { - AbstractCorrelatingMessageHandler.this.forceComplete(messageGroup); + forceReleaseProcessor.processMessageGroup(messageGroup); } catch (MessageDeliveryException e) { if (logger.isDebugEnabled()) { @@ -427,7 +452,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(messageGroup.getGroupId()), scheduledFuture); } else { - forceComplete(messageGroup); + this.forceReleaseProcessor.processMessageGroup(messageGroup); } } } @@ -735,4 +760,14 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } + private class ForceReleaseMessageGroupProcessor implements MessageGroupProcessor { + + @Override + public Object processMessageGroup(MessageGroup group) { + forceComplete(group); + return null; + } + + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index 2b836420cd..7b3078c620 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -22,6 +22,7 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; /** * Base class for parsers that create an instance of {@link AbstractCorrelatingMessageHandler} @@ -58,7 +59,8 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry"; - protected void doParse(BeanDefinitionBuilder builder, Element element, BeanMetadataElement processor, ParserContext parserContext){ + protected void doParse(BeanDefinitionBuilder builder, Element element, BeanMetadataElement processor, + ParserContext parserContext) { this.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy", element, builder, processor, parserContext); @@ -72,12 +74,19 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "lock-registry"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "empty-group-min-timeout", "minimumTimeoutForEmptyGroups"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "empty-group-min-timeout", + "minimumTimeoutForEmptyGroups"); BeanDefinition expressionDef = - IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("group-timeout", "group-timeout-expression", - parserContext, element, false); + IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("group-timeout", + "group-timeout-expression", parserContext, element, false); builder.addPropertyValue("groupTimeoutExpression", expressionDef); + + Element txElement = DomUtils.getChildElementByTagName(element, "expire-transactional"); + Element adviceChainElement = DomUtils.getChildElementByTagName(element, "expire-advice-chain"); + + IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, + builder.getRawBeanDefinition(), parserContext, "forceReleaseAdviceChain"); } protected void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute, @@ -92,8 +101,8 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo final boolean hasExpression = StringUtils.hasText(expression); if (hasBeanRef && hasExpression) { - parserContext.getReaderContext().error("Exactly one of the '" + beanRefAttribute + "' or '" + expressionAttribute + - "' attribute is allowed.", element); + parserContext.getReaderContext().error("Exactly one of the '" + beanRefAttribute + "' or '" + + expressionAttribute + "' attribute is allowed.", element); } BeanMetadataElement adapter = null; @@ -126,4 +135,5 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo } return builder.getBeanDefinition(); } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index a70c7c6db1..d58ef501e0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -46,7 +46,6 @@ import org.springframework.messaging.MessagingException; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; /** @@ -237,7 +236,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement for (Advice advice : delayedAdviceChain) { proxyFactory.addAdvice(advice); } - return (MessageHandler) proxyFactory.getProxy(ClassUtils.getDefaultClassLoader()); + return (MessageHandler) proxyFactory.getProxy(getApplicationContext().getClassLoader()); } return releaseHandler; } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 0ce6fa47b2..d5a8761a0d 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -3435,6 +3435,15 @@ + + + + 'transactional' or 'advice-chain' are applied only to 'forceComplete' operation. + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java index 380c0a1235..c29f149223 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/AggregatorWithCustomReleaseStrategyTests.java @@ -21,7 +21,12 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.junit.AfterClass; +import org.junit.Assume; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; +import org.junit.runners.model.Statement; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -37,6 +42,40 @@ import org.springframework.messaging.MessageChannel; */ public class AggregatorWithCustomReleaseStrategyTests { + @Rule + public TestWatcher longTests = new TestWatcher() { + + private static final String RUN_LONG_PROP = "RUN_LONG_INTEGRATION_TESTS"; + + private boolean shouldRun; + + { + for(String value: new String[]{System.getenv(RUN_LONG_PROP), System.getProperty(RUN_LONG_PROP)}) { + if ("true".equalsIgnoreCase(value)) { + this.shouldRun = true; + break; + } + } + } + + @Override + public Statement apply(Statement base, Description description) { + if (!this.shouldRun) { + return new Statement() { + + @Override + public void evaluate() throws Throwable { + Assume.assumeTrue(false); + } + }; + } + else { + return super.apply(base, description); + } + } + + }; + private static ExecutorService executor = Executors.newCachedThreadPool(); @AfterClass diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 4632039cef..a3be704595 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -308,7 +308,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override @ManagedAttribute public long getMessageCount() { - return jdbcTemplate.queryForObject(getQuery(Query.GET_MESSAGE_COUNT), Integer.class, region); + return jdbcTemplate.queryForObject(getQuery(Query.GET_MESSAGE_COUNT), Long.class, region); } @Override @@ -334,7 +334,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final long createdDate = System.currentTimeMillis(); Message result = this.getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) - .setHeader(CREATED_DATE_KEY, new Long(createdDate)).build(); + .setHeader(CREATED_DATE_KEY, createdDate).build(); Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers"); // using reflection to set ID since it is immutable through MessageHeaders @@ -452,7 +452,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } long timestamp = createDate.get().getTime(); - boolean complete = completeFlag.get().booleanValue(); + boolean complete = completeFlag.get(); SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, complete); messageGroup.setLastModified(updateDate.get().getTime()); @@ -710,8 +710,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override public Message mapRow(ResultSet rs, int rowNum) throws SQLException { - Message message = (Message) deserializer.convert(lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES")); - return message; + return (Message) deserializer.convert(lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES")); } + } + } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests-context.xml new file mode 100644 index 0000000000..5416b0bf78 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests-context.xml @@ -0,0 +1,47 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests.java new file mode 100644 index 0000000000..e14260d474 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/AggregatorIntegrationTests.java @@ -0,0 +1,102 @@ +/* + * Copyright 2014 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.integration.jdbc; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationAdapter; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class AggregatorIntegrationTests { + + @Autowired + private MessageChannel transactionalAggregatorInput; + + @Autowired + private MessageGroupStore messageGroupStore; + + @Test + public void testTransactionalAggregatorGroupTimeout() throws InterruptedException { + this.transactionalAggregatorInput.send(new GenericMessage(1, stubHeaders(1, 2, 1))); + + assertTrue(RollbackTxSync.latch.await(20, TimeUnit.SECONDS)); + + //As far as we have been within TX, the message group should still be in the MessageStore + assertEquals(1, this.messageGroupStore.messageGroupSize(1)); + } + + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) { + Map headers = new HashMap(); + headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); + headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); + headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); + return headers; + } + + @SuppressWarnings("unused") + private static class ExceptionMessageHandler implements MessageHandler { + + public void handleMessage(Message message) throws MessagingException { + TransactionSynchronizationManager.registerSynchronization(new RollbackTxSync()); + throw new RuntimeException("intentional"); + } + + } + + private static class RollbackTxSync extends TransactionSynchronizationAdapter { + + public static CountDownLatch latch = new CountDownLatch(1); + + @Override + public void afterCompletion(int status) { + if (TransactionSynchronization.STATUS_ROLLED_BACK == status) { + latch.countDown(); + } + } + + } + +} diff --git a/src/reference/docbook/aggregator.xml b/src/reference/docbook/aggregator.xml index 242c5f1422..db788a6bd6 100644 --- a/src/reference/docbook/aggregator.xml +++ b/src/reference/docbook/aggregator.xml @@ -376,7 +376,10 @@ then you should simply provide an implementation of the ReleaseStrate group-timeout-expression="size() ge 2 ? 100 : -1" ]]> ]]> ]]> ]]> ]]> @@ -610,6 +613,26 @@ then you should simply provide an implementation of the ReleaseStrate group-timeout-expression is not specified. + + + Since version 4.1. Allows a transaction to be started for the + forceComplete operation. It is initiated from a + group-timeout(-expression) or by a MessageGroupStoreReaper and + is not applied to the normal add/release/discard operations. Only this sub-element or + <expire-advice-chain/> is allowed. + + + + + Since version 4.1. Allows the configuration of any + Advice for the + forceComplete operation. It is initiated from a + group-timeout(-expression) or by a MessageGroupStoreReaper and + is not applied to the normal add/release/discard operations. Only this sub-element or + <expire-transactional/> is allowed. A transaction Advice + can also be configured here using the Spring tx namespace. + + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index b906ffc559..da6a5f7c6a 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -203,5 +203,14 @@ See . +
+ Aggregator Advice Chain + + Aggregators and Resequencers now support an + <expire-advice-chain/> and <expire-transactional/> sub-elements + to advise the forceComplete operation. + See for more information. + +