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 171dbad037..d069d28484 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 @@ -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(); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java new file mode 100644 index 0000000000..94e4756e58 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/BarrierParser.java @@ -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 }. + * + * @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; + } + +} + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java index 8b2a631c94..a84063e30c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java @@ -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()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java index 8578147998..f95bfc3c54 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java @@ -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(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/BarrierMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/BarrierMessageHandler.java new file mode 100644 index 0000000000..24d50eeaa5 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/BarrierMessageHandler.java @@ -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. + *

+ * The default {@link CorrelationStrategy} is a {@link HeaderAttributeCorrelationStrategy}. + *

+ * The default output processor is a {@link DefaultAggregatingMessageGroupProcessor}. + * + * @author Gary Russell + * + * @since 4.2 + */ +public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler implements MessageTriggerAction { + + private final ConcurrentMap>> suspensions = + new ConcurrentHashMap>>(); + + private final ConcurrentMap inProcess = new ConcurrentHashMap(); + + 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> 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> createOrObtainQueue(Object key) { + SynchronousQueue> syncQueue = new SynchronousQueue>(); + SynchronousQueue> 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> 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); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageTriggerAction.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageTriggerAction.java new file mode 100644 index 0000000000..eb1ef715c0 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageTriggerAction.java @@ -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); + +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd index 80157e160c..a4d7fae646 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd @@ -1659,6 +1659,91 @@ + + + + Defines an endpoint that suspends a thread until a corresponding release message is received. + The endpoint's message handler is a 'BarrierMessageHandler'. + + + + + + + + + + + + + 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. + + + + + + + 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. + + + + + + + + + + + + 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. + + + + + + + + + + + + A method defined on the bean referenced by correlation-strategy, that implements the + correlation decision algorithm + + + + + + + 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) + + + + + + + + + + + + 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. + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml new file mode 100644 index 0000000000..fd33ed8db9 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests-context.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java new file mode 100644 index 0000000000..8ee45f2364 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/BarrierParserTests.java @@ -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("foo")); + this.release.send(new GenericMessage("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; + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/BarrierMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/BarrierMessageHandlerTests.java new file mode 100644 index 0000000000..74dfca335f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/BarrierMessageHandlerTests.java @@ -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 dupCorrelation = new AtomicReference(); + 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 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 = new AtomicReference(); + 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); + } + + }; + } + + } + +} diff --git a/src/reference/asciidoc/barrier.adoc b/src/reference/asciidoc/barrier.adoc new file mode 100644 index 0000000000..3a6ad20380 --- /dev/null +++ b/src/reference/asciidoc/barrier.adoc @@ -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 `` 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 `` 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] +---- + + + + +---- + +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); + } + + }; + } + +} +---- diff --git a/src/reference/asciidoc/message-routing.adoc b/src/reference/asciidoc/message-routing.adoc index f4cbe22e9c..e776f958cd 100644 --- a/src/reference/asciidoc/message-routing.adoc +++ b/src/reference/asciidoc/message-routing.adoc @@ -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 diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 4e0c592594..85e853f9a7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -54,6 +54,14 @@ Zookeeper support has been added to the framework to assist when running on a cl See <> for more information. +[[x4.2-barrier]] +==== Thread Barrier + +A new thread `` component is available allowing a thread to be suspended until some asynchronous event +occurs. + +See <> for more information. + [[x4.2-general]] === General Changes