From 5a1846cfe502989f043ba353372ebcb30940cabe Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 2 Jul 2019 17:11:53 -0400 Subject: [PATCH] INT-2480: Add aggregate headers strategy JIRA: https://jira.spring.io/browse/INT-2480 * Introduce `headers-function` option into the `aggregator` for merging and computing headers for the output message based on the completed group * Implement a `DefaultAggregateHeadersFunction` and use it in the `AbstractAggregatingMessageGroupProcessor` for default behavior with possible injection for any other implementation * Add `DelegatingMessageGroupProcessor` to wrap any other `MessageGroupProcessor` implementations with possible usage of the `headersFunction` if result is not a `Message` or `MessageBuilder` * Make `AbstractCorrelatingMessageHandler.getOutputProcessor()` as `public` rto give access to this option from the `AggregatorSpec` to be able to inject a `headersFunction` in Java DSL configuration * Add `AbstractIntegrationMessageBuilder.getHeader()` to get access to some underlying header avoiding extra `Map` in case of `getHeaders()` * Change a logic in the `AbstractMessageProducingHandler.produceOutput()` to consult a `reply` for the `replyChannel` as well `routingSlip` header if the `reply` is a `Message` or `MessageBuilder` * Introduce a `AbstractMessageProducingHandler.messageBuilderForReply()` and use it in `AbstractMessageSplitter` to avoid duplication * Validate a new functionality in tests * Fix `FileOutboundGatewayParserTests` to rely on the `TemporaryFolder` to clean up test files after using * JavaDocs for `DefaultAggregateHeadersFunction` * Some `router.adoc` polishing * Fix link to Reactor in the `router.adoc` * Add docs for new `Function>` strategy * Doc polishing. --- ...tractAggregatingMessageGroupProcessor.java | 61 +++------ .../AbstractCorrelatingMessageHandler.java | 13 +- .../DefaultAggregateHeadersFunction.java | 88 +++++++++++++ .../DelegatingMessageGroupProcessor.java | 121 ++++++++++++++++++ .../ResequencingMessageGroupProcessor.java | 14 +- .../config/AggregatorFactoryBean.java | 61 ++++++--- .../config/xml/AggregatorParser.java | 23 +++- .../integration/dsl/AggregatorSpec.java | 45 ++++++- .../AbstractMessageProducingHandler.java | 63 ++++++--- ...ionEvaluatingRoutingSlipRouteStrategy.java | 4 +- .../splitter/AbstractMessageSplitter.java | 11 +- .../AbstractIntegrationMessageBuilder.java | 3 + .../integration/support/MessageBuilder.java | 27 ++-- .../support/MutableMessageBuilder.java | 20 ++- .../config/spring-integration-5.2.xsd | 13 ++ .../AggregatorIntegrationTests-context.xml | 7 +- .../AggregatorIntegrationTests.java | 23 +++- .../xml/HeaderEnricherTests-context.xml | 2 +- .../config/xml/HeaderEnricherTests.java | 7 +- .../correlation/CorrelationHandlerTests.java | 19 ++- ...FileOutboundGatewayParserTests-context.xml | 23 ++-- .../FileOutboundGatewayParserTests.java | 86 ++++--------- src/reference/asciidoc/aggregator.adoc | 11 ++ src/reference/asciidoc/router.adoc | 7 +- src/reference/asciidoc/whats-new.adoc | 3 + 25 files changed, 536 insertions(+), 219 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregateHeadersFunction.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aggregator/DelegatingMessageGroupProcessor.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index d0e88bddc0..2e3ad25b6f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -16,12 +16,8 @@ package org.springframework.integration.aggregator; -import java.util.HashMap; -import java.util.HashSet; import java.util.Map; -import java.util.Map.Entry; -import java.util.Objects; -import java.util.Set; +import java.util.function.Function; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -29,14 +25,12 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; /** @@ -56,6 +50,8 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR - final + private Function> headersFunction = new DefaultAggregateHeadersFunction(); + private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); private boolean messageBuilderFactorySet; @@ -67,6 +63,20 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag this.beanFactory = beanFactory; } + /** + * Specify a {@link Function} to map {@link MessageGroup} into composed headers for output message. + * @param headersFunction the {@link Function} to use. + * @since 5.2 + */ + public void setHeadersFunction(Function> headersFunction) { + Assert.notNull(headersFunction, "'headersFunction' must not be null"); + this.headersFunction = headersFunction; + } + + protected Function> getHeadersFunction() { + return this.headersFunction; + } + protected MessageBuilderFactory getMessageBuilderFactory() { if (!this.messageBuilderFactorySet) { if (this.beanFactory != null) { @@ -81,7 +91,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag public final Object processMessageGroup(MessageGroup group) { Assert.notNull(group, "MessageGroup must not be null"); Map headers = aggregateHeaders(group); - Object payload = this.aggregatePayloads(group, headers); + Object payload = aggregatePayloads(group, headers); AbstractIntegrationMessageBuilder builder; if (payload instanceof Message) { builder = getMessageBuilderFactory().fromMessage((Message) payload); @@ -104,40 +114,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag * @return The aggregated headers. */ protected Map aggregateHeaders(MessageGroup group) { - Map aggregatedHeaders = new HashMap<>(); - Set conflictKeys = doAggregateHeaders(group, aggregatedHeaders); - for (String keyToRemove : conflictKeys) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) " - + "in MessageGroup with correlation key: " + group.getGroupId()); - } - aggregatedHeaders.remove(keyToRemove); - } - return aggregatedHeaders; - } - - private Set doAggregateHeaders(MessageGroup group, Map aggregatedHeaders) { - Set conflictKeys = new HashSet<>(); - for (Message message : group.getMessages()) { - for (Entry entry : message.getHeaders().entrySet()) { - String key = entry.getKey(); - if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key) - || IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(key) - || IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(key)) { - continue; - } - Object value = entry.getValue(); - if (!aggregatedHeaders.containsKey(key)) { - aggregatedHeaders.put(key, value); - } - else { - if (!Objects.equals(value, aggregatedHeaders.get(key))) { - conflictKeys.add(key); - } - } - } - } - return conflictKeys; + return getHeadersFunction().apply(group); } protected abstract Object aggregatePayloads(MessageGroup group, Map defaultHeaders); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 8726cef29c..f05e5fe07b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -217,6 +217,15 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.outputProcessor = outputProcessor; } + /** + * Return a configured {@link MessageGroupProcessor}. + * @return the configured {@link MessageGroupProcessor} + * @since 5.2 + */ + public MessageGroupProcessor getOutputProcessor() { + return this.outputProcessor; + } + public void setDiscardChannel(MessageChannel discardChannel) { Assert.notNull(discardChannel, "'discardChannel' cannot be null"); this.discardChannel = discardChannel; @@ -372,10 +381,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP return this.expireGroupScheduledFutures; } - protected MessageGroupProcessor getOutputProcessor() { - return this.outputProcessor; - } - protected CorrelationStrategy getCorrelationStrategy() { return this.correlationStrategy; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregateHeadersFunction.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregateHeadersFunction.java new file mode 100644 index 0000000000..5e4106e40f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregateHeadersFunction.java @@ -0,0 +1,88 @@ +/* + * Copyright 2019 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 + * + * https://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.aggregator; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.function.Function; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.store.MessageGroup; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; + +/** + * The {@link Function} implementation for a default headers merging in the aggregator + * component. It takes all the unique headers from all the messages in group and removes + * those which are conflicted: have different values from different messages. + * + * @author Artem Bilan + * + * @since 5.2 + * + * @see AbstractAggregatingMessageGroupProcessor + */ +public class DefaultAggregateHeadersFunction implements Function> { + + private static final Log LOGGER = LogFactory.getLog(DefaultAggregateHeadersFunction.class); + + @Override + public Map apply(MessageGroup messageGroup) { + Map aggregatedHeaders = new HashMap<>(); + Set conflictKeys = doAggregateHeaders(messageGroup, aggregatedHeaders); + for (String keyToRemove : conflictKeys) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) " + + "in MessageGroup with correlation key: " + messageGroup.getGroupId()); + } + aggregatedHeaders.remove(keyToRemove); + } + return aggregatedHeaders; + } + + private Set doAggregateHeaders(MessageGroup group, Map aggregatedHeaders) { + Set conflictKeys = new HashSet<>(); + for (Message message : group.getMessages()) { + for (Map.Entry entry : message.getHeaders().entrySet()) { + String key = entry.getKey(); + if (MessageHeaders.ID.equals(key) + || MessageHeaders.TIMESTAMP.equals(key) + || IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(key) + || IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(key)) { + continue; + } + Object value = entry.getValue(); + if (!aggregatedHeaders.containsKey(key)) { + aggregatedHeaders.put(key, value); + } + else { + if (!Objects.equals(value, aggregatedHeaders.get(key))) { + conflictKeys.add(key); + } + } + } + } + return conflictKeys; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/DelegatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/DelegatingMessageGroupProcessor.java new file mode 100644 index 0000000000..6674e0e6ac --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/DelegatingMessageGroupProcessor.java @@ -0,0 +1,121 @@ +/* + * Copyright 2019 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 + * + * https://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.aggregator; + +import java.util.Map; +import java.util.function.Function; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.context.Lifecycle; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.DefaultMessageBuilderFactory; +import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; + +/** + * The {@link MessageGroupProcessor} implementation with delegation to the provided {@code delegate} + * and optional aggregation for headers. + *

+ * Unlike {@link AbstractAggregatingMessageGroupProcessor} this processor checks a result + * of the {@code delegate} call and aggregates headers into the output only + * if the result is not a {@link Message} or {@link AbstractIntegrationMessageBuilder}. + *

+ * This processor is used internally for wrapping provided non-standard {@link MessageGroupProcessor} + * when a aggregate headers {@link Function} is provided. + * For POJO method invoking or SpEL expression evaluation it is recommended to use an + * {@link AbstractAggregatingMessageGroupProcessor} implementations. + * + * + * @author Artem Bilan + * + * @since 5.2 + */ +public class DelegatingMessageGroupProcessor implements MessageGroupProcessor, BeanFactoryAware, Lifecycle { + + private final MessageGroupProcessor delegate; + + private final Function> headersFunction; + + private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); + + private volatile boolean messageBuilderFactorySet; + + private BeanFactory beanFactory; + + public DelegatingMessageGroupProcessor(MessageGroupProcessor delegate, + Function> headersFunction) { + + Assert.notNull(delegate, "'delegate' must not be null"); + Assert.notNull(headersFunction, "'headersFunction' must not be null"); + this.delegate = delegate; + this.headersFunction = headersFunction; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + if (this.delegate instanceof BeanFactoryAware) { + ((BeanFactoryAware) this.delegate).setBeanFactory(beanFactory); + } + } + + @Override + public Object processMessageGroup(MessageGroup group) { + Object result = this.delegate.processMessageGroup(group); + if (!(result instanceof Message) && !(result instanceof AbstractIntegrationMessageBuilder)) { + result = getMessageBuilderFactory() + .withPayload(result) + .copyHeadersIfAbsent(this.headersFunction.apply(group)); + } + return result; + } + + private MessageBuilderFactory getMessageBuilderFactory() { + if (!this.messageBuilderFactorySet) { + if (this.beanFactory != null) { + this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory); + } + this.messageBuilderFactorySet = true; + } + return this.messageBuilderFactory; + } + + @Override + public void start() { + if (this.delegate instanceof Lifecycle) { + ((Lifecycle) this.delegate).start(); + } + } + + @Override + public void stop() { + if (this.delegate instanceof Lifecycle) { + ((Lifecycle) this.delegate).stop(); + } + } + + @Override + public boolean isRunning() { + return this.delegate instanceof Lifecycle && ((Lifecycle) this.delegate).isRunning(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java index d511dd4a21..a926d9319c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java @@ -18,11 +18,10 @@ package org.springframework.integration.aggregator; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.Comparator; import java.util.List; -import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.store.MessageGroup; import org.springframework.messaging.Message; @@ -32,6 +31,8 @@ import org.springframework.messaging.Message; * @author Iwein Fuld * @author Dave Syer * @author Oleg Zhurakousky + * @author Artem Bilan + * * @since 2.0 */ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor { @@ -42,9 +43,9 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor Collection> messages = group.getMessages(); if (messages.size() > 0) { - List> sorted = new ArrayList>(messages); - Collections.sort(sorted, this.comparator); - ArrayList> partialSequence = new ArrayList>(); + List> sorted = new ArrayList<>(messages); + sorted.sort(this.comparator); + ArrayList> partialSequence = new ArrayList<>(); int previousSequence = extractSequenceNumber(sorted.get(0)); int currentSequence = previousSequence; for (Message message : sorted) { @@ -63,6 +64,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor } private Integer extractSequenceNumber(Message message) { - return new IntegrationMessageHeaderAccessor(message).getSequenceNumber(); + return StaticMessageHeaderAccessor.getSequenceNumber(message); } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index 43fb1fcdbb..315e4c9513 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -17,15 +17,20 @@ package org.springframework.integration.config; import java.util.List; +import java.util.Map; +import java.util.function.Function; import org.aopalliance.aop.Advice; import org.springframework.expression.Expression; +import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.DelegatingMessageGroupProcessor; import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.support.management.AbstractMessageHandlerMetrics; @@ -92,6 +97,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe private Boolean releaseLockBeforeSend; + private Function> headersFunction; + public void setProcessorBean(Object processorBean) { this.processorBean = processorBean; } @@ -181,6 +188,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe this.releaseLockBeforeSend = releaseLockBeforeSend; } + public void setHeadersFunction(Function> headersFunction) { + this.headersFunction = headersFunction; + } + @Override protected AggregatingMessageHandler createHandler() { MessageGroupProcessor outputProcessor; @@ -195,28 +206,38 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe outputProcessor = new MethodInvokingMessageGroupProcessor(this.processorBean, this.methodName); } } + + if (this.headersFunction != null) { + if (outputProcessor instanceof AbstractAggregatingMessageGroupProcessor) { + ((AbstractAggregatingMessageGroupProcessor) outputProcessor).setHeadersFunction(this.headersFunction); + } + else { + outputProcessor = new DelegatingMessageGroupProcessor(outputProcessor, this.headersFunction); + } + } + AggregatingMessageHandler aggregator = new AggregatingMessageHandler(outputProcessor); JavaUtils.INSTANCE - .acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion) - .acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout) - .acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName) - .acceptIfNotNull(this.metrics, aggregator::configureMetrics) - .acceptIfNotNull(this.statsEnabled, aggregator::setStatsEnabled) - .acceptIfNotNull(this.countsEnabled, aggregator::setCountsEnabled) - .acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry) - .acceptIfNotNull(this.messageStore, aggregator::setMessageStore) - .acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy) - .acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy) - .acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression) - .acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain) - .acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler) - .acceptIfNotNull(this.discardChannel, aggregator::setDiscardChannel) - .acceptIfNotNull(this.discardChannelName, aggregator::setDiscardChannelName) - .acceptIfNotNull(this.sendPartialResultOnExpiry, aggregator::setSendPartialResultOnExpiry) - .acceptIfNotNull(this.minimumTimeoutForEmptyGroups, aggregator::setMinimumTimeoutForEmptyGroups) - .acceptIfNotNull(this.expireGroupsUponTimeout, aggregator::setExpireGroupsUponTimeout) - .acceptIfNotNull(this.popSequence, aggregator::setPopSequence) - .acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend); + .acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion) + .acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout) + .acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName) + .acceptIfNotNull(this.metrics, aggregator::configureMetrics) + .acceptIfNotNull(this.statsEnabled, aggregator::setStatsEnabled) + .acceptIfNotNull(this.countsEnabled, aggregator::setCountsEnabled) + .acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry) + .acceptIfNotNull(this.messageStore, aggregator::setMessageStore) + .acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy) + .acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy) + .acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression) + .acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain) + .acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler) + .acceptIfNotNull(this.discardChannel, aggregator::setDiscardChannel) + .acceptIfNotNull(this.discardChannelName, aggregator::setDiscardChannelName) + .acceptIfNotNull(this.sendPartialResultOnExpiry, aggregator::setSendPartialResultOnExpiry) + .acceptIfNotNull(this.minimumTimeoutForEmptyGroups, aggregator::setMinimumTimeoutForEmptyGroups) + .acceptIfNotNull(this.expireGroupsUponTimeout, aggregator::setExpireGroupsUponTimeout) + .acceptIfNotNull(this.popSequence, aggregator::setPopSequence) + .acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend); return aggregator; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index 94823b4773..75b3bb9553 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -38,6 +38,7 @@ import org.springframework.util.StringUtils; * @author Dave Syer * @author Stefan Ferstl * @author Gary Russell + * @author Artem Bilan */ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { @@ -49,6 +50,7 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { parserContext); String ref = element.getAttribute(REF_ATTRIBUTE); BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatorFactoryBean.class); + String headersFunction = element.getAttribute("headers-function"); BeanMetadataElement processor = null; if (innerHandlerDefinition != null || StringUtils.hasText(ref)) { @@ -59,18 +61,25 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { processor = new RuntimeBeanReference(ref); } builder.addPropertyValue("processorBean", processor); + if (StringUtils.hasText(headersFunction)) { + builder.addPropertyReference("headersFunction", headersFunction); + } } else { + BeanDefinitionBuilder groupProcessorBuilder; if (StringUtils.hasText(element.getAttribute(EXPRESSION_ATTRIBUTE))) { String expression = element.getAttribute(EXPRESSION_ATTRIBUTE); - BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder - .genericBeanDefinition(ExpressionEvaluatingMessageGroupProcessor.class); - adapterBuilder.addConstructorArgValue(expression); - builder.addPropertyValue("processorBean", adapterBuilder.getBeanDefinition()); + groupProcessorBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingMessageGroupProcessor.class); + groupProcessorBuilder.addConstructorArgValue(expression); } else { - builder.addPropertyValue("processorBean", BeanDefinitionBuilder - .genericBeanDefinition(DefaultAggregatingMessageGroupProcessor.class).getBeanDefinition()); + groupProcessorBuilder = + BeanDefinitionBuilder.genericBeanDefinition(DefaultAggregatingMessageGroupProcessor.class); + } + builder.addPropertyValue("processorBean", groupProcessorBuilder.getBeanDefinition()); + if (StringUtils.hasText(headersFunction)) { + groupProcessorBuilder.addPropertyReference("headersFunction", headersFunction); } } @@ -79,7 +88,7 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { builder.addPropertyValue("methodName", method); } - this.doParse(builder, element, processor, parserContext); + doParse(builder, element, processor, parserContext); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXPIRE_GROUPS_UPON_COMPLETION); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java index 1d92474ad4..f50ccb191c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java @@ -16,11 +16,17 @@ package org.springframework.integration.dsl; +import java.util.Map; +import java.util.function.Function; + +import org.springframework.integration.aggregator.AbstractAggregatingMessageGroupProcessor; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; +import org.springframework.integration.aggregator.DelegatingMessageGroupProcessor; import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor; import org.springframework.integration.aggregator.MessageGroupProcessor; import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; +import org.springframework.integration.store.MessageGroup; /** * A {@link CorrelationHandlerSpec} for an {@link AggregatingMessageHandler}. @@ -31,6 +37,8 @@ import org.springframework.integration.aggregator.MethodInvokingMessageGroupProc */ public class AggregatorSpec extends CorrelationHandlerSpec { + private Function> headersFunction; + AggregatorSpec() { super(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor())); } @@ -59,9 +67,9 @@ public class AggregatorSpec extends CorrelationHandlerSpec> headersFunction) { + this.headersFunction = headersFunction; + return _this(); + } + + + @Override + public Map getComponentsToRegister() { + if (this.headersFunction != null) { + MessageGroupProcessor outputProcessor = this.handler.getOutputProcessor(); + if (outputProcessor instanceof AbstractAggregatingMessageGroupProcessor) { + ((AbstractAggregatingMessageGroupProcessor) outputProcessor).setHeadersFunction(this.headersFunction); + } + else { + this.handler.setOutputProcessor( + new DelegatingMessageGroupProcessor(outputProcessor, this.headersFunction)); + } + } + return super.getComponentsToRegister(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 60611633f0..bb91257308 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -158,9 +158,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan this.notPropagatedHeaders = headerPatterns.toArray(new String[0]); } - boolean hasAsterisk = headerPatterns.contains("*"); - - if (hasAsterisk) { + if (headerPatterns.contains("*")) { this.notPropagatedHeaders = new String[] { "*" }; this.noHeadersPropagation = true; } @@ -240,12 +238,11 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } protected void produceOutput(Object replyArg, final Message requestMessage) { - final MessageHeaders requestHeaders = requestMessage.getHeaders(); - + MessageHeaders requestHeaders = requestMessage.getHeaders(); Object reply = replyArg; Object replyChannel = null; if (getOutputChannel() == null) { - Map routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); + Map routingSlipHeader = obtainRoutingSlipHeader(requestHeaders, reply); if (routingSlipHeader != null) { Assert.isTrue(routingSlipHeader.size() == 1, "The RoutingSlip header value must be a SingletonMap"); @@ -260,18 +257,45 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan reply = addRoutingSlipHeader(reply, routingSlip, routingSlipIndex); } } - if (replyChannel == null) { - replyChannel = requestHeaders.getReplyChannel(); - if (replyChannel == null && reply instanceof Message) { - replyChannel = ((Message) reply).getHeaders().getReplyChannel(); - } + replyChannel = obtainReplyChannel(requestHeaders, reply); } } doProduceOutput(requestMessage, requestHeaders, reply, replyChannel); } - private void doProduceOutput(final Message requestMessage, final MessageHeaders requestHeaders, Object reply, + @Nullable + private Map obtainRoutingSlipHeader(MessageHeaders requestHeaders, Object reply) { + Map routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); + if (routingSlipHeader == null) { + if (reply instanceof Message) { + routingSlipHeader = ((Message) reply).getHeaders() + .get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); + } + else if (reply instanceof AbstractIntegrationMessageBuilder) { + routingSlipHeader = ((AbstractIntegrationMessageBuilder) reply) + .getHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); + } + } + return routingSlipHeader; + } + + @Nullable + private Object obtainReplyChannel(MessageHeaders requestHeaders, Object reply) { + Object replyChannel = requestHeaders.getReplyChannel(); + if (replyChannel == null) { + if (reply instanceof Message) { + replyChannel = ((Message) reply).getHeaders().getReplyChannel(); + } + else if (reply instanceof AbstractIntegrationMessageBuilder) { + replyChannel = ((AbstractIntegrationMessageBuilder) reply) + .getHeader(MessageHeaders.REPLY_CHANNEL, Object.class); + } + } + return replyChannel; + } + + private void doProduceOutput(Message requestMessage, MessageHeaders requestHeaders, Object reply, Object replyChannel) { if (this.async && (reply instanceof ListenableFuture || reply instanceof Publisher)) { @@ -296,19 +320,22 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan private AbstractIntegrationMessageBuilder addRoutingSlipHeader(Object reply, List routingSlip, AtomicInteger routingSlipIndex) { - //TODO Migrate to the SF MessageBuilder - AbstractIntegrationMessageBuilder builder = null; + return messageBuilderForReply(reply) + .setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP, + Collections.singletonMap(routingSlip, routingSlipIndex.get())); + } + + protected AbstractIntegrationMessageBuilder messageBuilderForReply(Object reply) { + AbstractIntegrationMessageBuilder builder; if (reply instanceof Message) { - builder = this.getMessageBuilderFactory().fromMessage((Message) reply); + builder = getMessageBuilderFactory().fromMessage((Message) reply); } else if (reply instanceof AbstractIntegrationMessageBuilder) { builder = (AbstractIntegrationMessageBuilder) reply; } else { - builder = this.getMessageBuilderFactory().withPayload(reply); + builder = getMessageBuilderFactory().withPayload(reply); } - builder.setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP, - Collections.singletonMap(routingSlip, routingSlipIndex.get())); return builder; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java index 951cd0819e..fe709ced7b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java @@ -59,6 +59,7 @@ import org.springframework.messaging.Message; * * @author Artem Bilan * @author Gary Russell + * * @since 4.1 */ public class ExpressionEvaluatingRoutingSlipRouteStrategy @@ -98,8 +99,7 @@ public class ExpressionEvaluatingRoutingSlipRouteStrategy @Override public Object getNextPath(Message requestMessage, Object reply) { - return this.expression.getValue(this.evaluationContext, new RequestAndReply(requestMessage, reply), - String.class); + return this.expression.getValue(this.evaluationContext, new RequestAndReply(requestMessage, reply)); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index 740fbc219f..eb4f25532d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -263,16 +263,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess private AbstractIntegrationMessageBuilder createBuilder(Object item, Map headers, Object correlationId, int sequenceNumber, int sequenceSize) { - AbstractIntegrationMessageBuilder builder; - if (item instanceof Message) { - builder = getMessageBuilderFactory().fromMessage((Message) item); - } - else if (item instanceof AbstractIntegrationMessageBuilder) { - builder = (AbstractIntegrationMessageBuilder) item; - } - else { - builder = getMessageBuilderFactory().withPayload(item); - } + AbstractIntegrationMessageBuilder builder = messageBuilderForReply(item); builder.copyHeadersIfAbsent(headers); if (this.applySequence) { builder.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java index 29f9efb939..7bb1b0e72b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java @@ -176,6 +176,9 @@ public abstract class AbstractIntegrationMessageBuilder { public abstract Map getHeaders(); + @Nullable + public abstract V getHeader(String key, Class type); + /** * Set the value for the given header name. If the provided value is null, the header will be removed. * @param headerName The header name. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java index beb156289b..baa21d0ff1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java @@ -64,7 +64,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder originalMessage) { + private MessageBuilder(T payload, @Nullable Message originalMessage) { Assert.notNull(payload, "payload must not be null"); this.payload = payload; this.originalMessage = originalMessage; @@ -84,6 +84,12 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder V getHeader(String key, Class type) { + return this.headerAccessor.getHeader(key, type); + } + /** * Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the * provided message. The payload of the provided Message will also be used as the payload for the new message. @@ -94,23 +100,21 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder MessageBuilder fromMessage(Message message) { Assert.notNull(message, "message must not be null"); - return new MessageBuilder(message.getPayload(), message); + return new MessageBuilder<>(message.getPayload(), message); } /** * Create a builder for a new {@link Message} instance with the provided payload. - * * @param payload the payload for the new message * @param The type of the payload. * @return A MessageBuilder. */ public static MessageBuilder withPayload(T payload) { - return new MessageBuilder(payload, null); + return new MessageBuilder<>(payload, null); } /** * Set the value for the given header name. If the provided value is null, the header will be removed. - * * @param headerName The header name. * @param headerValue The header value. * @return this MessageBuilder. @@ -123,7 +127,6 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder extends AbstractIntegrationMessageBuilder extends AbstractIntegrationMessageBuilder extends AbstractIntegrationMessageBuildernot overwrite any existing values. - * * @param headersToCopy The headers to copy. * @return this MessageBuilder. */ @@ -225,8 +224,8 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder pushSequenceDetails(Object correlationId, int sequenceNumber, int sequenceSize) { @@ -320,12 +319,13 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder build() { if (!this.modified && !this.headerAccessor.isModified() && this.originalMessage != null && !containsReadOnly(this.originalMessage.getHeaders())) { + return this.originalMessage; } if (this.payload instanceof Throwable) { return (Message) new ErrorMessage((Throwable) this.payload, this.headerAccessor.toMap()); } - return new GenericMessage(this.payload, this.headerAccessor.toMap()); + return new GenericMessage<>(this.payload, this.headerAccessor.toMap()); } private boolean containsReadOnly(MessageHeaders headers) { @@ -339,5 +339,4 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder extends AbstractIntegrationMessageBu return this.headers; } + @SuppressWarnings("unchecked") + @Nullable + @Override + public V getHeader(String key, Class type) { + Object value = this.headers.get(key); + if (value == null) { + return null; + } + if (!type.isAssignableFrom(value.getClass())) { + throw new IllegalArgumentException("Incorrect type specified for header '" + key + "'. Expected [" + type + + "] but actual type is [" + value.getClass() + "]"); + } + return (V) value; + } + /** * Create a builder for a new {@link Message} instance with the provided payload. * @param payload the payload for the new message @@ -143,7 +159,7 @@ public final class MutableMessageBuilder extends AbstractIntegrationMessageBu @Override public AbstractIntegrationMessageBuilder removeHeaders(String... headerPatterns) { - List headersToRemove = new ArrayList(); + List headersToRemove = new ArrayList<>(); for (String pattern : headerPatterns) { if (StringUtils.hasLength(pattern)) { if (pattern.contains("*")) { @@ -161,7 +177,7 @@ public final class MutableMessageBuilder extends AbstractIntegrationMessageBu } private List getMatchingHeaderNames(String pattern, Map headers) { - List matchingHeaderNames = new ArrayList(); + List matchingHeaderNames = new ArrayList<>(); if (headers != null) { for (Map.Entry header : headers.entrySet()) { if (PatternMatchUtils.simpleMatch(pattern, header.getKey())) { diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd index 62b2005384..6d28a8a53e 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.2.xsd @@ -3724,6 +3724,19 @@ + + + + + + + + + A reference to the 'Function' for merging and computing message headers for reply + based on the 'MessageGroup' to release. + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml index 3d91846303..7f17173b68 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml @@ -11,7 +11,12 @@ - + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index 64d8596e3f..8ba0891c87 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Queue; +import java.util.function.Function; import org.junit.Test; import org.junit.runner.RunWith; @@ -32,6 +33,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -92,6 +94,7 @@ public class AggregatorIntegrationTests { Message receive = output.receive(10000); assertThat(receive).isNotNull(); assertThat(receive.getPayload()).isEqualTo(1 + 2 + 3 + 4); + assertThat(receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)).isEqualTo(0); } @Test @@ -245,6 +248,18 @@ public class AggregatorIntegrationTests { } // configured in context associated with this test + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) { + Map headers = new HashMap<>(); + headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); + headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); + headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); + return headers; + } + + public static Function> firstMessageHeaders() { + return (messageGroup) -> messageGroup.getOne().getHeaders(); + } + public static class SummingAggregator { public Integer sum(List numbers) { int result = 0; @@ -255,12 +270,6 @@ public class AggregatorIntegrationTests { } } - private Map stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) { - Map headers = new HashMap<>(); - headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); - headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); - headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); - return headers; - } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml index 24531e1091..4c780eddff 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml @@ -83,7 +83,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java index b5bd9f45ab..62c17a55cc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java @@ -270,9 +270,10 @@ public class HeaderEnricherTests { @SuppressWarnings("unchecked") List routingSlipPath = (List) ((Map) routingSlip).keySet().iterator().next(); - assertThat(routingSlipPath.get(0)).isEqualTo("fooChannel"); - assertThat(routingSlipPath.get(1)).isInstanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class); - assertThat(routingSlipPath.get(2)).isEqualTo("bazRoutingSlip"); + assertThat(routingSlipPath.get(0)).isInstanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class); + assertThat(routingSlipPath.get(1)).isEqualTo("fooChannel"); + assertThat(routingSlipPath.get(2)).isInstanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class); + assertThat(routingSlipPath.get(3)).isEqualTo("bazRoutingSlip"); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java index ecedf33cf8..f23bb38c7d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java @@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -141,9 +142,10 @@ public class CorrelationHandlerTests { public void testSubscriberAggregateFlow() { this.subscriberAggregateFlowInput.send(new GenericMessage<>("test")); - Message receive1 = this.subscriberAggregateResult.receive(10000); - assertThat(receive1).isNotNull(); - assertThat(receive1.getPayload()).isEqualTo("Hello World!"); + Message receive = this.subscriberAggregateResult.receive(10000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("Hello World!"); + assertThat(receive.getHeaders().get("foo")).isEqualTo("bar"); } @@ -274,10 +276,13 @@ public class CorrelationHandlerTests { @Bean public IntegrationFlow publishSubscribeAggregateFlow() { return flow -> flow - .aggregate(a -> a.outputProcessor(g -> g.getMessages() - .stream() - .map(m -> (String) m.getPayload()) - .collect(Collectors.joining(" ")))) + .aggregate(a -> a + .outputProcessor((group) -> group + .getMessages() + .stream() + .map(m -> (String) m.getPayload()) + .collect(Collectors.joining(" "))) + .headersFunction((group) -> Collections.singletonMap("foo", "bar"))) .channel(MessageChannels.queue("subscriberAggregateResult")); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests-context.xml index 92008414f5..28b32100cc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests-context.xml @@ -3,21 +3,26 @@ xmlns:int-file="http://www.springframework.org/schema/integration/file" xmlns:beans="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" + xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/integration/file - https://www.springframework.org/schema/integration/file/spring-integration-file.xsd"> + https://www.springframework.org/schema/integration/file/spring-integration-file.xsd + http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd"> + + @@ -28,32 +33,32 @@ + directory="#{temporaryFolder.root}" requires-reply="false"/> + directory="#{temporaryFolder.root}"/> + directory="#{temporaryFolder.root}"/> + directory="#{temporaryFolder.root}"/> + directory="#{temporaryFolder.root}"/> + directory="#{temporaryFolder.root}"/> diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java index d814e6d926..192e3e8a01 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileOutboundGatewayParserTests.java @@ -17,11 +17,14 @@ package org.springframework.integration.file.config; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.io.File; +import org.junit.Before; +import org.junit.ClassRule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; @@ -39,8 +42,8 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.FileCopyUtils; /** @@ -49,10 +52,13 @@ import org.springframework.util.FileCopyUtils; * @author Artem Bilan * @author Tony Falabella */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@RunWith(SpringRunner.class) +@DirtiesContext public class FileOutboundGatewayParserTests { + @ClassRule + public static final TemporaryFolder tempFolder = new TemporaryFolder(); + @Autowired private EventDrivenConsumer ordered; @@ -83,6 +89,11 @@ public class FileOutboundGatewayParserTests { private static volatile int adviceCalled; + @Before + public void setup() { + tempFolder.delete(); + } + @Test public void checkOrderedGateway() { DirectFieldAccessor gatewayAccessor = new DirectFieldAccessor(ordered); @@ -109,7 +120,7 @@ public class FileOutboundGatewayParserTests { FileWritingMessageHandler handler = TestUtils.getPropertyValue(gatewayWithDirectoryExpression, "handler", FileWritingMessageHandler.class); assertThat(TestUtils.getPropertyValue(handler, "destinationDirectoryExpression", Expression.class) - .getExpressionString()).isEqualTo("'build/foo'"); + .getExpressionString()).isEqualTo("temporaryFolder.root"); handler.handleMessage(new GenericMessage<>("foo")); assertThat(adviceCalled).isEqualTo(1); } @@ -129,11 +140,7 @@ public class FileOutboundGatewayParserTests { messagingTemplate.setDefaultDestination(this.gatewayWithIgnoreModeChannel); final String expectedFileContent = "Initial File Content:"; - final File testFile = new File("test/fileToAppend.txt"); - - if (testFile.exists()) { - testFile.delete(); - } + final File testFile = new File(tempFolder.getRoot(), "fileToAppend.txt"); messagingTemplate.sendAndReceive(new GenericMessage<>("Initial File Content:")); @@ -165,29 +172,16 @@ public class FileOutboundGatewayParserTests { String expectedFileContent = "Initial File Content:"; - File testFile = new File("test/fileToAppend.txt"); - - if (testFile.exists()) { - testFile.delete(); - } + File testFile = new File(tempFolder.getRoot(), "fileToAppend.txt"); messagingTemplate.sendAndReceive(new GenericMessage<>("Initial File Content:")); final String actualFileContent = new String(FileCopyUtils.copyToByteArray(testFile)); assertThat(actualFileContent).isEqualTo(expectedFileContent); - try { - - messagingTemplate.sendAndReceive(new GenericMessage<>("String content:")); - - } - catch (MessageHandlingException e) { - assertThat(e.getMessage()).startsWith("The destination file already exists at '"); - return; - } - - fail("Was expecting a MessageHandlingException to be thrown."); - + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> messagingTemplate.sendAndReceive(new GenericMessage<>("String content:"))) + .withMessageStartingWith("The destination file already exists at '"); } /** @@ -200,35 +194,21 @@ public class FileOutboundGatewayParserTests { */ @Test public void gatewayWithFailModeLowercase() throws Exception { - final MessagingTemplate messagingTemplate = new MessagingTemplate(); messagingTemplate.setDefaultDestination(this.gatewayWithFailModeLowercaseChannel); String expectedFileContent = "Initial File Content:"; - File testFile = new File("test/fileToAppend.txt"); - - if (testFile.exists()) { - testFile.delete(); - } + File testFile = new File(tempFolder.getRoot(), "fileToAppend.txt"); messagingTemplate.sendAndReceive(new GenericMessage<>("Initial File Content:")); final String actualFileContent = new String(FileCopyUtils.copyToByteArray(testFile)); assertThat(actualFileContent).isEqualTo(expectedFileContent); - try { - - messagingTemplate.sendAndReceive(new GenericMessage<>("String content:")); - - } - catch (MessageHandlingException e) { - assertThat(e.getMessage()).startsWith("The destination file already exists at '"); - return; - } - - fail("Was expecting a MessageHandlingException to be thrown."); - + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> messagingTemplate.sendAndReceive(new GenericMessage<>("String content:"))) + .withMessageStartingWith("The destination file already exists at '"); } /** @@ -243,17 +223,12 @@ public class FileOutboundGatewayParserTests { */ @Test public void gatewayWithAppendMode() throws Exception { - final MessagingTemplate messagingTemplate = new MessagingTemplate(); messagingTemplate.setDefaultDestination(this.gatewayWithAppendModeChannel); String expectedFileContent = "Initial File Content:String content:"; - File testFile = new File("test/fileToAppend.txt"); - - if (testFile.exists()) { - testFile.delete(); - } + File testFile = new File(tempFolder.getRoot(), "fileToAppend.txt"); messagingTemplate.sendAndReceive(new GenericMessage<>("Initial File Content:")); Message m = messagingTemplate.sendAndReceive(new GenericMessage<>("String content:")); @@ -280,7 +255,6 @@ public class FileOutboundGatewayParserTests { */ @Test public void gatewayWithReplaceMode() throws Exception { - assertThat(TestUtils.getPropertyValue(this.gatewayWithReplaceModeHandler, "requiresReply", Boolean.class)) .isFalse(); @@ -289,11 +263,7 @@ public class FileOutboundGatewayParserTests { String expectedFileContent = "String content:"; - File testFile = new File("test/fileToAppend.txt"); - - if (testFile.exists()) { - testFile.delete(); - } + File testFile = new File(tempFolder.getRoot(), "fileToAppend.txt"); messagingTemplate.sendAndReceive(new GenericMessage<>("Initial File Content:")); Message m = messagingTemplate.sendAndReceive(new GenericMessage<>("String content:")); diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 41fdf334b4..da30a43459 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -67,6 +67,17 @@ public abstract class AbstractAggregatingMessageGroupProcessor ---- ==== +See `DefaultAggregatingMessageGroupProcessor`, `ExpressionEvaluatingMessageGroupProcessor` and `MethodInvokingMessageGroupProcessor` as out-of-the-box implementations of the `AbstractAggregatingMessageGroupProcessor`. + +Starting with version 5.2, a `Function>` strategy is available for the `AbstractAggregatingMessageGroupProcessor` to merge and compute (aggregate) headers for an output message. +The `DefaultAggregateHeadersFunction` implementation is available with logic that returns all headers that have no conflicts among the group; an absent header on one or more messages within the group is not considered a conflict. +Conflicting headers are omitted. +Along with the newly introduced `DelegatingMessageGroupProcessor`, this function is used for any arbitrary (non-`AbstractAggregatingMessageGroupProcessor`) `MessageGroupProcessor` implementation. +Essentially, the framework injects a provided function into an `AbstractAggregatingMessageGroupProcessor` instance and wraps all other implementations into a `DelegatingMessageGroupProcessor`. +The difference in logic between the `AbstractAggregatingMessageGroupProcessor` and the `DelegatingMessageGroupProcessor` that the latter doesn't compute headers in advance, before calling the delegate strategy, and doesn't invoke the function if the delegate returns a `Message` or `AbstractIntegrationMessageBuilder`. +In that case, the framework assumes that the target implementation has taken care of producing a proper set of headers populated into the returned result. +The `Function>` strategy is available as the `headers-function` reference attribute for XML configuration, as the `AggregatorSpec.headersFunction()` option for the Java DSL and as `AggregatorFactoryBean.setHeadersFunction()` for plain Java configuration. + The `CorrelationStrategy` is owned by the `AbstractCorrelatingMessageHandler` and has a default value based on the `IntegrationMessageHeaderAccessor.CORRELATION_ID` message header, as the following example shows: ==== diff --git a/src/reference/asciidoc/router.adoc b/src/reference/asciidoc/router.adoc index 1032e56d70..cb6da26d09 100644 --- a/src/reference/asciidoc/router.adoc +++ b/src/reference/asciidoc/router.adoc @@ -1218,8 +1218,7 @@ NOTE: The `requestMessage` argument is always a `Message`. Depending on context, the reply object may be a `Message`, an `AbstractIntegrationMessageBuilder`, or an arbitrary application domain object (when, for example, it is returned by a POJO method invoked by a service activator). In the first two cases, the usual `Message` properties (`payload` and `headers`) are available when using SpEL (or a Java implementation). For an arbitrary domain object, these properties are not available. -For this reason, be careful when you use routing slips in conjunction with POJO methods if the result is used to determine the -next path. +For this reason, be careful when you use routing slips in conjunction with POJO methods if the result is used to determine the next path. IMPORTANT: If a routing slip is involved in a distributed environment, we recommend not using inline expressions for the Routing Slip `path`. This recommendation applies to distributed environments such as cross-JVM applications, using a `request-reply` through a message broker (such as<<./amqp.adoc#amqp,AMQP Support>> or <<./jms.adoc#jms,JMS Support>>), or using a persistent `MessageStore` (<<./message-store.adoc#message-store,Message Store>>) in the integration flow. @@ -1252,7 +1251,7 @@ The routing slip algorithm works as follows when an endpoint produces a reply an * If a returned bean is an instance of `MessageChannel`, it is used as the next `outputChannel` and the `routingSlipIndex` is incremented in the reply message header (the routing slip `path` entries remain unchanged). * If a returned bean is an instance of `RoutingSlipRouteStrategy` and its `getNextPath` does not return an empty `String`, that result is used as a bean name for the next `outputChannel`. The `routingSlipIndex` remains unchanged. -* If `RoutingSlipRouteStrategy.getNextPath` returns an empty `String`, the `routingSlipIndex` is incremented and the `getOutputChannelFromRoutingSlip` is invoked recursively for the next Routing Slip `path` item. +* If `RoutingSlipRouteStrategy.getNextPath` returns an empty `String` or `null`, the `routingSlipIndex` is incremented and the `getOutputChannelFromRoutingSlip` is invoked recursively for the next Routing Slip `path` item. * If the next routing slip `path` entry is not a `String`, it must be an instance of `RoutingSlipRouteStrategy`. * When the `routingSlipIndex` exceeds the size of the routing slip `path` list, the algorithm moves to the default behavior for the standard `replyChannel` header. @@ -1265,7 +1264,7 @@ In addition to a bean name, the `RoutingSlipRouteStrategy` can return any `Messa This way, we can provide powerful dynamic routing logic when there is no way to predict which channel should be used. A `MessageChannel` can be created within the `RoutingSlipRouteStrategy` and returned. A `FixedSubscriberChannel` with an associated `MessageHandler` implementation is a good combination for such cases. -For example, you can route to a https://github.com/reactor/reactor/wiki/Streams[reactor stream], as the following example shows: +For example, you can route to a https://projectreactor.io/docs/core/release/reference/#getting-started[Reactive Streams], as the following example shows: ==== [source,java] diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d6fbb6e401..b72c051f8e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -56,6 +56,9 @@ See <<./splitter.adoc#splitter,Splitter>> for more information. The Control Bus can now handle `Pausable` (extension of `Lifecycle`) operations. See <<./control-bus.adoc#control-bus,Control Bus>> for more information. +The `Function>` strategy has been introduced for the aggregator component to merge and compute headers for output messages. +See <<./aggregator.adoc#aggregator-api,Aggregator Programming Model>> for more information. + [[x5.2-amqp]] ==== AMQP Changes