From bf8e79cb1e8049bb03eba55a42f74e38a36eeebd Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 26 Jun 2015 15:09:11 -0400 Subject: [PATCH] INT-3583: Allow MessageGroupProcessor in Agg, XML JIRA: https://jira.spring.io/browse/INT-3583 Previously, the `ref` or inner bean for an aggregator was wrapped in a `MethodInvokingMessageGroupProcessor` for POJO aggregation logic. Now, if the ref'd bean is a `MessageGroupProcessor`, it is used as the output processor directly. Also, the `SimpleMessageGroupProcessor` is added which simply returns the collection of messages. INT-3583: Polishing; PR Comments Fix AsciiDoc Revert `CorrelationMessageBarrier` deprecation Fix Docs for wrong chapter link --- .../PassThroughMessageGroupProcessor.java | 14 +- .../SimpleMessageGroupProcessor.java | 36 ++++ .../config/AggregatorFactoryBean.java | 175 ++++++++++++++++++ .../config/ReleaseStrategyFactoryBean.java | 6 +- .../config/xml/AggregatorParser.java | 13 +- .../CorrelatingMessageBarrierTests.java | 32 ++-- ...elatingMessageHandlerIntegrationTests.java | 10 +- .../config/AggregatorParserTests.java | 52 +++++- .../config/aggregatorParserTests.xml | 14 ++ ...ingAnnotationsWithBeanAnnotationTests.java | 5 +- src/reference/asciidoc/aggregator.adoc | 17 ++ src/reference/asciidoc/changes-3.0-4.0.adoc | 10 +- src/reference/asciidoc/file.adoc | 2 +- src/reference/asciidoc/gateway.adoc | 2 +- src/reference/asciidoc/whats-new.adoc | 16 +- 15 files changed, 362 insertions(+), 42 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleMessageGroupProcessor.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index 27ce3e0a49..2d778c1822 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 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 - * + * * 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. @@ -19,12 +19,16 @@ import org.springframework.integration.store.MessageGroup; * This implementation of MessageGroupProcessor will return all messages inside the group. * This is useful if there is no requirement to process the messages, but they should just be * blocked as a group until their ReleaseStrategy lets them pass through. - * + * + * @deprecated since 4.2; use {@link SimpleMessageGroupProcessor} + * * @author Iwein Fuld * @since 2.0.0 */ +@Deprecated public class PassThroughMessageGroupProcessor implements MessageGroupProcessor { + @Override public Object processMessageGroup(MessageGroup group) { return group.getMessages(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleMessageGroupProcessor.java new file mode 100644 index 0000000000..4f3c490c6b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleMessageGroupProcessor.java @@ -0,0 +1,36 @@ +/* + * 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.aggregator; + +import org.springframework.integration.store.MessageGroup; + +/** + * A {@link MessageGroupProcessor} that simply returns the messages in the group. + * It can be used to configure an aggregator as a barrier, such that when the group + * is complete, the grouped messages are released as individual messages. + * + * @author Gary Russell + * @since 4.2 + * + */ +public class SimpleMessageGroupProcessor implements MessageGroupProcessor { + + @Override + public Object processMessageGroup(MessageGroup group) { + return group.getMessages(); + } + +} 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 new file mode 100644 index 0000000000..e620989922 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -0,0 +1,175 @@ +/* + * 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; + +import java.util.List; + +import org.aopalliance.aop.Advice; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.expression.Expression; +import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.MessageGroupProcessor; +import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.core.DestinationResolver; +import org.springframework.scheduling.TaskScheduler; + + +/** + * {@link FactoryBean} to create an {@link AggregatingMessageHandler}. + * + * @author Gary Russell + * @since 4.2 + * + */ +public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBean + implements ApplicationContextAware, BeanNameAware, ApplicationEventPublisherAware { + + private final AggregatingMessageHandler aggregator; + + public AggregatorFactoryBean(Object processor) { + this(processor, null); + } + + public AggregatorFactoryBean(Object processor, String methodName) { + MessageGroupProcessor outputProcessor; + if (processor instanceof MessageGroupProcessor) { + outputProcessor = (MessageGroupProcessor) processor; + } + else { + if (methodName == null) { + outputProcessor = new MethodInvokingMessageGroupProcessor(processor); + } + else { + outputProcessor = new MethodInvokingMessageGroupProcessor(processor, methodName); + } + } + this.aggregator = new AggregatingMessageHandler(outputProcessor); + } + + public void setExpireGroupsUponCompletion(boolean expireGroupsUponCompletion) { + this.aggregator.setExpireGroupsUponCompletion(expireGroupsUponCompletion); + } + + public void setSendTimeout(long sendTimeout) { + this.aggregator.setSendTimeout(sendTimeout); + } + + public void setOutputChannelName(String outputChannelName) { + this.aggregator.setOutputChannelName(outputChannelName); + } + + public void configureMetrics(AbstractMessageHandlerMetrics metrics) { + this.aggregator.configureMetrics(metrics); + } + + @Override + public final void setBeanName(String beanName) { + this.aggregator.setBeanName(beanName); + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.aggregator.setApplicationContext(applicationContext); + } + + public void setChannelResolver(DestinationResolver channelResolver) { + this.aggregator.setChannelResolver(channelResolver); + } + + public void enableStats(boolean statsEnabled) { + this.aggregator.enableStats(statsEnabled); + } + + public void enableCounts(boolean countsEnabled) { + this.aggregator.enableCounts(countsEnabled); + } + + public void setLockRegistry(LockRegistry lockRegistry) { + this.aggregator.setLockRegistry(lockRegistry); + } + + public void setMessageStore(MessageGroupStore store) { + this.aggregator.setMessageStore(store); + } + + public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) { + this.aggregator.setCorrelationStrategy(correlationStrategy); + } + + public void setReleaseStrategy(ReleaseStrategy releaseStrategy) { + this.aggregator.setReleaseStrategy(releaseStrategy); + } + + public void setGroupTimeoutExpression(Expression groupTimeoutExpression) { + this.aggregator.setGroupTimeoutExpression(groupTimeoutExpression); + } + + public void setForceReleaseAdviceChain(List forceReleaseAdviceChain) { + this.aggregator.setForceReleaseAdviceChain(forceReleaseAdviceChain); + } + + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.aggregator.setTaskScheduler(taskScheduler); + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.aggregator.setApplicationEventPublisher(applicationEventPublisher); + } + + public void setDiscardChannel(MessageChannel discardChannel) { + this.aggregator.setDiscardChannel(discardChannel); + } + + public void setDiscardChannelName(String discardChannelName) { + this.aggregator.setDiscardChannelName(discardChannelName); + } + + public void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) { + this.aggregator.setSendPartialResultOnExpiry(sendPartialResultOnExpiry); + } + + public void setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups) { + this.aggregator.setMinimumTimeoutForEmptyGroups(minimumTimeoutForEmptyGroups); + } + + public void setReleasePartialSequences(boolean releasePartialSequences) { + this.aggregator.setReleasePartialSequences(releasePartialSequences); + } + + public void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) { + this.aggregator.setExpireGroupsUponTimeout(expireGroupsUponTimeout); + } + + @Override + protected AggregatingMessageHandler createHandler() { + return this.aggregator; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java index 06d88d0a04..f2b8288b57 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.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. @@ -73,7 +73,9 @@ public class ReleaseStrategyFactoryBean implements FactoryBean } else { if (logger.isWarnEnabled()) { - logger.warn("No annotated method found; falling back to SequenceSizeReleaseStrategy, target:" + logger.warn("No ReleaseStrategy annotated method found on " + + target.getClass().getSimpleName() + + "; falling back to SequenceSizeReleaseStrategy, target:" + target + ", methodName:" + methodName); } } 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 f6fc664e2a..ed05dec0d7 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 @@ -23,10 +23,9 @@ import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor; -import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; +import org.springframework.integration.config.AggregatorFactoryBean; import org.springframework.util.StringUtils; /** @@ -51,20 +50,17 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { String ref = element.getAttribute(REF_ATTRIBUTE); BeanDefinitionBuilder builder; - builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatingMessageHandler.class); - BeanDefinitionBuilder processorBuilder = null; + builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatorFactoryBean.class); BeanMetadataElement processor = null; if (innerHandlerDefinition != null || StringUtils.hasText(ref)) { - processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageGroupProcessor.class); - builder.addConstructorArgValue(processorBuilder.getBeanDefinition()); if (innerHandlerDefinition != null) { processor = innerHandlerDefinition; } else { processor = new RuntimeBeanReference(ref); } - processorBuilder.addConstructorArgValue(processor); + builder.addConstructorArgValue(processor); } else { if (StringUtils.hasText(element.getAttribute(EXPRESSION_ATTRIBUTE))) { @@ -81,8 +77,7 @@ public class AggregatorParser extends AbstractCorrelatingMessageHandlerParser { if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) { String method = element.getAttribute(METHOD_ATTRIBUTE); - processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, - "java.lang.String"); + builder.addConstructorArgValue(method); } this.doParse(builder, element, processor, parserContext); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java index d79cf0b902..38f8aed703 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 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 @@ -12,26 +12,34 @@ */ package org.springframework.integration.aggregator; +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; +import static org.junit.Assert.assertThat; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; + import org.springframework.integration.store.MessageGroup; import org.springframework.integration.support.MessageBuilder; - -import java.util.concurrent.*; - -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.Matchers.notNullValue; -import static org.hamcrest.Matchers.nullValue; -import static org.junit.Assert.assertThat; -import static org.mockito.Mockito.*; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; /** * @author Iwein Fuld + * @author Gary Russell */ @RunWith(MockitoJUnitRunner.class) public class CorrelatingMessageBarrierTests { @@ -97,6 +105,7 @@ public class CorrelatingMessageBarrierTests { private void sendAsynchronously(final MessageHandler handler, final Message message, final CountDownLatch start, final CountDownLatch sent) { Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { start.await(); @@ -121,6 +130,7 @@ public class CorrelatingMessageBarrierTests { private static class OneMessagePerKeyReleaseStrategy implements ReleaseStrategy { private final ConcurrentMap keyLocks = new ConcurrentHashMap(); + @Override public boolean canRelease(MessageGroup messageGroup) { Object correlationKey = messageGroup.getGroupId(); Semaphore lock = lockForKey(correlationKey); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java index 280d137cb0..83fd57678c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.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. @@ -33,13 +33,19 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +/** + * @author Iwein Fuld + * @author Dave Syer + * @author Gary Russell + * + */ public class CorrelatingMessageHandlerIntegrationTests { private final MessageGroupStore store = new SimpleMessageStore(100); private final MessageChannel outputChannel = mock(MessageChannel.class); - private final MessageGroupProcessor processor = new PassThroughMessageGroupProcessor(); + private final MessageGroupProcessor processor = new SimpleMessageGroupProcessor(); private final AggregatingMessageHandler defaultHandler = new AggregatingMessageHandler(processor, store); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 7079b385ae..9362629f94 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -26,6 +26,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.concurrent.atomic.AtomicReference; @@ -38,6 +39,7 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.CorrelationStrategy; @@ -46,6 +48,8 @@ import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStr import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.aggregator.SimpleMessageGroupProcessor; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.utils.IntegrationUtils; @@ -96,6 +100,38 @@ public class AggregatorParserTests { assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.messageBuilderFactory")); } + @Test + public void testAggregationWithMessageGroupProcessor() { + QueueChannel output = this.context.getBean("outputChannel", QueueChannel.class); + output.purge(null); + MessageChannel input = (MessageChannel) context.getBean("aggregatorWithMGPReferenceInput"); + List> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage("123", "id1", 3, 1, null)); + outboundMessages.add(createMessage("789", "id1", 3, 3, null)); + outboundMessages.add(createMessage("456", "id1", 3, 2, null)); + for (Message message : outboundMessages) { + input.send(message); + } + assertEquals(3, output.getQueueSize()); + output.purge(null); + } + + @Test + public void testAggregationWithMessageGroupProcessorAndStrategies() { + QueueChannel output = this.context.getBean("outputChannel", QueueChannel.class); + output.purge(null); + MessageChannel input = (MessageChannel) context.getBean("aggregatorWithCustomMGPReferenceInput"); + List> outboundMessages = new ArrayList>(); + outboundMessages.add(createMessage("123", "id1", 3, 1, null)); + outboundMessages.add(createMessage("789", "id1", 3, 3, null)); + outboundMessages.add(createMessage("456", "id1", 3, 2, null)); + for (Message message : outboundMessages) { + input.send(message); + } + assertEquals(3, output.getQueueSize()); + output.purge(null); + } + @Test public void testAggregationByExpression() { MessageChannel input = (MessageChannel) context.getBean("aggregatorWithExpressionsInput"); @@ -266,5 +302,19 @@ public class AggregatorParserTests { return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize) .setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build(); } - + + public static class MyMGP extends SimpleMessageGroupProcessor { + + @org.springframework.integration.annotation.ReleaseStrategy + public boolean release(Collection> messages) { + return messages.size() >= 3; + } + + @org.springframework.integration.annotation.CorrelationStrategy + public Object correlate(Message m) { + return new IntegrationMessageHeaderAccessor(m).getCorrelationId(); + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml index 3927a5f7ea..fd193992cf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml @@ -18,6 +18,14 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java index 6404513f4b..2a3e8d53a7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java @@ -45,7 +45,7 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy; import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy; -import org.springframework.integration.aggregator.PassThroughMessageGroupProcessor; +import org.springframework.integration.aggregator.SimpleMessageGroupProcessor; import org.springframework.integration.annotation.Filter; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Poller; @@ -76,6 +76,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Artem Bilan + * @author Gary Russell * @since 4.0 */ @ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class) @@ -188,7 +189,7 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Bean @ServiceActivator(inputChannel = "aggregatorChannel") public MessageHandler aggregator() { - AggregatingMessageHandler handler = new AggregatingMessageHandler(new PassThroughMessageGroupProcessor()); + AggregatingMessageHandler handler = new AggregatingMessageHandler(new SimpleMessageGroupProcessor()); handler.setCorrelationStrategy(new ExpressionEvaluatingCorrelationStrategy("1")); handler.setReleaseStrategy(new ExpressionEvaluatingReleaseStrategy("size() == 10")); handler.setOutputChannelName("splitterChannel"); diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 372279d1bd..a1155733eb 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -114,6 +114,23 @@ This method will be invoked for aggregating messages as follows: NOTE: In the interest of code simplicity, and promoting best practices such as low coupling, testability, etc., the preferred way of implementing the aggregation logic is through a POJO, and using the XML or annotation support for configuring it in the application. +If the `MessageGroupProcessor` 's `processMessageGroup` method returns a collection, it must be a collection of +`Messge` s. +In this case, the messages are released individually. +Prior to _version 4.2_, it was not possible to provide a `MessageGroupProcessor` using XML configuration, only POJO +methods could be used for aggregation. +Now, if the framework detects that the referenced (or inner) bean implements `MessageProcessor`, it is used as the +aggregator's output processor. + +If you wish to release a collection of objects from a custom `MessageGroupProcessor` as the payload of a message, your +class should extend `AbstractAggregatingMessageGroupProcessor` and implement `aggregatePayloads()`. + +Also, since _version 4.2_, a `SimpleMessageGroupProcessor` is provided; which simply returns the collection of +messages from the group, which, as indicated above, causes the released messages to be sent individually. + +This allows the aggregator to work as a message barrier where arriving messages are held until the release strategy +fires, and the group is released, as a sequence of individual messages. + ===== ReleaseStrategy The `ReleaseStrategy` interface is defined as follows: diff --git a/src/reference/asciidoc/changes-3.0-4.0.adoc b/src/reference/asciidoc/changes-3.0-4.0.adoc index 8c019f48d1..156a6f2d67 100644 --- a/src/reference/asciidoc/changes-3.0-4.0.adoc +++ b/src/reference/asciidoc/changes-3.0-4.0.adoc @@ -17,13 +17,13 @@ See <> ===== @EnableIntegration The `@EnableIntegration` annotation has been added, to permit declaration of standard Spring Integration beans when using `@Configuration` classes. -See <> for more information. +See <> for more information. [[x4.0-component-scan]] ===== @IntegrationComponentScan The `@IntegrationComponentScan` annotation has been added, to permit classpath scanning for Spring Integration specific components. -See <> for more information. +See <> for more information. [[x4.0-message-history]] ===== @EnableMessageHistory @@ -50,19 +50,19 @@ For more information seehttp://docs.spring.io/spring-boot/docs/current/reference ===== @GlobalChannelInterceptor As well as the `@EnableIntegration` annotation mentioned above, the `@GlobalChannelInterceptor` annotation has bean introduced. -For more information, see <>. +For more information, see <>. [[x4.0-integration-converter]] ===== @IntegrationConverter The `@IntegrationConverter` annotation has bean introduced, as an analogue of `` component. -For more information, see <>. +For more information, see <>. [[x4.0-enable-publisher]] ===== @EnablePublisher The `@EnablePublisher` annotation has been added, to allow the specification of a `default-publisher-channel` for `@Publisher` annotations. -See <> for more information. +See <> for more information. [[x4.0-redis-cms]] ===== Redis Channel Message Stores diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 82c0d7ca89..1eb57555de 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -187,7 +187,7 @@ IMPORTANT: It is important to understand that filters (including patterns, regex Any of these attributes set on the adapter are subsequently injected into the scanner. For this reason, if you need to provide a custom scanner and you have multiple file inbound adapters in the same application context, each adapter must be provided with its own instance of the scanner, either by declaring separate beans, or declaring `scope="prototype"` on the scanner bean so that the context will create a new instance for each use. -===== Limiting Memory Consumption +==== Limiting Memory Consumption A `HeadDirectoryScanner` can be used to limit the number of files retained in memory. This can be useful when scanning large directories. diff --git a/src/reference/asciidoc/gateway.adoc b/src/reference/asciidoc/gateway.adoc index 93ae59a4cc..af04180d0e 100644 --- a/src/reference/asciidoc/gateway.adoc +++ b/src/reference/asciidoc/gateway.adoc @@ -272,7 +272,7 @@ public interface TestGateway { ---- As with the XML version, Spring Integration creates the `proxy` implementation with its messaging infrastructure, when discovering these annotations during a component scan. -To perform this scan and register the `BeanDefinition` in the application context, add the `@IntegrationComponentScan` annotation to a `@Configuration` class - see also <>. +To perform this scan and register the `BeanDefinition` in the application context, add the `@IntegrationComponentScan` annotation to a `@Configuration` class - see also <>. [[gateway-calling-no-argument-methods]] ==== Invoking No-Argument Methods diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b2e1fd8ded..e21a82edc3 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -176,13 +176,23 @@ See <> for more information. ===== MessagingGateway Annotation The request and reply timeout properties are now `String` instead of `Long` to allow configuration with property -placeholders or SpEL. See [[messaging-gateway-annotation]]. +placeholders or SpEL. See <>. -[[x4.2-aggregator-perf]] -==== Aggregator Performance +[[x4.2-aggregator-changes]] +==== Aggregator Changes + +===== Aggregator Performance This release includes some performance improvements for aggregating components (aggregator, resequencer, etc), by more efficiently removing messages from groups when they are released. New methods (`removeMessagesFromGroup`) have been added to the message store. Set the `removeBatchSize` property (default `100`) to adjust the number of messages deleted in each operation. Currently, JDBC, Redis and MongoDB message stores support this property. + +===== Output MessageGroupProcessor + +When using a `ref` or innner bean for the aggregator, it is now possible to bind a `MessageGroupProcessor` directly. +In addition, a `SimpleMessageGroupProcessor` is provided that simply returns the collection of messages in the group. +When an output processor produces a collection of `Message`, the aggregator releases those messages individually. +Configuring the `SimpleMessageGroupProcessor` makes the aggregator a message barrier, were messages are held up +until they all arrive, and are then released individually. See <> for more information.