INT-3736: Process Barrier Component
JIRA: https://jira.spring.io/browse/INT-3736 INT-3736: Polishing INT-3736: Use o-c-a For Release Avoids unusual parser logic, allows release channel to be of any type. INT-3736: Schema Polishing INT-3736: Polishing - Add TriggerMessageHandler interface - Use SyncQueue - suspend the trigger thread if the main thread hasn't arrived yet - allows clean up of state regardless of whether just a trigger or suspend is processed - reply-required INT-3736: Polishing and Docs Polishing code style and fixing typos in docs
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
4266bb53b4
commit
c695129bd5
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2014 the original author or authors.
|
||||
* Copyright 2002-2015 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
|
||||
@@ -10,18 +10,16 @@
|
||||
* 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.config.xml;
|
||||
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.BeanMetadataElement;
|
||||
import org.springframework.beans.factory.config.BeanDefinition;
|
||||
import org.springframework.beans.factory.config.RuntimeBeanReference;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
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;
|
||||
|
||||
/**
|
||||
@@ -30,8 +28,9 @@ import org.springframework.util.xml.DomUtils;
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Stefan Ferstl
|
||||
* @author Artem Bilan
|
||||
* @since 2.1
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 2.1
|
||||
*/
|
||||
public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractConsumerEndpointParser {
|
||||
|
||||
@@ -62,11 +61,11 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
|
||||
private static final String EXPIRE_GROUPS_UPON_TIMEOUT = "expire-groups-upon-timeout";
|
||||
|
||||
protected void doParse(BeanDefinitionBuilder builder, Element element, BeanMetadataElement processor,
|
||||
ParserContext parserContext) {
|
||||
this.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE,
|
||||
ParserContext parserContext) {
|
||||
IntegrationNamespaceUtils.injectPropertyWithAdapter(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE,
|
||||
CORRELATION_STRATEGY_EXPRESSION_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy",
|
||||
element, builder, processor, parserContext);
|
||||
this.injectPropertyWithAdapter(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE,
|
||||
IntegrationNamespaceUtils.injectPropertyWithAdapter(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE,
|
||||
RELEASE_STRATEGY_EXPRESSION_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy",
|
||||
element, builder, processor, parserContext);
|
||||
|
||||
@@ -93,51 +92,4 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_TIMEOUT);
|
||||
}
|
||||
|
||||
protected void injectPropertyWithAdapter(String beanRefAttribute, String methodRefAttribute,
|
||||
String expressionAttribute, String beanProperty, String adapterClass, Element element,
|
||||
BeanDefinitionBuilder builder, BeanMetadataElement processor, ParserContext parserContext) {
|
||||
|
||||
final String beanRef = element.getAttribute(beanRefAttribute);
|
||||
final String beanMethod = element.getAttribute(methodRefAttribute);
|
||||
final String expression = element.getAttribute(expressionAttribute);
|
||||
|
||||
final boolean hasBeanRef = StringUtils.hasText(beanRef);
|
||||
final boolean hasExpression = StringUtils.hasText(expression);
|
||||
|
||||
if (hasBeanRef && hasExpression) {
|
||||
parserContext.getReaderContext().error("Exactly one of the '" + beanRefAttribute + "' or '"
|
||||
+ expressionAttribute + "' attribute is allowed.", element);
|
||||
}
|
||||
|
||||
BeanMetadataElement adapter = null;
|
||||
if (hasBeanRef) {
|
||||
adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass);
|
||||
}
|
||||
else if (hasExpression) {
|
||||
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
|
||||
.genericBeanDefinition(IntegrationConfigUtils.BASE_PACKAGE + ".aggregator.ExpressionEvaluating"
|
||||
+ adapterClass);
|
||||
adapterBuilder.addConstructorArgValue(expression);
|
||||
adapter = adapterBuilder.getBeanDefinition();
|
||||
}
|
||||
else if (processor != null) {
|
||||
adapter = this.createAdapter(processor, beanMethod, adapterClass);
|
||||
}
|
||||
else {
|
||||
adapter = this.createAdapter(null, beanMethod, adapterClass);
|
||||
}
|
||||
builder.addPropertyValue(beanProperty, adapter);
|
||||
}
|
||||
|
||||
private BeanMetadataElement createAdapter(BeanMetadataElement ref, String method, String unqualifiedClassName) {
|
||||
BeanDefinitionBuilder builder = BeanDefinitionBuilder
|
||||
.genericBeanDefinition(IntegrationConfigUtils.BASE_PACKAGE + ".config." + unqualifiedClassName
|
||||
+ "FactoryBean");
|
||||
builder.addConstructorArgValue(ref);
|
||||
if (StringUtils.hasText(method)) {
|
||||
builder.addConstructorArgValue(method);
|
||||
}
|
||||
return builder.getBeanDefinition();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
/*
|
||||
* Copyright 2015 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.config.xml;
|
||||
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.handler.BarrierMessageHandler;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Parser for {@code <int:barrier/>}.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 4.2
|
||||
*/
|
||||
public class BarrierParser extends AbstractConsumerEndpointParser {
|
||||
|
||||
@Override
|
||||
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
|
||||
BeanDefinitionBuilder handlerBuilder = BeanDefinitionBuilder
|
||||
.genericBeanDefinition(BarrierMessageHandler.class);
|
||||
handlerBuilder.addConstructorArgValue(element.getAttribute("timeout"));
|
||||
String processor = element.getAttribute("output-processor");
|
||||
if (StringUtils.hasText(processor)) {
|
||||
handlerBuilder.addConstructorArgReference(processor);
|
||||
}
|
||||
IntegrationNamespaceUtils.injectConstructorWithAdapter("correlation-strategy",
|
||||
"correlation-strategy-method", "correlation-strategy-expression",
|
||||
"CorrelationStrategy", element, handlerBuilder, null, parserContext);
|
||||
return handlerBuilder;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -84,6 +84,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
|
||||
registerBeanDefinitionParser("scatter-gather", new ScatterGatherParser());
|
||||
registerBeanDefinitionParser("idempotent-receiver", new IdempotentReceiverInterceptorParser());
|
||||
registerBeanDefinitionParser("management", new IntegrationManagementParser());
|
||||
registerBeanDefinitionParser("barrier", new BarrierParser());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ import org.w3c.dom.Element;
|
||||
import org.w3c.dom.Node;
|
||||
import org.w3c.dom.NodeList;
|
||||
|
||||
import org.springframework.beans.BeanMetadataElement;
|
||||
import org.springframework.beans.factory.config.BeanDefinition;
|
||||
import org.springframework.beans.factory.config.BeanDefinitionHolder;
|
||||
import org.springframework.beans.factory.config.BeanReference;
|
||||
@@ -564,4 +565,66 @@ public abstract class IntegrationNamespaceUtils {
|
||||
lifecycles.add(new RuntimeBeanReference(beanName));
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
final boolean hasBeanRef = StringUtils.hasText(beanRef);
|
||||
final boolean hasExpression = StringUtils.hasText(expression);
|
||||
|
||||
if (hasBeanRef && hasExpression) {
|
||||
parserContext.getReaderContext().error("Exactly one of the '" + beanRefAttribute + "' or '"
|
||||
+ expressionAttribute + "' attribute is allowed.", element);
|
||||
}
|
||||
|
||||
BeanMetadataElement adapter = null;
|
||||
if (hasBeanRef) {
|
||||
adapter = createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass);
|
||||
}
|
||||
else if (hasExpression) {
|
||||
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
|
||||
.genericBeanDefinition(IntegrationConfigUtils.BASE_PACKAGE + ".aggregator.ExpressionEvaluating"
|
||||
+ adapterClass);
|
||||
adapterBuilder.addConstructorArgValue(expression);
|
||||
adapter = adapterBuilder.getBeanDefinition();
|
||||
}
|
||||
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");
|
||||
builder.addConstructorArgValue(ref);
|
||||
if (StringUtils.hasText(method)) {
|
||||
builder.addConstructorArgValue(method);
|
||||
}
|
||||
return builder.getBeanDefinition();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
/*
|
||||
* Copyright 2015 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.handler;
|
||||
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.SynchronousQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.aggregator.CorrelationStrategy;
|
||||
import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
|
||||
import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
|
||||
import org.springframework.integration.aggregator.MessageGroupProcessor;
|
||||
import org.springframework.integration.store.SimpleMessageGroup;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandlingException;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* A message handler that suspends the thread until a message with corresponding
|
||||
* correlation is passed into the {@link #trigger(Message) trigger} method or
|
||||
* the timeout occurs. Only one thread with a particular correlation (result of invoking
|
||||
* the {@link CorrelationStrategy}) can be suspended at a time. If the inbound thread does
|
||||
* not arrive before the trigger thread, the latter is suspended until it does, or the
|
||||
* timeout occurs.
|
||||
* <p>
|
||||
* The default {@link CorrelationStrategy} is a {@link HeaderAttributeCorrelationStrategy}.
|
||||
* <p>
|
||||
* The default output processor is a {@link DefaultAggregatingMessageGroupProcessor}.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 4.2
|
||||
*/
|
||||
public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler implements MessageTriggerAction {
|
||||
|
||||
private final ConcurrentMap<Object, SynchronousQueue<Message<?>>> suspensions =
|
||||
new ConcurrentHashMap<Object, SynchronousQueue<Message<?>>>();
|
||||
|
||||
private final ConcurrentMap<Object, Thread> inProcess = new ConcurrentHashMap<Object, Thread>();
|
||||
|
||||
private final long timeout;
|
||||
|
||||
private final CorrelationStrategy correlationStrategy;
|
||||
|
||||
private final MessageGroupProcessor messageGroupProcessor;
|
||||
|
||||
/**
|
||||
* Construct an instance with the provided timeout and default correlation and
|
||||
* output strategies.
|
||||
* @param timeout the timeout in milliseconds.
|
||||
*/
|
||||
public BarrierMessageHandler(long timeout) {
|
||||
this(timeout, new DefaultAggregatingMessageGroupProcessor());
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance with the provided timeout and output processor, and default
|
||||
* correlation strategy.
|
||||
* @param timeout the timeout in milliseconds.
|
||||
* @param outputProcessor the output {@link MessageGroupProcessor}.
|
||||
*/
|
||||
public BarrierMessageHandler(long timeout, MessageGroupProcessor outputProcessor) {
|
||||
this(timeout, outputProcessor,
|
||||
new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance with the provided timeout and correlation strategy, and default
|
||||
* output processor.
|
||||
* @param timeout the timeout in milliseconds.
|
||||
* @param correlationStrategy the correlation strategy.
|
||||
*/
|
||||
public BarrierMessageHandler(long timeout, CorrelationStrategy correlationStrategy) {
|
||||
this(timeout, new DefaultAggregatingMessageGroupProcessor(), correlationStrategy);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct an instance with the provided timeout and output processor, and default
|
||||
* correlation strategy.
|
||||
* @param timeout the timeout in milliseconds.
|
||||
* @param outputProcessor the output {@link MessageGroupProcessor}.
|
||||
* @param correlationStrategy the correlation strategy.
|
||||
*/
|
||||
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.timeout = timeout;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Object handleRequestMessage(Message<?> requestMessage) {
|
||||
Object key = this.correlationStrategy.getCorrelationKey(requestMessage);
|
||||
if (key == null) {
|
||||
throw new MessagingException(requestMessage, "Correlation Strategy returned null");
|
||||
}
|
||||
Thread existing = this.inProcess.putIfAbsent(key, Thread.currentThread());
|
||||
if (existing != null) {
|
||||
throw new MessagingException(requestMessage, "Correlation key ("
|
||||
+ key + ") is already in use by " + existing.getName());
|
||||
}
|
||||
SynchronousQueue<Message<?>> syncQueue = createOrObtainQueue(key);
|
||||
try {
|
||||
Message<?> releaseMessage = syncQueue.poll(this.timeout, TimeUnit.MILLISECONDS);
|
||||
if (releaseMessage != null) {
|
||||
return processRelease(key, requestMessage, releaseMessage);
|
||||
}
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new MessageHandlingException(requestMessage, "Interrupted while waiting for release", e);
|
||||
}
|
||||
finally {
|
||||
this.inProcess.remove(key);
|
||||
this.suspensions.remove(key);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private Object processRelease(Object key, Message<?> requestMessage, Message<?> releaseMessage) {
|
||||
this.suspensions.remove(key);
|
||||
if (releaseMessage.getPayload() instanceof Throwable) {
|
||||
throw new MessagingException(requestMessage, "Releasing flow returned a throwable",
|
||||
(Throwable) releaseMessage.getPayload());
|
||||
}
|
||||
else {
|
||||
return buildResult(key, requestMessage, releaseMessage);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Override to change the default mechanism by which the inbound and release messages
|
||||
* are returned as a result.
|
||||
* @param key The correlation key.
|
||||
* @param requestMessage the inbound message.
|
||||
* @param releaseMessage the release message.
|
||||
* @return the result.
|
||||
*/
|
||||
protected Object buildResult(Object key, Message<?> requestMessage, Message<?> releaseMessage) {
|
||||
SimpleMessageGroup group = new SimpleMessageGroup(key);
|
||||
group.add(requestMessage);
|
||||
group.add(releaseMessage);
|
||||
return this.messageGroupProcessor.processMessageGroup(group);
|
||||
}
|
||||
|
||||
private SynchronousQueue<Message<?>> createOrObtainQueue(Object key) {
|
||||
SynchronousQueue<Message<?>> syncQueue = new SynchronousQueue<Message<?>>();
|
||||
SynchronousQueue<Message<?>> existing = this.suspensions.putIfAbsent(key, syncQueue);
|
||||
if (existing != null) {
|
||||
syncQueue = existing;
|
||||
}
|
||||
return syncQueue;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void trigger(Message<?> message) {
|
||||
Object key = this.correlationStrategy.getCorrelationKey(message);
|
||||
if (key == null) {
|
||||
throw new MessagingException(message, "Correlation Strategy returned null");
|
||||
}
|
||||
SynchronousQueue<Message<?>> syncQueue = createOrObtainQueue(key);
|
||||
try {
|
||||
if (!syncQueue.offer(message, timeout, TimeUnit.MILLISECONDS)) {
|
||||
this.logger.error("Suspending thread timed out or did not arrive within timeout for: " + message);
|
||||
this.suspensions.remove(key);
|
||||
}
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
this.logger.error("Interrupted while waiting for the suspending thread for: " + message);
|
||||
this.suspensions.remove(key);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
* Copyright 2015 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.handler;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* Classes implementing this interface can take some action when a trigger {@link Message}
|
||||
* is received.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @since 4.2
|
||||
*
|
||||
*/
|
||||
public interface MessageTriggerAction {
|
||||
|
||||
/**
|
||||
* Take some action based on the message.
|
||||
* @param message the message.
|
||||
*/
|
||||
void trigger(Message<?> message);
|
||||
|
||||
}
|
||||
@@ -1659,6 +1659,91 @@
|
||||
<xsd:attribute name="id" type="xsd:string" use="required" />
|
||||
</xsd:complexType>
|
||||
|
||||
<xsd:element name="barrier">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines an endpoint that suspends a thread until a corresponding release message is received.
|
||||
The endpoint's message handler is a 'BarrierMessageHandler'.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexType>
|
||||
<xsd:choice minOccurs="0" maxOccurs="2">
|
||||
<xsd:element name="request-handler-advice-chain" type="handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
|
||||
<xsd:element ref="poller" />
|
||||
</xsd:choice>
|
||||
<xsd:attributeGroup ref="inputOutputChannelGroup" />
|
||||
<xsd:attribute name="id" type="xsd:string" use="required"/>
|
||||
<xsd:attribute name="timeout" type="xsd:string" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
The time in milliseconds to suspend the thread. See `requires-reply`.
|
||||
Also, if the trigger message is received first, the time that thread will
|
||||
wait before logging an error and exiting.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="requires-reply" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Specify whether the barrier must return a non-null value. This value will be
|
||||
'false' by default, but if set to 'true', a ReplyRequiredException will be thrown when
|
||||
the barrier times out.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="correlation-strategy" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type type="java.lang.Object" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
<xsd:documentation>
|
||||
A reference to a bean that implements the decision algorithm as to whether a given
|
||||
message group is complete. The bean can be an implementation of the
|
||||
CorrelationStrategy interface or a POJO.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="correlation-strategy-method" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation>
|
||||
<tool:expected-method type-ref="@correlation-strategy" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
<xsd:documentation>
|
||||
A method defined on the bean referenced by correlation-strategy, that implements the
|
||||
correlation decision algorithm
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="correlation-strategy-expression" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
A SpEL expression which implements correlation decision
|
||||
algorithm to apply to the Message (e.g., payload.getPerson().getId() - correlate
|
||||
based on the 'id' of the 'person' attribute of the message payload object)
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="output-processor" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type type="org.springframework.integration.aggregator.MessageGroupProcessor" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
<xsd:documentation>
|
||||
A reference to a bean that implements 'MessageGroupProcessor'. The processor is invoked to
|
||||
produce the result when the release is triggered. By default the payloads of the two
|
||||
messages are aggregated as a 'Collection' and the message headers are merged.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="bridge">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:int="http://www.springframework.org/schema/integration"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
|
||||
|
||||
<int:channel id="in">
|
||||
<int:queue />
|
||||
</int:channel>
|
||||
|
||||
<int:barrier id="barrier1" input-channel="in" output-channel="out" correlation-strategy-expression="'foo'"
|
||||
timeout="10000">
|
||||
<int:poller fixed-delay="100" />
|
||||
</int:barrier>
|
||||
|
||||
<int:channel id="out">
|
||||
<int:queue />
|
||||
</int:channel>
|
||||
|
||||
<int:channel id="release" />
|
||||
|
||||
<int:outbound-channel-adapter channel="release" ref="barrier1.handler" method="trigger" />
|
||||
|
||||
<int:barrier id="barrier2" input-channel="nullChannel" timeout="10000" auto-startup="false">
|
||||
<int:poller fixed-delay="1000000" />
|
||||
</int:barrier>
|
||||
|
||||
<int:barrier id="barrier3" input-channel="release" auto-startup="false" timeout="123"
|
||||
output-processor="mgp" correlation-strategy="cs" requires-reply="true" />
|
||||
|
||||
<bean id="mgp" class="org.springframework.integration.config.xml.BarrierParserTests$TestMGP" />
|
||||
|
||||
<bean id="cs" class="org.springframework.integration.config.xml.BarrierParserTests$TestCS" />
|
||||
|
||||
</beans>
|
||||
@@ -0,0 +1,113 @@
|
||||
/*
|
||||
* Copyright 2015 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.config.xml;
|
||||
|
||||
import static org.hamcrest.Matchers.instanceOf;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.integration.aggregator.CorrelationStrategy;
|
||||
import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
|
||||
import org.springframework.integration.aggregator.MessageGroupProcessor;
|
||||
import org.springframework.integration.endpoint.EventDrivenConsumer;
|
||||
import org.springframework.integration.endpoint.PollingConsumer;
|
||||
import org.springframework.integration.handler.BarrierMessageHandler;
|
||||
import org.springframework.integration.store.MessageGroup;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
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;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 4.2
|
||||
*
|
||||
*/
|
||||
@ContextConfiguration
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@DirtiesContext
|
||||
public class BarrierParserTests {
|
||||
|
||||
@Autowired
|
||||
private MessageChannel in;
|
||||
|
||||
@Autowired
|
||||
private MessageChannel release;
|
||||
|
||||
@Autowired
|
||||
private PollableChannel out;
|
||||
|
||||
@Autowired
|
||||
private PollingConsumer barrier1;
|
||||
|
||||
@Autowired
|
||||
private PollingConsumer barrier2;
|
||||
|
||||
@Autowired
|
||||
private EventDrivenConsumer barrier3;
|
||||
|
||||
@Test
|
||||
public void parserTestsWithMessage() {
|
||||
this.in.send(new GenericMessage<String>("foo"));
|
||||
this.release.send(new GenericMessage<String>("bar"));
|
||||
Message<?> received = out.receive(10000);
|
||||
assertNotNull(received);
|
||||
this.barrier1.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void parserFieldPopulationTests() {
|
||||
BarrierMessageHandler handler = TestUtils.getPropertyValue(this.barrier1, "handler",
|
||||
BarrierMessageHandler.class);
|
||||
assertEquals(10000L, TestUtils.getPropertyValue(handler, "timeout"));
|
||||
assertThat(TestUtils.getPropertyValue(this.barrier2, "handler.correlationStrategy"),
|
||||
instanceOf(HeaderAttributeCorrelationStrategy.class));
|
||||
assertThat(TestUtils.getPropertyValue(this.barrier3, "handler.messageGroupProcessor"),
|
||||
instanceOf(TestMGP.class));
|
||||
assertThat(TestUtils.getPropertyValue(this.barrier3, "handler.correlationStrategy"),
|
||||
instanceOf(TestCS.class));
|
||||
|
||||
}
|
||||
|
||||
public static class TestMGP implements MessageGroupProcessor {
|
||||
|
||||
@Override
|
||||
public Object processMessageGroup(MessageGroup group) {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class TestCS implements CorrelationStrategy {
|
||||
|
||||
@Override
|
||||
public Object getCorrelationKey(Message<?> message) {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
/*
|
||||
* Copyright 2015 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.handler;
|
||||
|
||||
import static org.hamcrest.Matchers.containsString;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertSame;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.spy;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.integration.annotation.Poller;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.test.util.TestUtils;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @since 4.2
|
||||
*
|
||||
*/
|
||||
@ContextConfiguration
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@DirtiesContext
|
||||
public class BarrierMessageHandlerTests {
|
||||
|
||||
@Autowired
|
||||
private MessageChannel in;
|
||||
|
||||
@Autowired
|
||||
private PollableChannel out;
|
||||
|
||||
@Autowired
|
||||
private MessageChannel release;
|
||||
|
||||
@Test
|
||||
public void testRequestBeforeReply() throws Exception {
|
||||
final BarrierMessageHandler handler = new BarrierMessageHandler(10000);
|
||||
QueueChannel outputChannel = new QueueChannel();
|
||||
handler.setOutputChannel(outputChannel);
|
||||
handler.setBeanFactory(mock(BeanFactory.class));
|
||||
handler.afterPropertiesSet();
|
||||
final AtomicReference<Exception> dupCorrelation = new AtomicReference<Exception>();
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Runnable runnable = new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
|
||||
}
|
||||
catch (MessagingException e) {
|
||||
dupCorrelation.set(e);
|
||||
}
|
||||
latch.countDown();
|
||||
}
|
||||
|
||||
};
|
||||
ExecutorService exec = Executors.newCachedThreadPool();
|
||||
exec.execute(runnable);
|
||||
exec.execute(runnable);
|
||||
Map<?, ?> suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class);
|
||||
int n = 0;
|
||||
while (n++ < 100 && suspensions.size() == 0) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
Map<?, ?> inProcess = TestUtils.getPropertyValue(handler, "inProcess", Map.class);
|
||||
assertEquals(1, inProcess.size());
|
||||
assertTrue("suspension did not appear in time", n < 100);
|
||||
assertTrue(latch.await(10, TimeUnit.SECONDS));
|
||||
assertNotNull(dupCorrelation.get());
|
||||
assertThat(dupCorrelation.get().getMessage(), Matchers.startsWith("Correlation key (foo) is already in use by"));
|
||||
handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build());
|
||||
Message<?> received = outputChannel.receive(10000);
|
||||
assertNotNull(received);
|
||||
List<?> result = (List<?>) received.getPayload();
|
||||
assertEquals("foo", result.get(0));
|
||||
assertEquals("bar", result.get(1));
|
||||
assertEquals(0, suspensions.size());
|
||||
assertEquals(0, inProcess.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testReplyBeforeRequest() throws Exception {
|
||||
final BarrierMessageHandler handler = new BarrierMessageHandler(10000);
|
||||
QueueChannel outputChannel = new QueueChannel();
|
||||
handler.setOutputChannel(outputChannel);
|
||||
handler.setBeanFactory(mock(BeanFactory.class));
|
||||
handler.afterPropertiesSet();
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build());
|
||||
}
|
||||
|
||||
});
|
||||
Map<?, ?> suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class);
|
||||
int n = 0;
|
||||
while (n++ < 100 && suspensions.size() == 0) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
assertTrue("suspension did not appear in time", n < 100);
|
||||
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
|
||||
Message<?> received = outputChannel.receive(10000);
|
||||
assertNotNull(received);
|
||||
List<?> result = (ArrayList<?>) received.getPayload();
|
||||
assertEquals("foo", result.get(0));
|
||||
assertEquals("bar", result.get(1));
|
||||
assertEquals(0, suspensions.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLateReply() throws Exception {
|
||||
final BarrierMessageHandler handler = new BarrierMessageHandler(0);
|
||||
QueueChannel outputChannel = new QueueChannel();
|
||||
handler.setOutputChannel(outputChannel);
|
||||
handler.setBeanFactory(mock(BeanFactory.class));
|
||||
handler.afterPropertiesSet();
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
|
||||
}
|
||||
|
||||
});
|
||||
Map<?, ?> suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class);
|
||||
int n = 0;
|
||||
while (n++ < 100 && suspensions.size() != 0) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
assertTrue("suspension not removed", n < 100);
|
||||
Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class));
|
||||
new DirectFieldAccessor(handler).setPropertyValue("logger", logger);
|
||||
handler.trigger(MessageBuilder.withPayload("bar").setCorrelationId("foo").build());
|
||||
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
|
||||
verify(logger).error(captor.capture());
|
||||
assertThat(captor.getValue(),
|
||||
Matchers.allOf(containsString("Suspending thread timed out or did not arrive within timeout for:"),
|
||||
containsString("payload=bar")));
|
||||
assertEquals(0, suspensions.size());
|
||||
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
|
||||
assertEquals(0, suspensions.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExceptionReply() throws Exception {
|
||||
final BarrierMessageHandler handler = new BarrierMessageHandler(10000);
|
||||
QueueChannel outputChannel = new QueueChannel();
|
||||
handler.setOutputChannel(outputChannel);
|
||||
handler.setBeanFactory(mock(BeanFactory.class));
|
||||
handler.afterPropertiesSet();
|
||||
final AtomicReference<Exception> exception = new AtomicReference<Exception>();
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
Executors.newSingleThreadExecutor().execute(new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
handler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("foo").build());
|
||||
}
|
||||
catch (Exception e) {
|
||||
exception.set(e);
|
||||
latch.countDown();
|
||||
}
|
||||
}
|
||||
|
||||
});
|
||||
Map<?, ?> suspensions = TestUtils.getPropertyValue(handler, "suspensions", Map.class);
|
||||
int n = 0;
|
||||
while (n++ < 100 && suspensions.size() == 0) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
assertTrue("suspension did not appear in time", n < 100);
|
||||
Exception exc = new RuntimeException();
|
||||
handler.trigger(MessageBuilder.withPayload(exc).setCorrelationId("foo").build());
|
||||
assertTrue(latch.await(10, TimeUnit.SECONDS));
|
||||
assertSame(exc, exception.get().getCause());
|
||||
assertEquals(0, suspensions.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testJavaConfig() {
|
||||
Message<?> releasing = MessageBuilder.withPayload("bar").setCorrelationId("foo").build();
|
||||
this.release.send(releasing);
|
||||
Message<?> suspending = MessageBuilder.withPayload("foo").setCorrelationId("foo").build();
|
||||
this.in.send(suspending);
|
||||
Message<?> out = this.out.receive(10000);
|
||||
assertNotNull(out);
|
||||
assertEquals("[foo, bar]", out.getPayload().toString());
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableIntegration
|
||||
public static class Config {
|
||||
|
||||
@Bean
|
||||
public MessageChannel in() {
|
||||
return new DirectChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessageChannel out() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessageChannel release() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@ServiceActivator(inputChannel="in")
|
||||
@Bean
|
||||
public BarrierMessageHandler barrier() {
|
||||
BarrierMessageHandler barrier = new BarrierMessageHandler(10000);
|
||||
barrier.setOutputChannel(out());
|
||||
return barrier;
|
||||
}
|
||||
|
||||
@ServiceActivator (inputChannel="release", poller=@Poller(fixedDelay="0"))
|
||||
@Bean
|
||||
public MessageHandler releaser() {
|
||||
return new MessageHandler() {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
barrier().trigger(message);
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
87
src/reference/asciidoc/barrier.adoc
Normal file
87
src/reference/asciidoc/barrier.adoc
Normal file
@@ -0,0 +1,87 @@
|
||||
[[barrier]]
|
||||
=== Thread Barrier
|
||||
|
||||
Sometimes, we need to suspend a message flow thread until some other asynchronous event occurs.
|
||||
For example, consider an HTTP request that publishes a message to RabbitMQ.
|
||||
We might wish to not reply to the user until the RabbitMQ broker has issued an acknowledgment that the message was
|
||||
received.
|
||||
|
||||
Since _version 4.2_ Spring Integration introduced the `<barrier/>` component for this purpose.
|
||||
The underlying `MessageHandler` is the `BarrierMessageHandler`; this class also implements
|
||||
`MessageTriggerAction` where a message passed to the `trigger()` method releases a corresponding thread in the
|
||||
`handleRequestMessage()` method (if present).
|
||||
|
||||
The suspended thread and trigger thread are correlated by invoking a `CorrelationStrategy` on the messages.
|
||||
When a message is sent to the `input-channel`, the thread is suspended for up to `timeout` milliseconds, waiting for
|
||||
a corresponding trigger message.
|
||||
The default correlation strategy uses the `IntegrationMessageHeaderAccessor.CORRELATION_ID` header.
|
||||
When a trigger message arrives with the same correlation, the thread is released.
|
||||
The message sent to the `output-channel` after release is constructed using a `MessageGroupProcessor`.
|
||||
By default, the message is a `Collection<?>` of the two payloads and the headers are merged, using a
|
||||
`DefaultAggregatingMessageGroupProcessor`.
|
||||
|
||||
CAUTION: If the `trigger()` method is invoked first (or after the main thread times out), it will be suspended
|
||||
for up to `timeout` waiting for the suspending message to arrive.
|
||||
If you do not want to suspend the trigger thread, consider handing off to a `TaskExecutor` instead so its thread
|
||||
will be suspended instead.
|
||||
|
||||
The `requires-reply` property determines the action if the suspended thread times out before the trigger message
|
||||
arrives.
|
||||
By default, it is `false` which means the endpoint simply returns `null`, the flow ends and the thread returns to the
|
||||
caller.
|
||||
When `true`, a `ReplyRequiredException` is thrown.
|
||||
|
||||
You can call the `trigger()` method programmatically (obtain the bean reference using the name `barrier.handler`
|
||||
- where _barrier_ is the bean name of the barrier endpoint) or you can configure
|
||||
an `<outbound-channel-adapter/>` to trigger the release.
|
||||
|
||||
IMPORTANT: Only one thread can be suspended with the same correlation; the same correlation can be used multiple times
|
||||
but only once concurrently.
|
||||
|
||||
[source, xml]
|
||||
----
|
||||
<int:barrier id="barrier1" input-channel="in" output-channel="out"
|
||||
correlation-strategy-expression="headers['myHeader']"
|
||||
output-processor="myOutputProcessor"
|
||||
timeout="10000">
|
||||
</int:barrier>
|
||||
|
||||
<int:outbound-channel-adapter channel="release" ref="barrier1.handler" method="trigger" />
|
||||
----
|
||||
|
||||
In this example, a custom header is used for correlation.
|
||||
Either the thread sending a message to `in` or the one sending a message to `release` will wait for
|
||||
up to 10 seconds until the other arrives.
|
||||
When the message is released, the `out` channel will be sent a message combining the result of invoking the
|
||||
custom `MessageGroupProcessor` bean `myOutputProcessor`.
|
||||
Java configuration is shown below.
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Configuration
|
||||
@EnableIntegration
|
||||
public class Config {
|
||||
|
||||
@ServiceActivator(inputChannel="in")
|
||||
@Bean
|
||||
public BarrierMessageHandler barrier() {
|
||||
BarrierMessageHandler barrier = new BarrierMessageHandler(10000);
|
||||
barrier.setOutputChannel(out());
|
||||
return barrier;
|
||||
}
|
||||
|
||||
@ServiceActivator (inputChannel="release")
|
||||
@Bean
|
||||
public MessageHandler releaser() {
|
||||
return new MessageHandler() {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
barrier().trigger(message);
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
----
|
||||
@@ -15,4 +15,6 @@ include::./resequencer.adoc[]
|
||||
include::./chain.adoc[]
|
||||
|
||||
include::./scatter-gather.adoc[]
|
||||
|
||||
include::./barrier.adoc[]
|
||||
// BE SURE TO PRECEDE ALL include:: with a blank line - see https://github.com/asciidoctor/asciidoctor/issues/1297
|
||||
|
||||
@@ -54,6 +54,14 @@ Zookeeper support has been added to the framework to assist when running on a cl
|
||||
|
||||
See <<zookeeper>> for more information.
|
||||
|
||||
[[x4.2-barrier]]
|
||||
==== Thread Barrier
|
||||
|
||||
A new thread `<int:barrier/>` component is available allowing a thread to be suspended until some asynchronous event
|
||||
occurs.
|
||||
|
||||
See <<barrier>> for more information.
|
||||
|
||||
[[x4.2-general]]
|
||||
=== General Changes
|
||||
|
||||
|
||||
Reference in New Issue
Block a user