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

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();
 	}

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();
+	}
+
+}

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<AggregatingMessageHandler>
+				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<MessageChannel> 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<Advice> 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;
+	}
+
+}

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<ReleaseStrategy>
 				}
 				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);
 					}
 				}

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);

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<Object> 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<Object, Semaphore> keyLocks = new ConcurrentHashMap<Object, Semaphore>();
 
+		@Override
 		public boolean canRelease(MessageGroup messageGroup) {
 			Object correlationKey = messageGroup.getGroupId();
 			Semaphore lock = lockForKey(correlationKey);

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);
 

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<Message<?>> outboundMessages = new ArrayList<Message<?>>();
+		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<Message<?>> outboundMessages = new ArrayList<Message<?>>();
+		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<Message<?>> messages) {
+			return messages.size() >= 3;
+		}
+
+		@org.springframework.integration.annotation.CorrelationStrategy
+		public Object correlate(Message<?> m) {
+			return new IntegrationMessageHeaderAccessor(m).getCorrelationId();
+		}
+
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/config/aggregatorParserTests.xml

@@ -18,6 +18,14 @@
 	<aggregator id="aggregatorWithReference" ref="aggregatorBean"
 				input-channel="aggregatorWithReferenceInput" output-channel="outputChannel"/>
 
+	<channel id="aggregatorWithMGPReferenceInput"/>
+	<aggregator id="aggregatorWithMGPReference" ref="aggregatorMGPBean"
+				input-channel="aggregatorWithMGPReferenceInput" output-channel="outputChannel"/>
+
+	<channel id="aggregatorWithCustomMGPReferenceInput"/>
+	<aggregator id="aggregatorWithCustomMGPReference" ref="aggregatorCustomMGPBean"
+				input-channel="aggregatorWithCustomMGPReferenceInput" output-channel="outputChannel"/>
+
 	<channel id="completelyDefinedAggregatorInput"/>
 	<aggregator id="completelyDefinedAggregator"
 		input-channel="completelyDefinedAggregatorInput"
@@ -75,6 +83,12 @@
 	<beans:bean id="aggregatorBean"
 		class="org.springframework.integration.config.TestAggregatorBean" />
 
+	<beans:bean id="aggregatorMGPBean"
+		class="org.springframework.integration.aggregator.SimpleMessageGroupProcessor" />
+
+	<beans:bean id="aggregatorCustomMGPBean"
+		class="org.springframework.integration.config.AggregatorParserTests$MyMGP" />
+
 	 <beans:bean id="adderBean"
 		class="org.springframework.integration.config.Adder" />
 

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");

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:

src/reference/asciidoc/changes-3.0-4.0.adoc

@@ -17,13 +17,13 @@ See <<mqtt>>
 ===== @EnableIntegration
 
 The `@EnableIntegration` annotation has been added, to permit declaration of standard Spring Integration beans when using `@Configuration` classes.
-See <<enable-integration>> for more information.
+See <<annotations>> for more information.
 
 [[x4.0-component-scan]]
 ===== @IntegrationComponentScan
 
 The `@IntegrationComponentScan` annotation has been added, to permit classpath scanning for Spring Integration specific components.
-See <<enable-integration>> for more information.
+See <<annotations>> 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 <<enable-integration>>.
+For more information, see <<annotations>>.
 
 [[x4.0-integration-converter]]
 ===== @IntegrationConverter
 
 The `@IntegrationConverter` annotation has bean introduced, as an analogue of `<int:converter/>` component.
-For more information, see <<enable-integration>>.
+For more information, see <<annotations>>.
 
 [[x4.0-enable-publisher]]
 ===== @EnablePublisher
 
 The `@EnablePublisher` annotation has been added, to allow the specification of a `default-publisher-channel` for `@Publisher` annotations.
-See <<enable-integration>> for more information.
+See <<annotations>> for more information.
 
 [[x4.0-redis-cms]]
 ===== Redis Channel Message Stores

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.

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 <<enable-integration>>.
+To perform this scan and register the `BeanDefinition` in the application context, add the `@IntegrationComponentScan` annotation to a `@Configuration` class - see also <<annotations>>.
 
 [[gateway-calling-no-argument-methods]]
 ==== Invoking No-Argument Methods

src/reference/asciidoc/whats-new.adoc

@@ -176,13 +176,23 @@ See <<gw-completable-future>> 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 <<messaging-gateway-annotation>>.
 
-[[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 <<aggregator>> for more information.