INT-275: Implement Scatter-Gather Pattern

JIRA: https://jira.spring.io/browse/INT-275

In addition fix the `Lifecycle` issue in the `ServiceActivatorAnnotationPostProcessor`

Add Namespace support, Addition tests
   and fix some typos in the XSD

INT-275: Fix failed tests

INT-275: Fix for `replyChannel` Header

Add `sync` reply test-case

INT-275: Make `ScatterGatherHandler` sync

Fix some `MessageHandler`s from `SmartLifecycle`

INT-275: Fix `ScatterGatherHandler.handleRequestMessage` logic

INT-275: Fix `ScatterGatherHandler` JMX proxying issues

* Make `AbstractCorrelatingMessageHandler.getMessageStore()` as `public`
* Move `ScatterGatherHandlerIntegrationTests` to the JMX module to be sure that `ScatterGatherHandler`
works well with `@EnableIntegrationMBeanExport`
* Add xml config sample how to use an internal gatherer's `MessageStore` in the `MessageGroupStoreReaper`
* Add `What's New` notice

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -304,7 +304,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		return "aggregator";
 	}
 
-	protected MessageGroupStore getMessageStore() {
+	public MessageGroupStore getMessageStore() {
 		return messageStore;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java

@@ -21,6 +21,7 @@ import java.lang.reflect.Method;
 import java.util.List;
 
 import org.springframework.beans.factory.ListableBeanFactory;
+import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.env.Environment;
@@ -58,15 +59,7 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
 			 		 * Return a reply-producing message handler so that we still get 'produced no reply' messages
 			 		 * and the super class will inject the advice chain to advise the handler method if needed.
 			 		 */
-					return new AbstractReplyProducingMessageHandler() {
-
-						@Override
-						protected Object handleRequestMessage(Message<?> requestMessage) {
-
-							((MessageHandler) target).handleMessage(requestMessage);
-							return null;
-						}
-					};
+					return new ReplyProducingMessageHandlerWrapper((MessageHandler) target);
 				}
 				else {
 					serviceActivator = new ServiceActivatingHandler(target);
@@ -89,4 +82,41 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
 		return serviceActivator;
 	}
 
+	private class ReplyProducingMessageHandlerWrapper extends AbstractReplyProducingMessageHandler
+			implements Lifecycle {
+
+		private final MessageHandler target;
+
+		private ReplyProducingMessageHandlerWrapper(MessageHandler target) {
+			this.target = target;
+		}
+
+		@Override
+		protected Object handleRequestMessage(Message<?> requestMessage) {
+			this.target.handleMessage(requestMessage);
+			return null;
+		}
+
+		@Override
+		public void start() {
+			if (this.target instanceof Lifecycle) {
+				((Lifecycle) this.target).start();
+			}
+
+		}
+
+		@Override
+		public void stop() {
+			if (this.target instanceof Lifecycle) {
+				((Lifecycle) this.target).stop();
+			}
+		}
+
+		@Override
+		public boolean isRunning() {
+			return !(this.target instanceof Lifecycle) || ((Lifecycle) this.target).isRunning();
+		}
+
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java

@@ -81,6 +81,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
 		RetryAdviceParser retryParser = new RetryAdviceParser();
 		registerBeanDefinitionParser("handler-retry-advice", retryParser);
 		registerBeanDefinitionParser("retry-advice", retryParser);
+		registerBeanDefinitionParser("scatter-gather", new ScatterGatherParser());
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.config.xml;
+
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.parsers.ParserConfigurationException;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.handler.ScatterGatherHandler;
+import org.springframework.integration.router.RecipientListRouter;
+import org.springframework.util.StringUtils;
+import org.springframework.util.xml.DomUtils;
+
+/**
+ * Parser for the <scatter-gather> element.
+ *
+ * @author Artem Bilan
+ * @since 4.1
+ */
+public class ScatterGatherParser extends AbstractConsumerEndpointParser {
+
+	private static final RecipientListRouterParser SCATTERER_PARSER = new RecipientListRouterParser();
+
+	private static final AggregatorParser GATHERER_PARSER = new AggregatorParser();
+
+	private static final DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();
+
+	@Override
+	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
+		String scatterChannel = element.getAttribute("scatter-channel");
+		boolean hasScatterChannel = StringUtils.hasText(scatterChannel);
+		Element scatterer = DomUtils.getChildElementByTagName(element, "scatterer");
+		boolean hasScatterer = scatterer != null;
+
+		if (hasScatterChannel & hasScatterer) {
+			parserContext.getReaderContext()
+					.error("'scatter-channel' attribute and 'scatterer' sub-element are mutually exclusive", element);
+		}
+
+		if (!hasScatterChannel & !hasScatterer) {
+			parserContext.getReaderContext()
+					.error("The 'scatter-channel' attribute or 'scatterer' sub-element must be specified", element);
+		}
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ScatterGatherHandler.class);
+		AbstractBeanDefinition scatterGatherDefinition = builder.getRawBeanDefinition();
+		String id = resolveId(element, scatterGatherDefinition, parserContext);
+
+		if (hasScatterChannel) {
+			builder.addConstructorArgReference(scatterChannel);
+		}
+		else {
+			BeanDefinition scattererDefinition = null;
+			if (!hasScatterer) {
+				scattererDefinition = new RootBeanDefinition(RecipientListRouter.class);
+			}
+			else {
+				scattererDefinition = SCATTERER_PARSER.parse(scatterer,
+						new ParserContext(parserContext.getReaderContext(), parserContext.getDelegate(),
+								scatterGatherDefinition));
+			}
+
+			String scattererId = id + ".scatterer";
+			if (hasScatterer && scatterer.hasAttribute(ID_ATTRIBUTE)) {
+				scattererId = scatterer.getAttribute(ID_ATTRIBUTE);
+			}
+			parserContext.getRegistry().registerBeanDefinition(scattererId, scattererDefinition);
+			builder.addConstructorArgValue(new RuntimeBeanReference(scattererId));
+		}
+
+		Element gatherer = DomUtils.getChildElementByTagName(element, "gatherer");
+
+		BeanDefinition gathererDefinition = null;
+		if (gatherer == null) {
+			try {
+				gatherer = documentBuilderFactory.newDocumentBuilder().newDocument().createElement("aggregator");
+			}
+			catch (ParserConfigurationException e) {
+				parserContext.getReaderContext().error(e.getMessage(), element);
+			}
+		}
+		gathererDefinition = GATHERER_PARSER.parse(gatherer, new ParserContext(parserContext.getReaderContext(),
+				parserContext.getDelegate(), scatterGatherDefinition));
+		String gathererId = id + ".gatherer";
+		if (gatherer != null && gatherer.hasAttribute(ID_ATTRIBUTE)) {
+			gathererId = gatherer.getAttribute(ID_ATTRIBUTE);
+		}
+		parserContext.getRegistry().registerBeanDefinition(gathererId, gathererDefinition);
+		builder.addConstructorArgValue(new RuntimeBeanReference(gathererId));
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "gather-channel");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "gather-timeout");
+
+		return builder;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/ScatterGatherHandler.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.handler;
+
+import org.springframework.aop.support.AopUtils;
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.endpoint.PollingConsumer;
+import org.springframework.integration.router.RecipientListRouter;
+import org.springframework.integration.support.channel.HeaderChannelRegistry;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.util.Assert;
+
+/**
+ * The {@link MessageHandler} implementation for the
+ * <a href="http://www.eaipatterns.com/BroadcastAggregate.html">Scatter-Gather</a> EIP pattern.
+ *
+ * @author Artem Bilan
+ * @since 4.1
+ */
+public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler implements Lifecycle {
+
+	private static final String GATHER_RESULT_CHANNEL = "gatherResultChannel";
+
+	private final MessageChannel scatterChannel;
+
+	private final MessageHandler gatherer;
+
+	private MessageChannel gatherChannel;
+
+	private long gatherTimeout = -1;
+
+	private AbstractEndpoint gatherEndpoint;
+
+	private HeaderChannelRegistry replyChannelRegistry;
+
+
+	public ScatterGatherHandler(MessageChannel scatterChannel, MessageHandler gatherer) {
+		Assert.notNull(scatterChannel);
+		Assert.notNull(gatherer);
+		Class<?> gathererClass = AopUtils.getTargetClass(gatherer);
+		Assert.isAssignable(AggregatingMessageHandler.class, gathererClass,
+				"the 'gatherer' must be an AggregatingMessageHandler instance");
+		this.scatterChannel = scatterChannel;
+		this.gatherer = gatherer;
+	}
+
+	public ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer) {
+		this(new FixedSubscriberChannel(scatterer), gatherer);
+		Assert.notNull(scatterer);
+		Class<?> scatterClass = AopUtils.getTargetClass(scatterer);
+		Assert.isAssignable(RecipientListRouter.class, scatterClass,
+				"the 'scatterer' must be a RecipientListRouter instance");
+	}
+
+	public void setGatherChannel(MessageChannel gatherChannel) {
+		this.gatherChannel = gatherChannel;
+	}
+
+	public void setGatherTimeout(long gatherTimeout) {
+		this.gatherTimeout = gatherTimeout;
+	}
+
+	@Override
+	protected void doInit() {
+		if (this.gatherChannel == null) {
+			this.gatherChannel = new FixedSubscriberChannel(this.gatherer);
+		}
+		else {
+			if (this.gatherChannel instanceof SubscribableChannel) {
+				this.gatherEndpoint = new EventDrivenConsumer((SubscribableChannel) this.gatherChannel, this.gatherer);
+			}
+			else if (this.gatherChannel instanceof PollableChannel) {
+				this.gatherEndpoint = new PollingConsumer((PollableChannel) this.gatherChannel, this.gatherer);
+				((PollingConsumer) this.gatherEndpoint).setReceiveTimeout(this.gatherTimeout);
+			}
+			else {
+				throw new MessagingException("Unsupported 'replyChannel' type [" + this.gatherChannel.getClass() + "]."
+						+ "SubscribableChannel or PollableChannel type are supported.");
+			}
+			this.gatherEndpoint.setBeanFactory(this.getBeanFactory());
+			this.gatherEndpoint.afterPropertiesSet();
+		}
+
+		((MessageProducer) this.gatherer).setOutputChannel(new FixedSubscriberChannel(new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				MessageHeaders headers = message.getHeaders();
+				if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
+					Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
+					if (gatherResultChannel instanceof MessageChannel) {
+						messagingTemplate.send((MessageChannel) gatherResultChannel, message);
+					}
+					else if (gatherResultChannel instanceof String) {
+						messagingTemplate.send((String) gatherResultChannel, message);
+					}
+				}
+				else {
+					throw new MessageDeliveryException(message,
+							"The 'gatherResultChannel' header is required to delivery gather result.");
+				}
+			}
+
+		}));
+
+		this.replyChannelRegistry = getBeanFactory()
+				.getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
+						HeaderChannelRegistry.class);
+	}
+
+	@Override
+	protected Object handleRequestMessage(Message<?> requestMessage) {
+		PollableChannel gatherResultChannel = new QueueChannel();
+
+		Object gatherResultChannelName = this.replyChannelRegistry.channelToChannelName(gatherResultChannel);
+
+		Message<?> scatterMessage = getMessageBuilderFactory()
+				.fromMessage(requestMessage)
+				.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName)
+				.setReplyChannel(this.gatherChannel)
+				.build();
+
+		this.messagingTemplate.send(this.scatterChannel, scatterMessage);
+
+		Message<?> gatherResult = gatherResultChannel.receive(this.gatherTimeout);
+		if (gatherResult != null) {
+			return gatherResult.getPayload();
+		}
+
+		return null;
+	}
+
+	@Override
+	public void start() {
+		if (this.gatherEndpoint != null) {
+			gatherEndpoint.start();
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.gatherEndpoint != null) {
+			gatherEndpoint.start();
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.gatherEndpoint == null || this.gatherEndpoint.isRunning();
+	}
+
+}

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd

@@ -680,8 +680,8 @@
 					<xsd:documentation>
 										<![CDATA[
 				An expression that will be used to generate the payload for all methods in the service interface
-				unless explicitly overridden by a method declaration. Variables include #args, #methodName, #methodString
-				and #methodObject; a bean resolver is also available, enabling expressions like "@someBean(#args)".
+				unless explicitly overridden by a method declaration. Variables include #args, #gatewayMethod;
+				a bean resolver is also available, enabling expressions like "@someBean(#args)".
 										]]>
 					</xsd:documentation>
 				</xsd:annotation>
@@ -762,7 +762,7 @@
 					but a custom executor can return any Future. If the method return type is
 					not compatible with the executor the flow will run on the caller's thread and
 					the flow must return an appropriate Future. Finally, you can disable the
-					gateway ansync handling by setting this attribute to "". This allows the downstream
+					gateway async handling by setting this attribute to "". This allows the downstream
 					flow to return a Future that would otherwise have been compatible with the
 					default executor.
 							]]>
@@ -1446,7 +1446,7 @@
 			<xsd:annotation>
 				<xsd:documentation>
 					Boolean value indicating whether any payload that implements Cloneable should be cloned
-					prior to sending the Message to the request chanenl for acquiring the enriching data.
+					prior to sending the Message to the request channel for acquiring the enriching data.
 					The cloned version would be used as the target payload for the ultimate reply.
 
 					If the payload does NOT implement 'Cloneable', then setting this
@@ -1800,7 +1800,7 @@
 	The java.util.concurrent.TimeUnit enum value. This can ONLY be used in combination
 	with the 'fixed-delay' or 'fixed-rate' attributes. If combined with either 'cron'
 	or a 'trigger' reference attribute, it will cause a failure. The minimal supported
-	granularity for a PeriodicTrigger is MILLISEONDS. Therefore, the only available options
+	granularity for a PeriodicTrigger is MILLISECONDS. Therefore, the only available options
 	are MILLISECONDS and SECONDS. If this value is not provided, then any 'fixed-delay' or
 	'fixed-rate' value will be interpreted as MILLISECONDS by default. Basically this enum
 	provides a convenience for SECONDS-based interval trigger values. For hourly, daily,
@@ -2885,7 +2885,7 @@
 
     <xsd:group name="routerCommonGroup">
         <xsd:sequence>
-            <xsd:element name="expression" type="innerExpressionType"        minOccurs="0" maxOccurs="1">
+            <xsd:element name="expression" type="innerExpressionType" minOccurs="0" maxOccurs="1">
                 <xsd:annotation>
                     <xsd:documentation>
                     <![CDATA[
@@ -2903,7 +2903,7 @@
                     ]]></xsd:documentation>
                 </xsd:annotation>
             </xsd:element>
-            <xsd:element name="mapping"    type="mappingValueChannelType" minOccurs="0" maxOccurs="unbounded">
+            <xsd:element name="mapping" type="mappingValueChannelType" minOccurs="0" maxOccurs="unbounded">
                 <xsd:annotation>
                     <xsd:documentation>
                     <![CDATA[
@@ -2912,7 +2912,7 @@
                     ]]></xsd:documentation>
                 </xsd:annotation>
             </xsd:element>
-            <xsd:any namespace="##other" processContents="strict"            minOccurs="0" maxOccurs="1" />
+            <xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="1" />
         </xsd:sequence>
     </xsd:group>
 
@@ -3319,7 +3319,7 @@
 
                     If set to 'true', a MessagingException will be raised in case
                     the channel cannot be resolved. Setting this attribute to 'false',
-                    will cause any unresovable channels to be ignored.
+                    will cause any unresolvable channels to be ignored.
 
                     If not explicitly set, 'resolution-required' will
                     default to 'true'.
@@ -3699,7 +3699,7 @@
 				<xsd:element name="wire-tap" type="wireTapType" minOccurs="0" maxOccurs="unbounded">
 					<xsd:annotation>
 						<xsd:documentation>
-						Alows you to configure a Wire Tap interceptor that will send a copy of the message to a
+						Allows you to configure a Wire Tap interceptor that will send a copy of the message to a
 						channel identified by 'channel' attribute.
 						</xsd:documentation>
 					</xsd:annotation>
@@ -4210,6 +4210,80 @@ The list of component name patterns you want to track (e.g.,  tracked-components
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="scatter-gather">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+                Defines a Scatter-Gather.
+            ]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="scatter-gather-type">
+					<xsd:sequence>
+						<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
+					</xsd:sequence>
+					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:complexType name="scatter-gather-type">
+		<xsd:annotation>
+			<xsd:documentation>
+				Base type for 'scatter-gather' elements.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:choice minOccurs="0" maxOccurs="2">
+			<xsd:element name="scatterer">
+				<xsd:complexType>
+					<xsd:complexContent>
+						<xsd:extension base="abstractRouterType">
+							<xsd:group ref="commonRecipientListRouterGroup"/>
+						</xsd:extension>
+					</xsd:complexContent>
+				</xsd:complexType>
+			</xsd:element>
+			<xsd:element name="gatherer" type="aggregator-type"/>
+		</xsd:choice>
+		<xsd:attribute name="id" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					'id' value:
+					- Identifies the underlying Spring bean definition (AbstractEndpoint)
+					- as MessageHandler bean alias together with suffix '.handler'
+					- as a RecipientListRouter bean together with suffix '.scatterer'
+					- as a AggregatingMessageHandler bean together with suffix '.gatherer'
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="scatter-channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Identifies the channel to send a message for 'auction' Scatter-Gather pattern variant.
+					Mutually exclusive with 'scatterer' sub-element.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="gather-channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Identifies the channel to receive reply Messages for gathering.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="gather-timeout" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Allows to specify how long the Scatter-Gather will wait for reply Messages for gathering.
+					By default it will wait indefinitely. Value is specified in milliseconds.
+					It will be applied only if the 'gather-channel' is specified and it is some blocking channel,
+					e.g. 'QueueChannel'.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
 	<xsd:complexType name="control-bus-type">
 		<xsd:annotation>
 			<xsd:documentation><![CDATA[
@@ -4276,7 +4350,7 @@ The list of component name patterns you want to track (e.g.,  tracked-components
 					<xsd:attribute name="maximum" type="xsd:string">
 						<xsd:annotation>
 							<xsd:documentation><![CDATA[
-								The maxumum interval in milliseconds between attempts; caps an interval
+								The maximum interval in milliseconds between attempts; caps an interval
 								calculated using the multiplier. Default 30000.
 							]]></xsd:documentation>
 						</xsd:annotation>
@@ -4374,7 +4448,7 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 	<xsd:attributeGroup name="transactionSyncAttributeGroup">
 		<xsd:annotation>
 			<xsd:documentation><![CDATA[
-				Attributes provided in either a <transactional/> or <psedo-transactional/> poller
+				Attributes provided in either a <transactional/> or <pseudo-transactional/> poller
 				sub element.
 				Used to take action after the transaction completes (<transactional/>) or after
 				the channel.send() is complete (<pseudo-transactional/>).
@@ -4431,7 +4505,7 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 		<xsd:attribute name="send-timeout">
 			<xsd:annotation>
 				<xsd:documentation><![CDATA[
-					A timout used when sending expression evaluation results to the success or
+					A timeout used when sending expression evaluation results to the success or
 					failure channels. Only applies if the channel can block on a send, such as
 					a limited-capacity QueueChannel that is currently full.
 				]]></xsd:documentation>

spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml

@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<int:scatter-gather id="scatterGather1" input-channel="input1" scatter-channel="scatterChannel"/>
+
+	<bean id="reaper" class="org.springframework.integration.store.MessageGroupStoreReaper">
+		<property name="messageGroupStore" value="#{@'scatterGather1.gatherer'.messageStore}"/>
+	</bean>
+
+	<bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
+
+	<int:scatter-gather id="scatterGather2" input-channel="input2" gather-channel="gatherChannel" gather-timeout="100">
+		<int:scatterer id="myScatterer" apply-sequence="true">
+			<int:recipient channel="distributionChannel"/>
+		</int:scatterer>
+		<int:gatherer id="myGatherer" message-store="messageStore"/>
+	</int:scatter-gather>
+
+	<int:channel id="gatherChannel"/>
+
+	<int:channel id="distributionChannel"/>
+
+	<int:channel id="scatterChannel"/>
+
+
+	<!--Invalid configurations-->
+
+	<!--<int:scatter-gather input-channel="inputInvalid1"/>-->
+
+	<!--<int:scatter-gather input-channel="inputInvalid2" scatter-channel="scatterChannel">
+		<int:scatterer/>
+	</int:scatter-gather>-->
+
+
+</beans>

spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2014 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.scattergather.config;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collection;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.channel.FixedSubscriberChannel;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.handler.ScatterGatherHandler;
+import org.springframework.integration.router.RecipientListRouter;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Artem Bilan
+ * @since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class ScatterGatherParserTests {
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	@Test
+	public void testAuction() {
+		MessageHandler scatterGather = this.beanFactory.getBean("scatterGather1.handler", MessageHandler.class);
+		assertThat(scatterGather, instanceOf(ScatterGatherHandler.class));
+		assertSame(this.beanFactory.getBean("scatterChannel"),
+				TestUtils.getPropertyValue(scatterGather, "scatterChannel"));
+		assertTrue(this.beanFactory.containsBean("scatterGather1.gatherer"));
+		AggregatingMessageHandler gatherer =
+				this.beanFactory.getBean("scatterGather1.gatherer", AggregatingMessageHandler.class);
+		assertSame(gatherer, TestUtils.getPropertyValue(scatterGather, "gatherer"));
+
+		Object reaper = this.beanFactory.getBean("reaper");
+		assertSame(gatherer.getMessageStore(), TestUtils.getPropertyValue(reaper, "messageGroupStore"));
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testDistribution() {
+		MessageHandler scatterGather = this.beanFactory.getBean("scatterGather2.handler", MessageHandler.class);
+		assertSame(this.beanFactory.getBean("gatherChannel"),
+				TestUtils.getPropertyValue(scatterGather, "gatherChannel"));
+		assertNotNull(TestUtils.getPropertyValue(scatterGather, "gatherEndpoint"));
+		assertThat(TestUtils.getPropertyValue(scatterGather, "gatherEndpoint"), instanceOf(EventDrivenConsumer.class));
+		assertTrue(TestUtils.getPropertyValue(scatterGather, "gatherEndpoint.running", Boolean.class));
+		assertEquals(100L, TestUtils.getPropertyValue(scatterGather, "gatherTimeout"));
+
+		assertTrue(this.beanFactory.containsBean("myGatherer"));
+		Object gatherer = this.beanFactory.getBean("myGatherer");
+		assertSame(gatherer, TestUtils.getPropertyValue(scatterGather, "gatherer"));
+		assertSame(this.beanFactory.getBean("messageStore"), TestUtils.getPropertyValue(gatherer, "messageStore"));
+		assertSame(gatherer, TestUtils.getPropertyValue(scatterGather, "gatherEndpoint.handler"));
+
+		assertTrue(this.beanFactory.containsBean("myScatterer"));
+		Object scatterer = this.beanFactory.getBean("myScatterer");
+		assertTrue(TestUtils.getPropertyValue(scatterer, "applySequence", Boolean.class));
+
+		Collection<RecipientListRouter.Recipient> recipients = TestUtils.getPropertyValue(scatterer, "recipients",
+				Collection.class);
+		assertEquals(1, recipients.size());
+		assertSame(this.beanFactory.getBean("distributionChannel"), recipients.iterator().next().getChannel());
+
+		Object scatterChannel = TestUtils.getPropertyValue(scatterGather, "scatterChannel");
+		assertThat(scatterChannel, instanceOf(FixedSubscriberChannel.class));
+		assertSame(scatterer, TestUtils.getPropertyValue(scatterChannel, "handler"));
+
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 xmlns="http://www.springframework.org/schema/integration"
+			 xmlns:task="http://www.springframework.org/schema/task"
+			 xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	   http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
+
+	<channel id="output">
+		<queue/>
+	</channel>
+
+	<!--Auction scenario-->
+	<scatter-gather input-channel="inputAuction" output-channel="output" scatter-channel="auctionChannel">
+		<gatherer release-strategy-expression="^[payload gt 5] != null or size() == 3"/>
+	</scatter-gather>
+
+	<task:executor id="threadPoolTaskExecutor" pool-size="10"/>
+
+	<publish-subscribe-channel id="auctionChannel" apply-sequence="true" task-executor="threadPoolTaskExecutor"/>
+
+	<bridge input-channel="auctionChannel" output-channel="serviceChannel1"/>
+
+	<bridge input-channel="auctionChannel" output-channel="serviceChannel1"/>
+
+	<bridge input-channel="auctionChannel" output-channel="serviceChannel1"/>
+
+	<service-activator input-channel="serviceChannel1" expression="T(java.lang.Math).random() * 10"/>
+
+	<!--Distribution scenario-->
+	<scatter-gather input-channel="inputDistribution" output-channel="output" gather-channel="gatherChannel">
+		<scatterer apply-sequence="true">
+			<recipient channel="distribution1Channel"/>
+			<recipient channel="distribution2Channel"/>
+			<recipient channel="distribution3Channel"/>
+		</scatterer>
+		<gatherer release-strategy-expression="^[payload gt 5] != null or size() == 3"/>
+	</scatter-gather>
+
+	<channel id="gatherChannel">
+		<queue/>
+	</channel>
+
+	<bridge input-channel="distribution1Channel" output-channel="serviceChannel2"/>
+
+	<bridge input-channel="distribution2Channel" output-channel="serviceChannel2"/>
+
+	<bridge input-channel="distribution3Channel" output-channel="serviceChannel2"/>
+
+	<service-activator input-channel="serviceChannel2" output-channel="gatherChannel"
+					   expression="T(java.lang.Math).random() * 10"/>
+
+	<!--Sync scenario-->
+
+	<gateway id="gateway" default-request-channel="gatewayAuction"/>
+
+	<scatter-gather input-channel="gatewayAuction" scatter-channel="auctionChannel">
+		<gatherer release-strategy-expression="^[payload gt 5] != null or size() == 3"/>
+	</scatter-gather>
+
+</beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2014 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.scattergather.config;
+
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+
+import java.util.List;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Artem Bilan
+ * @since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class ScatterGatherTests {
+
+	@Autowired
+	private PollableChannel output;
+
+	@Autowired
+	private MessageChannel inputAuction;
+
+	@Autowired
+	private MessageChannel inputDistribution;
+
+	@Autowired
+	private RequestReplyExchanger gateway;
+
+	@Test
+	public void testAuction() {
+		this.inputAuction.send(new GenericMessage<String>("foo"));
+		Message<?> bestQuoteMessage = this.output.receive(10000);
+		assertNotNull(bestQuoteMessage);
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(List.class));
+		assertThat(((List<?>) payload).size(), greaterThanOrEqualTo(1));
+	}
+
+	@Test
+	public void testDistribution() {
+		this.inputDistribution.send(new GenericMessage<String>("foo"));
+		Message<?> bestQuoteMessage = this.output.receive(10000);
+		assertNotNull(bestQuoteMessage);
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(List.class));
+		assertThat(((List<?>) payload).size(), greaterThanOrEqualTo(1));
+	}
+
+	@Test
+	public void testGatewayScatterGather() {
+		Message<?> bestQuoteMessage = this.gateway.exchange(new GenericMessage<String>("foo"));
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(List.class));
+		assertThat(((List<?>) payload).size(), greaterThanOrEqualTo(1));
+	}
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -22,6 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.ip.tcp.TcpOutboundGateway;
 
 /**
  * Parser for the <outbound-gateway> element of the integration 'jms' namespace.
@@ -31,8 +32,6 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
  */
 public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 
-	private static final String BASE_PACKAGE = "org.springframework.integration.ip.tcp";
-
 	@Override
 	protected String getInputChannelAttributeName() {
 		return "request-channel";
@@ -40,8 +39,7 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE +
-				".TcpOutboundGateway");
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TcpOutboundGateway.class);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
@@ -52,10 +50,6 @@ public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
 				IpAdapterParserUtils.REMOTE_TIMEOUT);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.REPLY_TIMEOUT, "sendTimeout");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
-				IntegrationNamespaceUtils.AUTO_STARTUP);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
-				IntegrationNamespaceUtils.PHASE);
 		return builder;
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java

@@ -22,6 +22,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -50,7 +51,8 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @since 2.0
  */
-public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler implements TcpSender, TcpListener, SmartLifecycle {
+public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
+		implements TcpSender, TcpListener, Lifecycle {
 
 	private volatile AbstractClientConnectionFactory connectionFactory;
 
@@ -64,10 +66,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
 
 	private volatile long requestTimeout = 10000;
 
-	private volatile boolean autoStartup = true;
-
-	private volatile int phase;
-
 	/**
 	 * @param requestTimeout the requestTimeout to set
 	 */
@@ -232,29 +230,6 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp
 		return this.connectionFactory.isRunning();
 	}
 
-	@Override
-	public int getPhase() {
-		return this.phase;
-	}
-
-	@Override
-	public boolean isAutoStartup() {
-		return this.autoStartup;
-	}
-
-	@Override
-	public void stop(Runnable callback) {
-		this.connectionFactory.stop(callback);
-	}
-
-	public void setAutoStartup(boolean autoStartup) {
-		this.autoStartup = autoStartup;
-	}
-
-	public void setPhase(int phase) {
-		this.phase = phase;
-	}
-
 	/**
 	 * @return the connectionFactory
 	 */

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java

@@ -496,8 +496,6 @@ public class ParserUnitTests {
 		assertEquals("ip:tcp-outbound-gateway", tcpOutboundGateway.getComponentType());
 		assertTrue(cfC2.isLookupHost());
 		assertEquals(24, dfa.getPropertyValue("order"));
-		assertFalse(tcpOutboundGateway.isAutoStartup());
-		assertEquals(127, tcpOutboundGateway.getPhase());
 	}
 
 	@Test

spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java

@@ -42,7 +42,7 @@ import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
-import org.springframework.context.SmartLifecycle;
+import org.springframework.context.Lifecycle;
 import org.springframework.expression.Expression;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.MessageTimeoutException;
@@ -74,7 +74,7 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Artem Bilan
  */
-public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements SmartLifecycle, MessageListener {
+public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements Lifecycle, MessageListener {
 
 	private volatile Destination requestDestination;
 
@@ -126,8 +126,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 
 	private final Object initializationMonitor = new Object();
 
-	private volatile boolean autoStartup;
-
 	private volatile boolean active;
 
 	private final AtomicLong correlationId = new AtomicLong();
@@ -484,20 +482,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 				session, replyDestinationName, this.replyPubSubDomain);
 	}
 
-	@Override
-	public int getPhase() {
-		return Integer.MAX_VALUE;
-	}
-
-	@Override
-	public boolean isAutoStartup() {
-		return this.autoStartup;
-	}
-
-	public void setAutoStartup(boolean autoStartup) {
-		this.autoStartup = autoStartup;
-	}
-
 	@Override
 	protected void doInit() {
 		synchronized (this.initializationMonitor) {
@@ -663,12 +647,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
 		return this.active;
 	}
 
-	@Override
-	public void stop(Runnable callback) {
-		this.stop();
-		callback.run();
-	}
-
 	@Override
 	protected Object handleRequestMessage(final Message<?> message) {
 		if (!this.initialized) {

spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java

@@ -0,0 +1,361 @@
+/*
+ * Copyright 2014 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.monitor;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.lessThan;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Executors;
+
+import javax.management.MBeanServer;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor;
+import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy;
+import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
+import org.springframework.integration.aggregator.MessageCountReleaseStrategy;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.BridgeHandler;
+import org.springframework.integration.handler.ScatterGatherHandler;
+import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
+import org.springframework.integration.router.RecipientListRouter;
+import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.jmx.support.MBeanServerFactoryBean;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Artem Bilan
+ * @since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class ScatterGatherHandlerIntegrationTests {
+
+	@Autowired
+	private PollableChannel output;
+
+	@Autowired
+	private MessageChannel inputAuctionWithoutGatherChannel;
+
+	@Autowired
+	private MessageChannel inputAuctionWithGatherChannel;
+
+	@Autowired
+	private MessageChannel distributionChannel;
+
+	@Test
+	public void testSimpleAuction() {
+		Message<String> quoteMessage = MessageBuilder.withPayload("testQuote").build();
+
+		this.inputAuctionWithoutGatherChannel.send(quoteMessage);
+		Message<?> bestQuoteMessage = this.output.receive(10000);
+		assertNotNull(bestQuoteMessage);
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(Double.class));
+		assertThat((Double) payload, lessThan(10D));
+	}
+
+	@Test
+	public void testAuctionWithGatherChannel() {
+		Message<String> quoteMessage = MessageBuilder.withPayload("testQuote").build();
+
+		this.inputAuctionWithGatherChannel.send(quoteMessage);
+		Message<?> bestQuoteMessage = this.output.receive(10000);
+		assertNotNull(bestQuoteMessage);
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(List.class));
+		assertEquals(3, ((List) payload).size());
+	}
+
+	@Test
+	public void testDistribution() {
+		Message<String> quoteMessage = MessageBuilder.withPayload("testQuote").build();
+
+		this.distributionChannel.send(quoteMessage);
+		Message<?> bestQuoteMessage = this.output.receive(10000);
+		assertNotNull(bestQuoteMessage);
+		Object payload = bestQuoteMessage.getPayload();
+		assertThat(payload, instanceOf(Double.class));
+		assertThat((Double) payload, lessThan(10D));
+	}
+
+
+	@Configuration
+	@EnableIntegration
+	@EnableIntegrationMBeanExport(server = "mBeanServer")
+	public static class ContextConfiguration {
+
+		@Bean
+		public static MBeanServerFactoryBean mBeanServer() {
+			return new MBeanServerFactoryBean();
+		}
+
+		@Bean
+		public PollableChannel output() {
+			return new QueueChannel();
+		}
+
+		@Bean
+		public SubscribableChannel scatterAuctionWithoutGatherChannel() {
+			PublishSubscribeChannel channel = new PublishSubscribeChannel();
+			channel.setApplySequence(true);
+			return channel;
+		}
+
+		@Bean
+		public MessageHandler gatherer1() {
+			return new AggregatingMessageHandler(
+					new ExpressionEvaluatingMessageGroupProcessor("^[payload gt 5] ?: -1D"),
+					new SimpleMessageStore(),
+					new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID),
+					new ExpressionEvaluatingReleaseStrategy("size() == 2"));
+		}
+
+		@Bean
+		public MessageChannel inputAuctionWithoutGatherChannel() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "inputAuctionWithoutGatherChannel")
+		public MessageHandler scatterGatherAuctionWithoutGatherChannel() {
+			ScatterGatherHandler handler = new ScatterGatherHandler(scatterAuctionWithoutGatherChannel(), gatherer1());
+			handler.setOutputChannel(output());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithoutGatherChannel")
+		public MessageHandler auctionWithoutGatherChannelBridge1() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel1());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithoutGatherChannel")
+		public MessageHandler auctionWithoutGatherChannelBridge2() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel1());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithoutGatherChannel")
+		public MessageHandler auctionWithoutGatherChannelBridge3() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel1());
+			return handler;
+		}
+
+		@Bean
+		public MessageChannel serviceChannel1() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "serviceChannel1")
+		public MessageHandler service1() {
+			return new AbstractReplyProducingMessageHandler() {
+
+				@Override
+				protected Object handleRequestMessage(Message<?> requestMessage) {
+					return Math.random() * 10;
+				}
+
+			};
+		}
+
+		@Bean
+		public MessageHandler distributor() {
+			RecipientListRouter router = new RecipientListRouter();
+			router.setApplySequence(true);
+			router.setChannels(Arrays.asList(distributionChannel1(), distributionChannel2(), distributionChannel3()));
+			return router;
+		}
+
+		@Bean
+		public MessageChannel distributionChannel1() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public MessageChannel distributionChannel2() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public MessageChannel distributionChannel3() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public MessageChannel distributionChannel() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "distributionChannel")
+		public MessageHandler scatterGatherDistribution() {
+			ScatterGatherHandler handler = new ScatterGatherHandler(distributor(), gatherer1());
+			handler.setOutputChannel(output());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "distributionChannel1")
+		public MessageHandler distributionBridge1() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel1());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "distributionChannel2")
+		public MessageHandler distributionBridge2() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel1());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "distributionChannel3")
+		public MessageHandler distributionBridge3() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel1());
+			return handler;
+		}
+
+		@Bean
+		public MessageChannel inputAuctionWithGatherChannel() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public MessageChannel gatherChannel() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		public SubscribableChannel scatterAuctionWithGatherChannel() {
+			PublishSubscribeChannel channel = new PublishSubscribeChannel(Executors.newCachedThreadPool());
+			channel.setApplySequence(true);
+			return channel;
+		}
+
+		@Bean
+		public MessageHandler gatherer2() {
+			return new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(),
+					new SimpleMessageStore(),
+					new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID),
+					new MessageCountReleaseStrategy(3));
+		}
+
+
+		@Bean
+		@ServiceActivator(inputChannel = "inputAuctionWithGatherChannel")
+		public MessageHandler scatterGatherAuctionWithGatherChannel() {
+			ScatterGatherHandler handler = new ScatterGatherHandler(scatterAuctionWithGatherChannel(), gatherer2());
+			handler.setGatherChannel(gatherChannel());
+			handler.setOutputChannel(output());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel")
+		public MessageHandler auctionWithGatherChannelBridge1() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel2());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel")
+		public MessageHandler auctionWithGatherChannelBridge2() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel2());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel")
+		public MessageHandler auctionWithGatherChannelBridge3() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel2());
+			return handler;
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "scatterAuctionWithGatherChannel")
+		public MessageHandler auctionWithGatherChannelBridge4() {
+			BridgeHandler handler = new BridgeHandler();
+			handler.setOutputChannel(serviceChannel2());
+			return handler;
+		}
+
+		@Bean
+		public MessageChannel serviceChannel2() {
+			return new DirectChannel();
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "serviceChannel2")
+		public MessageHandler service2() {
+			return new AbstractReplyProducingMessageHandler() {
+
+				{
+					setOutputChannel(gatherChannel());
+				}
+
+				@Override
+				protected Object handleRequestMessage(Message<?> requestMessage) {
+					return Math.random();
+				}
+
+			};
+		}
+
+	}
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java

@@ -57,8 +57,6 @@ public final class MqttParserUtils {
 			}
 		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "converter");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
 	}
 

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

@@ -16,7 +16,7 @@
 
 package org.springframework.integration.mqtt.outbound;
 
-import org.springframework.context.SmartLifecycle;
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
 import org.springframework.integration.mqtt.support.MqttHeaders;
@@ -32,7 +32,7 @@ import org.springframework.util.Assert;
  * @since 4.0
  *
  */
-public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements SmartLifecycle {
+public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements Lifecycle {
 
 	private final String url;
 
@@ -48,10 +48,6 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 
 	private boolean running;
 
-	private volatile int phase;
-
-	private volatile boolean autoStartup;
-
 	private volatile int clientInstance;
 
 	public AbstractMqttMessageHandler(String url, String clientId) {
@@ -118,6 +114,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 	@Override
 	public final void start() {
 		this.doStart();
+		this.running = true;
 	}
 
 	protected abstract void doStart();
@@ -125,6 +122,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 	@Override
 	public final void stop() {
 		this.doStop();
+		this.running = false;
 	}
 
 	protected abstract void doStop();
@@ -134,30 +132,6 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 		return this.running;
 	}
 
-	@Override
-	public int getPhase() {
-		return this.phase;
-	}
-
-	public void setPhase(int phase) {
-		this.phase = phase;
-	}
-
-	public void setAutoStartup(boolean autoStartup) {
-		this.autoStartup = autoStartup;
-	}
-
-	@Override
-	public boolean isAutoStartup() {
-		return this.autoStartup;
-	}
-
-	@Override
-	public void stop(Runnable callback) {
-		this.stop();
-		callback.run();
-	}
-
 	@Override
 	protected void handleMessageInternal(Message<?> message) throws Exception {
 		this.connectIfNeeded();

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java

@@ -68,8 +68,6 @@ public class MqttOutboundChannelAdapterParserTests {
 	@Test
 	public void testWithConverter() throws Exception {
 		assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(withConverterHandler, "url"));
-		assertFalse(TestUtils.getPropertyValue(withConverterHandler, "autoStartup", Boolean.class));
-		assertEquals(25, TestUtils.getPropertyValue(withConverterHandler, "phase"));
 		assertEquals("foo", TestUtils.getPropertyValue(withConverterHandler, "clientId"));
 		assertEquals("bar", TestUtils.getPropertyValue(withConverterHandler, "defaultTopic"));
 		assertSame(converter, TestUtils.getPropertyValue(withConverterHandler, "converter"));
@@ -90,8 +88,6 @@ public class MqttOutboundChannelAdapterParserTests {
 	@Test
 	public void testWithDefaultConverter() {
 		assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(withDefaultConverterHandler, "url"));
-		assertFalse(TestUtils.getPropertyValue(withDefaultConverterHandler, "autoStartup", Boolean.class));
-		assertEquals(25, TestUtils.getPropertyValue(withDefaultConverterHandler, "phase"));
 		assertEquals("foo", TestUtils.getPropertyValue(withDefaultConverterHandler, "clientId"));
 		assertEquals("bar", TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultTopic"));
 		assertEquals(1, TestUtils.getPropertyValue(withDefaultConverterHandler, "defaultQos"));

src/reference/docbook/message-routing.xml

@@ -10,5 +10,6 @@
    <xi:include href="./aggregator.xml"/>
    <xi:include href="./resequencer.xml"/>
    <xi:include href="./chain.xml"/>
+   <xi:include href="./scatter-gather.xml"/>
 
 </chapter>

src/reference/docbook/scatter-gather.xml

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<section xmlns="http://docbook.org/ns/docbook" version="5.0" xml:id="scatter-gather">
+	<title>Scatter-Gather</title>
+
+	<section id="scatter-gather-introduction">
+		<title>Introduction</title>
+		<para>
+			TBD
+		</para>
+	</section>
+
+	<section id="scatter-gather-functionality">
+		<title>Functionality</title>
+		<para>
+			TBD
+		</para>
+	</section>
+
+	<section id="scatter-gather-namespace">
+		<title>Configuring a Scatter-Gather</title>
+		<para>
+			TBD
+		</para>
+	</section>
+
+</section>

src/reference/docbook/whats-new.xml

@@ -27,6 +27,13 @@
 				See <xref linkend="web-sockets"/> for more information.
 			</para>
 		</section>
+		<section id="4.1-scatter-gather">
+			<title>Scatter-Gather EIP pattern</title>
+			<para>
+				The <emphasis>Scatter-Gather</emphasis> EIP pattern is now implemented.
+				See <xref linkend="scatter-gather"/> for more information.
+			</para>
+		</section>
 		<section id="4.1-BoonJsonObjectMapper">
 			<title>BoonJsonObjectMapper</title>
 			<para>