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

INT-2649: DelayHandler: tx & adviceChain support

Browse Source 
* 'delayer-type' XSD: add `<transactional>` & `<advice-chain>`
* move `PollerParser#configureAdviceChain` into `IntegrationNamespaceUtils`
* DelayerParser: parsing `<transactional>` & `<advice-chain>` via `IntegrationNamespaceUtils#configureAdviceChain`
* DelayHandler: add `adviceChain` property
* DelayHandler: introduce `ReleaseMessageHandler` to apply `adviceChain` capabilities around `DelayHandler#doReleaseMessage`
* DelayerParserTests: tests for introduced sub-elements
* DelayerHandlerRescheduleIntegrationTests: test for transaction boundaries with intentional rollback in the message-flow after message release

JIRA: https://jira.springsource.org/browse/INT-2649

Resolve conflicts; polishing.
This commit is contained in:
Artem Bilan
2012-07-10 21:38:47 +03:00
committed by Gary Russell
parent fc1417ed39
commit d4e135b13d
10 changed files with 265 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java
View File

@@ -16,12 +16,12 @@
package org.springframework.integration.config.xml;
import org.springframework.integration.handler.DelayHandler;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.handler.DelayHandler;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Parser for the &lt;delayer&gt; element.
@@ -68,6 +68,13 @@ public class DelayerParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-store");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
Element adviceChainElement = DomUtils.getChildElementByTagName(element, "advice-chain");
IntegrationNamespaceUtils.configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, builder,
parserContext, "delayedAdviceChain");
return builder;
}

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

@@ -292,17 +292,26 @@ public abstract class IntegrationNamespaceUtils {
* @see AbstractPollingEndpoint
*/
public static BeanDefinition configureTransactionAttributes(Element txElement) {
BeanDefinition txDefinition = configureTransactionDefinition(txElement);
BeanDefinitionBuilder attributeSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(MatchAlwaysTransactionAttributeSource.class);
attributeSourceBuilder.addPropertyValue("transactionAttribute", txDefinition);
BeanDefinitionBuilder txInterceptorBuilder = BeanDefinitionBuilder.genericBeanDefinition(TransactionInterceptor.class);
txInterceptorBuilder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
txInterceptorBuilder.addPropertyValue("transactionAttributeSource", attributeSourceBuilder.getBeanDefinition());
return txInterceptorBuilder.getBeanDefinition();
}
/**
* Parse attributes of "transactional" element and configure a {@link DefaultTransactionAttribute}
* with provided "transactionDefinition" properties.
*/
public static BeanDefinition configureTransactionDefinition(Element txElement) {
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"));
BeanDefinitionBuilder attributeSourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(MatchAlwaysTransactionAttributeSource.class);
attributeSourceBuilder.addPropertyValue("transactionAttribute", txDefinitionBuilder.getBeanDefinition());
BeanDefinitionBuilder txInterceptorBuilder = BeanDefinitionBuilder.genericBeanDefinition(TransactionInterceptor.class);
txInterceptorBuilder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
txInterceptorBuilder.addPropertyValue("transactionAttributeSource", attributeSourceBuilder.getBeanDefinition());
return txInterceptorBuilder.getBeanDefinition();
return txDefinitionBuilder.getBeanDefinition();
}
public static String[] generateAlias(Element element) {
@@ -314,12 +323,17 @@ public abstract class IntegrationNamespaceUtils {
return handlerAlias;
}
@SuppressWarnings({ "rawtypes" })
public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
BeanDefinitionBuilder parentBuilder, ParserContext parserContext) {
configureAndSetAdviceChainIfPresent(adviceChainElement, txElement, parentBuilder, parserContext, "adviceChain");
}
@SuppressWarnings({ "rawtypes" })
public static void configureAndSetAdviceChainIfPresent(Element adviceChainElement, Element txElement,
BeanDefinitionBuilder parentBuilder, ParserContext parserContext, String propertyName) {
ManagedList adviceChain = configureAdviceChain(adviceChainElement, txElement, parentBuilder, parserContext);
if (adviceChain != null) {
parentBuilder.addPropertyValue("adviceChain", adviceChain);
parentBuilder.addPropertyValue(propertyName, adviceChain);
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java
View File

@@ -81,9 +81,12 @@ public class PollerParser extends AbstractBeanDefinitionParser {
parserContext.getReaderContext().error(
"the 'ref' attribute must not be present on the top-level 'poller' element", element);
}
configureTrigger(element, metadataBuilder, parserContext);
IntegrationNamespaceUtils.setValueIfAttributeDefined(metadataBuilder, element, "max-messages-per-poll");
IntegrationNamespaceUtils.setValueIfAttributeDefined(metadataBuilder, element, "receive-timeout");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "task-executor");
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
Element adviceChainElement = DomUtils.getChildElementByTagName(element, "advice-chain");
@@ -103,7 +106,6 @@ public class PollerParser extends AbstractBeanDefinitionParser {
pseudoTxElement = pseudoTxElement == null ? txSyncElement : pseudoTxElement;
configureTransactionSync(pseudoTxElement, metadataBuilder, parserContext);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "task-executor");
String errorChannel = element.getAttribute("error-channel");
if (StringUtils.hasText(errorChannel)) {
BeanDefinitionBuilder errorHandler = BeanDefinitionBuilder.genericBeanDefinition(MessagePublishingErrorHandler.class);

64
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -18,11 +18,15 @@ package org.springframework.integration.handler;
import java.io.Serializable;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.store.MessageGroup;
@@ -34,6 +38,8 @@ import org.springframework.jmx.export.annotation.ManagedResource;
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;
/**
* A {@link MessageHandler} that is capable of delaying the continuation of a
@@ -75,8 +81,12 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
private volatile MessageGroupStore messageStore;
private volatile List<Advice> delayedAdviceChain;
private final AtomicBoolean initialized = new AtomicBoolean();
private volatile MessageHandler releaseHandler = new ReleaseMessageHandler();
/**
* Create a DelayHandler with the given 'messageGroupId' that is used as 'key' for {@link MessageGroup}
* to store delayed Messages in the {@link MessageGroupStore}. The sending of Messages after
@@ -126,6 +136,17 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
this.messageStore = messageStore;
}
/**
* Specify the <code>List<Advice></code> to advise {@link DelayHandler.ReleaseMessageHandler} proxy.
* Usually used to add transactions to delayed messages retrieved from a transactional message store.
*
* @see #createReleaseMessageTask
*/
public void setDelayedAdviceChain(List<Advice> delayedAdviceChain) {
Assert.notNull(delayedAdviceChain, "delayedAdviceChain must not be null");
this.delayedAdviceChain = delayedAdviceChain;
}
@Override
public String getComponentType() {
return "delayer";
@@ -140,6 +161,21 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
else {
Assert.isInstanceOf(MessageStore.class, this.messageStore);
}
this.releaseHandler = this.createReleaseMessageTask();
}
private MessageHandler createReleaseMessageTask() {
ReleaseMessageHandler releaseHandler = new ReleaseMessageHandler();
if (!CollectionUtils.isEmpty(this.delayedAdviceChain)) {
ProxyFactory proxyFactory = new ProxyFactory(releaseHandler);
for (Advice advice : delayedAdviceChain) {
proxyFactory.addAdvice(advice);
}
return (MessageHandler) proxyFactory.getProxy(ClassUtils.getDefaultClassLoader());
}
return releaseHandler;
}
/**
@@ -150,7 +186,8 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
*
* @param requestMessage - the Message which may be delayed.
* @return - <code>null</code> if 'requestMessage' is delayed,
* otherwise - 'payload' from 'requestMessage'.
* otherwise - 'payload' from 'requestMessage'.
*
* @see #releaseMessage
*/
@@ -216,6 +253,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
private void releaseMessage(Message<?> message) {
this.releaseHandler.handleMessage(message);
}
private void doReleaseMessage(Message<?> message) {
if (this.messageStore instanceof SimpleMessageStore
|| ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null) {
this.messageStore.removeMessageFromGroup(this.messageGroupId, message);
@@ -265,7 +306,8 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
* in the 'parent-child' contexts, e.g. in the Spring-MVC applications.
*
* @param event - {@link ContextRefreshedEvent} which occurs
* after Application context is completely initialized.
* after Application context is completely initialized.
*
* @see #reschedulePersistedMessages
*/
public void onApplicationEvent(ContextRefreshedEvent event) {
@@ -274,6 +316,23 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
}
/**
* Delegate {@link MessageHandler} implementation for 'release Message task'.
* Used as 'pointcut' to wrap 'release Message task' with <code>adviceChain</code>.
*
* @see @createReleaseMessageTask
* @see @releaseMessage
*/
private class ReleaseMessageHandler implements MessageHandler {
public void handleMessage(Message<?> message) throws MessagingException {
DelayHandler.this.doReleaseMessage(message);
}
}
private static final class DelayedMessageWrapper implements Serializable {
private static final long serialVersionUID = -4739802369074947045L;
@@ -308,6 +367,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
public int hashCode() {
return this.original.hashCode();
}
}
}

10
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
View File

@@ -1298,6 +1298,16 @@
</xsd:element>
<xsd:complexType name="delayer-type">
<xsd:choice>
<xsd:annotation>
<xsd:documentation>
'transactional' and 'advice-chain' elements specify the configuration List of AOP Advice
to proxying DelayHandler's 'release Message task'.
</xsd:documentation>
</xsd:annotation>
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
</xsd:choice>
<xsd:attribute name="id" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>