INT-3460: Add AbstractMessageProducingHandler

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

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

@@ -37,10 +37,8 @@ import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.NullChannel;
-import org.springframework.integration.core.MessageProducer;
-import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.expression.IntegrationEvaluationContextAware;
-import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback;
@@ -79,11 +77,11 @@ import org.springframework.util.CollectionUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author David Liu
  * @since 2.0
  */
-public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler
-		implements MessageProducer, DisposableBean, IntegrationEvaluationContextAware,
-		ApplicationEventPublisherAware {
+public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler
+		implements DisposableBean, IntegrationEvaluationContextAware, ApplicationEventPublisherAware {
 
 	private static final Log logger = LogFactory.getLog(AbstractCorrelatingMessageHandler.class);
 
@@ -99,12 +97,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 
 	private volatile ReleaseStrategy releaseStrategy;
 
-	private MessageChannel outputChannel;
-
-	private String outputChannelName;
-
-	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
-
 	private volatile MessageChannel discardChannel;
 
 	private volatile String discardChannelName;
@@ -179,17 +171,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
 	}
 
-	@Override
-	public void setOutputChannel(MessageChannel outputChannel) {
-		Assert.notNull(outputChannel, "'outputChannel' must not be null");
-		this.outputChannel = outputChannel;
-	}
-
-	public void setOutputChannelName(String outputChannelName) {
-		Assert.hasText(outputChannelName, "'outputChannelName' must not be empty");
-		this.outputChannelName = outputChannelName;
-	}
-
 	public void setGroupTimeoutExpression(Expression groupTimeoutExpression) {
 		this.groupTimeoutExpression = groupTimeoutExpression;
 	}
@@ -218,7 +199,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 			Assert.state(!(this.discardChannelName != null && this.discardChannel != null),
 					"'discardChannelName' and 'discardChannel' are mutually exclusive.");
 
-			Assert.state(!(this.outputChannelName != null && this.outputChannel != null),
+			Assert.state(!(getOutputChannelName() != null && getOutputChannel() != null),
 					"'outputChannelName' and 'outputChannel' are mutually exclusive.");
 
 			if (this.outputProcessor instanceof BeanFactoryAware) {
@@ -260,9 +241,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		this.discardChannelName = discardChannelName;
 	}
 
-	public void setSendTimeout(long sendTimeout) {
-		this.messagingTemplate.setSendTimeout(sendTimeout);
-	}
 
 	public void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) {
 		this.sendPartialResultOnExpiry = sendPartialResultOnExpiry;
@@ -323,18 +301,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		return releaseStrategy;
 	}
 
-	protected MessageChannel getOutputChannel() {
-		return outputChannel;
-	}
-
-	protected String getOutputChannelName() {
-		return outputChannelName;
-	}
-
-	protected MessagingTemplate getMessagingTemplate() {
-		return messagingTemplate;
-	}
-
 	protected MessageChannel getDiscardChannel() {
 		return discardChannel;
 	}
@@ -594,7 +560,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 			if (logger.isDebugEnabled()) {
 				logger.debug("Prematurely releasing partially complete group with key ["
 						+ correlationKey + "] to: "
-						+ (this.outputChannelName != null ? this.outputChannelName : this.outputChannel));
+						+ (getOutputChannelName() != null ? getOutputChannelName() : getOutputChannel()));
 			}
 			completeGroup(correlationKey, group);
 		}
@@ -649,24 +615,23 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 		if (message != null) {
 			replyChannelHeader = message.getHeaders().getReplyChannel();
 		}
-
-		if (this.outputChannelName != null) {
+		if (getOutputChannelName() != null) {
 			synchronized (this) {
-				if (this.outputChannelName != null) {
+				if (getOutputChannelName() != null) {
 					try {
-						this.outputChannel = getBeanFactory().getBean(this.outputChannelName, MessageChannel.class);
-						this.outputChannelName = null;
+						setOutputChannel(getBeanFactory().getBean(getOutputChannelName(), MessageChannel.class));
+						setOutputChannelName(null);
 					}
 					catch (BeansException e) {
 						throw new DestinationResolutionException("Failed to look up MessageChannel with name '"
-								+ this.outputChannelName + "' in the BeanFactory.");
+								+ getOutputChannelName() + "' in the BeanFactory.");
 					}
 				}
 			}
 		}
 
-		Object replyChannel = this.outputChannel;
-		if (this.outputChannel == null) {
+		Object replyChannel = getOutputChannel();
+		if (replyChannel == null) {
 			replyChannel = replyChannelHeader;
 		}
 		Assert.notNull(replyChannel, "no outputChannel or replyChannel header available");
@@ -753,6 +718,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
 			}
 			return false;
 		}
+
 	}
 
 }

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

@@ -39,6 +39,7 @@ import org.springframework.util.CollectionUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author David Liu
  */
 public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageHandler>
 		implements FactoryBean<MessageHandler>, BeanFactoryAware {
@@ -124,6 +125,12 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean<H extends MessageH
 					this.handler instanceof AbstractReplyProducingMessageHandler) {
 				((AbstractReplyProducingMessageHandler) this.handler).setAdviceChain(this.adviceChain);
 			}
+			else if(!(this.handler instanceof AbstractReplyProducingMessageHandler)) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("adviceChain can only be set to AbstractReplyProducingMessageHandler or its subclass, "
+					+ ((IntegrationObjectSupport) this.handler).getComponentName() + " doesn't support it.");
+				}
+			}
 			if (this.handler instanceof Orderable && this.order != null) {
 				((Orderable) this.handler).setOrder(this.order);
 			}

spring-integration-core/src/main/java/org/springframework/integration/config/AbstractStandardMessageHandlerFactoryBean.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. You may obtain a copy of the License at
@@ -22,6 +22,7 @@ import org.springframework.expression.Expression;
 import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.SpelParserConfiguration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.messaging.MessageHandler;
@@ -35,6 +36,7 @@ import org.springframework.util.StringUtils;
  * @author Alexander Peters
  * @author Gary Russell
  * @author Artem Bilan
+ * @author David Liu
  */
 abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean<MessageHandler> {
 
@@ -75,8 +77,8 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 		if (this.targetObject != null) {
 			Assert.state(this.expression == null,
 					"The 'targetObject' and 'expression' properties are mutually exclusive.");
-			AbstractReplyProducingMessageHandler actualHandler = this.extractTypeIfPossible(targetObject,
-										AbstractReplyProducingMessageHandler.class);
+			AbstractMessageProducingHandler actualHandler = this.extractTypeIfPossible(this.targetObject,
+					AbstractMessageProducingHandler.class);
 			boolean targetIsDirectReplyProducingHandler = actualHandler != null
 							&& this.canBeUsedDirect(actualHandler) // give subclasses a say
 							&& this.methodIsHandleMessageOrEmpty(this.targetMethodName);
@@ -85,7 +87,7 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 			}
 			else if (targetIsDirectReplyProducingHandler) {
 				if (logger.isDebugEnabled()) {
-					logger.debug("Wiring handler (" + targetObject + ") directly into endpoint");
+					logger.debug("Wiring handler (" + this.targetObject + ") directly into endpoint");
 				}
 				this.checkReuse(actualHandler);
 				this.postProcessReplyProducer(actualHandler);
@@ -117,7 +119,7 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 		}
 	}
 
-	private void checkReuse(AbstractReplyProducingMessageHandler replyHandler) {
+	private void checkReuse(AbstractMessageProducingHandler replyHandler) {
 		Assert.isTrue(!referencedReplyProducers.contains(replyHandler),
 				"An AbstractReplyProducingMessageHandler may only be referenced once (" +
 				replyHandler.getComponentName() + ") - use scope=\"prototype\"");
@@ -168,11 +170,11 @@ abstract class AbstractStandardMessageHandlerFactoryBean extends AbstractSimpleM
 				|| "handleMessage".equals(targetMethodName));
 	}
 
-	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+	protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
 		return false;
 	}
 
-	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.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.
@@ -17,13 +17,14 @@
 package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessageHandler;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
 import org.springframework.integration.filter.MessageFilter;
 import org.springframework.integration.filter.MethodInvokingSelector;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -32,6 +33,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author David Liu
  * @since 2.0
  */
 public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
@@ -110,9 +112,9 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 
 
 	@Override
-	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 		if (this.sendTimeout != null) {
-			handler.setSendTimeout(this.sendTimeout.longValue());
+			handler.setSendTimeout(this.sendTimeout);
 		}
 		if (!(handler instanceof MessageFilter)) {
 			Assert.isNull(this.throwExceptionOnRejection, "Cannot set throwExceptionOnRejection if the referenced bean is "
@@ -132,12 +134,11 @@ public class FilterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 	 * MessageSelector, MesageSelector wins and gets wrapped in a MessageFilter.
 	 */
 	@Override
-	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+	protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
 		return handler instanceof MessageFilter
 				|| (!(handler instanceof MessageSelector)
 						&& this.discardChannel == null && this.throwExceptionOnRejection == null
 						&& this.discardWithinAdvice == null);
 	}
 
-
 }

spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.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. You may obtain a copy of the License at
@@ -16,7 +16,7 @@ package org.springframework.integration.config;
 import java.util.Map;
 
 import org.springframework.expression.Expression;
-import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.router.AbstractMappingMessageRouter;
 import org.springframework.integration.router.AbstractMessageRouter;
 import org.springframework.integration.router.ExpressionEvaluatingRouter;
@@ -34,6 +34,7 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @author Dave Syer
  * @author Gary Russell
+ * @author David Liu
  */
 public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
 
@@ -102,10 +103,9 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 	}
 
 	private AbstractMappingMessageRouter createMethodInvokingRouter(Object targetObject, String targetMethodName) {
-		MethodInvokingRouter router = (StringUtils.hasText(targetMethodName))
+		return (StringUtils.hasText(targetMethodName))
 				? new MethodInvokingRouter(targetObject, targetMethodName)
 				: new MethodInvokingRouter(targetObject);
-		return router;
 	}
 
 	private AbstractMessageRouter configureRouter(AbstractMessageRouter router) {
@@ -113,7 +113,7 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 			router.setDefaultOutputChannel(this.defaultOutputChannel);
 		}
 		if (this.timeout != null) {
-			router.setTimeout(timeout.longValue());
+			router.setTimeout(this.timeout);
 		}
 		if (this.applySequence != null) {
 			router.setApplySequence(this.applySequence);
@@ -137,7 +137,7 @@ public class RouterFactoryBean extends AbstractStandardMessageHandlerFactoryBean
 	}
 
 	@Override
-	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+	protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
 		return noRouterAttributesProvided();
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.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.
@@ -17,6 +17,7 @@
 package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.handler.MessageProcessor;
@@ -30,6 +31,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author David Liu
  * @since 2.0
  */
 public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
@@ -61,14 +63,14 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 
 	/**
 	 * If the target object is a {@link MessageHandler} and the method is 'handleMessage', return an
-	 * {@link AbstractReplyProducingMessageHandler} that wraps it.
+	 * {@link AbstractMessageProducingHandler} that wraps it.
 	 */
 	private MessageHandler createDirectHandlerIfPossible(final Object targetObject, String targetMethodName) {
 		MessageHandler handler = null;
 		if (targetObject instanceof MessageHandler
 				&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
-			if (targetObject instanceof AbstractReplyProducingMessageHandler) {
-				// should never happen but just return it if it's already an ARPMH
+			if (targetObject instanceof AbstractMessageProducingHandler) {
+				// should never happen but just return it if it's already an AMPH
 				return (MessageHandler) targetObject;
 			}
 			/*
@@ -108,21 +110,29 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 
 
 	/**
-	 * Always returns true - any {@link AbstractReplyProducingMessageHandler} can
+	 * Always returns true - any {@link AbstractMessageProducingHandler} can
 	 * be used directly.
 	 */
 	@Override
-	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+	protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
 		return true;
 	}
 
 	@Override
-	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 		if (this.sendTimeout != null) {
 			handler.setSendTimeout(this.sendTimeout);
 		}
 		if (this.requiresReply != null) {
-			handler.setRequiresReply(this.requiresReply);
+			if(handler instanceof AbstractReplyProducingMessageHandler) {
+				((AbstractReplyProducingMessageHandler) handler).setRequiresReply(this.requiresReply);
+			}
+			else {
+				if (this.requiresReply && logger.isDebugEnabled()) {
+					logger.debug("requires-reply can only be set to AbstractReplyProducingMessageHandler or its subclass, "
+					+ handler.getComponentName() + " doesn't support it.");
+				}
+			}
 		}
 	}
 

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.splitter.AbstractMessageSplitter;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
@@ -32,6 +33,7 @@ import org.springframework.util.StringUtils;
  * @author Mark Fisher
  * @author Iwein Fuld
  * @author Gary Russell
+ * @author David Liu
  */
 public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
 
@@ -106,18 +108,24 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 	}
 
 	@Override
-	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
+	protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
 		return handler instanceof AbstractMessageSplitter
 				|| (this.applySequence == null && this.delimiters == null);
 	}
 
 	@Override
-	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 		if (this.sendTimeout != null) {
 			handler.setSendTimeout(sendTimeout);
 		}
 		if (this.requiresReply != null) {
-			handler.setRequiresReply(requiresReply);
+			if(handler instanceof AbstractReplyProducingMessageHandler) {
+				((AbstractReplyProducingMessageHandler) handler).setRequiresReply(this.requiresReply);
+			}
+			else if (this.requiresReply && logger.isDebugEnabled()) {
+			      logger.debug("requires-reply can only be set to AbstractReplyProducingMessageHandler or its subclass, "
+			                     + handler.getComponentName() + " doesn't support it.");
+			 }
 		}
 		if (!(handler instanceof AbstractMessageSplitter)) {
 			Assert.isNull(this.applySequence, "Cannot set applySequence if the referenced bean is "
@@ -138,5 +146,4 @@ public class SplitterFactoryBean extends AbstractStandardMessageHandlerFactoryBe
 		}
 	}
 
-
 }

spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.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.
@@ -17,7 +17,7 @@
 package org.springframework.integration.config;
 
 import org.springframework.expression.Expression;
-import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
 import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
@@ -31,6 +31,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author David Liu
  */
 public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactoryBean {
 
@@ -72,20 +73,19 @@ public class TransformerFactoryBean extends AbstractStandardMessageHandlerFactor
 	}
 
 	@Override
-	protected void postProcessReplyProducer(AbstractReplyProducingMessageHandler handler) {
+	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 		if (this.sendTimeout != null) {
-			handler.setSendTimeout(this.sendTimeout.longValue());
+			handler.setSendTimeout(this.sendTimeout);
 		}
 	}
 
 	/**
-	 * Always returns true - any {@link AbstractReplyProducingMessageHandler} can
+	 * Always returns true - any {@link AbstractMessageProducingHandler} can
 	 * be used directly.
 	 */
 	@Override
-	protected boolean canBeUsedDirect(AbstractReplyProducingMessageHandler handler) {
-		return true; // Any ARPMH can be a transformer
+	protected boolean canBeUsedDirect(AbstractMessageProducingHandler handler) {
+		return true; // Any AMPH can be a transformer
 	}
 
-
 }

spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java

@@ -39,6 +39,7 @@ import org.springframework.util.Assert;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author David Liu
  */
 public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHandler {
 
@@ -150,7 +151,7 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa
 				}
 			}
 			if (this.discardChannel != null) {
-				this.getMessagingTemplate().send(this.discardChannel, message);
+				this.messagingTemplate.send(this.discardChannel, message);
 			}
 			if (this.throwExceptionOnRejection) {
 				throw new MessageRejectedException(message, "MessageFilter '" + this.getComponentName()

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

@@ -0,0 +1,63 @@
+/*
+ * 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.integration.core.MessageProducer;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}.
+ *
+ * @author David Liu
+ * since 4.1
+ */
+public abstract class AbstractMessageProducingHandler extends AbstractMessageHandler
+		implements MessageProducer {
+
+	protected final MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+	private MessageChannel outputChannel;
+
+	private String outputChannelName;
+
+	/**
+	 * Set the timeout for sending reply Messages.
+	 * @param sendTimeout The send timeout.
+	 */
+	public void setSendTimeout(long sendTimeout) {
+		this.messagingTemplate.setSendTimeout(sendTimeout);
+	}
+
+	@Override
+	public void setOutputChannel(MessageChannel outputChannel) {
+		this.outputChannel = outputChannel;
+	}
+
+	public void setOutputChannelName(String outputChannelName) {
+		this.outputChannelName = outputChannelName;
+	}
+
+	public MessageChannel getOutputChannel() {
+		return outputChannel;
+	}
+
+	public String getOutputChannelName() {
+		return outputChannelName;
+	}
+
+}

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

@@ -23,8 +23,6 @@ import org.aopalliance.aop.Advice;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanClassLoaderAware;
-import org.springframework.integration.core.MessageProducer;
-import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -44,17 +42,10 @@ import org.springframework.util.CollectionUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author David Liu
  */
-public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler
-		implements MessageProducer, BeanClassLoaderAware {
-
-	private MessageChannel outputChannel;
-
-	private String outputChannelName;
-
-	private volatile boolean requiresReply = false;
-
-	private final MessagingTemplate messagingTemplate;
+public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageProducingHandler
+		implements BeanClassLoaderAware {
 
 	private volatile RequestHandler advisedRequestHandler;
 
@@ -62,39 +53,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 
 	private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
 
-
-
-	public AbstractReplyProducingMessageHandler() {
-		this.messagingTemplate = new MessagingTemplate();
-	}
-
-
-	@Override
-	public void setOutputChannel(MessageChannel outputChannel) {
-		this.outputChannel = outputChannel;
-	}
-
-	public void setOutputChannelName(String outputChannelName) {
-		Assert.hasText(outputChannelName, "'outputChannelName' must not be empty");
-		this.outputChannelName = outputChannelName;
-	}
-
-	/**
-	 * Set the timeout for sending reply Messages.
-	 * @param sendTimeout The send timeout.
-	 */
-	public void setSendTimeout(long sendTimeout) {
-		this.messagingTemplate.setSendTimeout(sendTimeout);
-	}
-
-	/**
-	 * Set the DestinationResolver<MessageChannel> to be used when there is no default output channel.
-	 * @param channelResolver The channel resolver.
-	 */
-	public void setChannelResolver(DestinationResolver<MessageChannel> channelResolver) {
-		Assert.notNull(channelResolver, "'channelResolver' must not be null");
-		this.messagingTemplate.setDestinationResolver(channelResolver);
-	}
+	private volatile boolean requiresReply = false;
 
 	/**
 	 * Flag whether a reply is required. If true an incoming message MUST result in a reply message being sent.
@@ -106,11 +65,12 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 	}
 
 	/**
-	 * Provides access to the {@link MessagingTemplate} for subclasses.
-	 * @return The messaging template.
+	 * Set the DestinationResolver&lt;MessageChannel&gt; to be used when there is no default output channel.
+	 * @param channelResolver The channel resolver.
 	 */
-	protected MessagingTemplate getMessagingTemplate() {
-		return this.messagingTemplate;
+	public void setChannelResolver(DestinationResolver<MessageChannel> channelResolver) {
+		Assert.notNull(channelResolver, "'channelResolver' must not be null");
+		this.messagingTemplate.setDestinationResolver(channelResolver);
 	}
 
 
@@ -131,7 +91,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 
 	@Override
 	protected final void onInit() {
-		Assert.state(!(this.outputChannelName != null && this.outputChannel != null),
+		Assert.state(!(getOutputChannelName() != null && getOutputChannel() != null),
 				"'outputChannelName' and 'outputChannel' are mutually exclusive.");
 		if (this.getBeanFactory() != null) {
 			this.messagingTemplate.setBeanFactory(getBeanFactory());
@@ -225,24 +185,23 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 		if (logger.isDebugEnabled()) {
 			logger.debug("handler '" + this + "' sending reply Message: " + replyMessage);
 		}
-
-		if (this.outputChannelName != null) {
+		if (getOutputChannelName() != null) {
 			synchronized (this) {
-				if (this.outputChannelName != null) {
+				if (getOutputChannelName() != null) {
 					try {
-						this.outputChannel = this.getBeanFactory().getBean(this.outputChannelName, MessageChannel.class);
-						this.outputChannelName = null;
+						setOutputChannel(this.getBeanFactory().getBean(getOutputChannelName(), MessageChannel.class));
+						setOutputChannelName(null);
 					}
 					catch (BeansException e) {
 						throw new DestinationResolutionException("Failed to look up MessageChannel with name '"
-								+ this.outputChannelName + "' in the BeanFactory.");
+								+ getOutputChannelName() + "' in the BeanFactory.");
 					}
 				}
 			}
 		}
 
-		if (this.outputChannel != null) {
-			this.sendMessage(replyMessage, this.outputChannel);
+		if (getOutputChannel() != null) {
+			this.sendMessage(replyMessage, getOutputChannel());
 		}
 		else if (replyChannelHeaderValue != null) {
 			this.sendMessage(replyMessage, replyChannelHeaderValue);
@@ -305,6 +264,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 
 		@Override
 		String toString();
+
 	}
 
 	private class AdvisedRequestHandler implements RequestHandler {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest-context.xml

@@ -1,18 +0,0 @@
-<?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:resequencer id="resequencerLight" input-channel="resequencerLightInput" output-channel="outputChannel" 
-					 release-partial-sequences="true"/>
-	
-	<int:channel id="outputChannel">
-		<int:queue/>
-	</int:channel>
-	
-	<int:resequencer id="resequencerDeep" input-channel="resequencerDeepInput" output-channel="outputChannel" release-partial-sequences="true"/>
-
-</beans>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests-context.xml

@@ -0,0 +1,43 @@
+<?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:resequencer id="resequencerLight" input-channel="resequencerLightInput" output-channel="outputChannel"
+					 release-partial-sequences="true"/>
+
+	<int:channel id="outputChannel">
+		<int:queue/>
+	</int:channel>
+
+	<int:resequencer id="resequencerDeep" input-channel="resequencerDeepInput" output-channel="outputChannel"
+					 release-partial-sequences="true"/>
+
+	<int:service-activator id="customResequencer" ref="resequencer"
+						   input-channel="inputChannel" output-channel="outputChannel"/>
+
+	<bean id="resequencer" class="org.springframework.integration.aggregator.ResequencingMessageHandler">
+		<constructor-arg name="releaseStrategy" ref="releaseStrategy"/>
+		<constructor-arg name="processor" ref="customResequencerMessageGroupProcessor"/>
+		<constructor-arg name="store" ref="testMessageStore"/>
+		<constructor-arg name="correlationStrategy" ref="customCorrelationStrategy"/>
+		<property name="sendPartialResultOnExpiry" value="true"/>
+	</bean>
+
+	<bean id="customResequencerMessageGroupProcessor"
+		  class="org.springframework.integration.aggregator.ResequencingMessageGroupProcessor"/>
+
+	<bean id="releaseStrategy"
+		  class="org.springframework.integration.config.ResequencerParserTests.TestReleaseStrategy"/>
+
+	<bean id="customCorrelationStrategy"
+		  class="org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy">
+		<constructor-arg name="attributeName" value="correlationId"/>
+	</bean>
+
+	<bean id="testMessageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
+
+</beans>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 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.
@@ -16,11 +16,15 @@
 
 package org.springframework.integration.aggregator.integration;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
 import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.aggregator.ResequencingMessageHandler;
 import org.springframework.integration.channel.QueueChannel;
@@ -28,21 +32,28 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Oleg Zhurakousky
+ * @author David Liu
  */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+@DirtiesContext
 public class ResequencerIntegrationTests {
 
+	@Autowired
+	private ApplicationContext context;
+
 	@Test
-	public void validateUnboundedResequencerLight(){
-		ApplicationContext context = new ClassPathXmlApplicationContext("ResequencerIntegrationTest-context.xml",  ResequencerIntegrationTests.class);
-		MessageChannel inputChannel = context .getBean("resequencerLightInput", MessageChannel.class);
-		QueueChannel outputChannel = context .getBean("outputChannel", QueueChannel.class);
+	public void validateUnboundedResequencerLight() {
+		MessageChannel inputChannel = context.getBean("resequencerLightInput", MessageChannel.class);
+		QueueChannel outputChannel = context.getBean("outputChannel", QueueChannel.class);
 		EventDrivenConsumer edc = context.getBean("resequencerLight", EventDrivenConsumer.class);
 		ResequencingMessageHandler handler = TestUtils.getPropertyValue(edc, "handler", ResequencingMessageHandler.class);
 		MessageGroupStore store = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
@@ -60,15 +71,15 @@ public class ResequencerIntegrationTests {
 		inputChannel.send(message1);
 		message1 = outputChannel.receive(0);
 		assertNotNull(message1);
-		assertEquals((Integer)1, new IntegrationMessageHeaderAccessor(message1).getSequenceNumber());
+		assertEquals((Integer) 1, new IntegrationMessageHeaderAccessor(message1).getSequenceNumber());
 
 		inputChannel.send(message2);
 		message2 = outputChannel.receive(0);
 		message3 = outputChannel.receive(0);
 		assertNotNull(message2);
 		assertNotNull(message3);
-		assertEquals((Integer)2, new IntegrationMessageHeaderAccessor(message2).getSequenceNumber());
-		assertEquals((Integer)3, new IntegrationMessageHeaderAccessor(message3).getSequenceNumber());
+		assertEquals((Integer) 2, new IntegrationMessageHeaderAccessor(message2).getSequenceNumber());
+		assertEquals((Integer) 3, new IntegrationMessageHeaderAccessor(message3).getSequenceNumber());
 
 		inputChannel.send(message5);
 		assertNull(outputChannel.receive(0));
@@ -83,19 +94,18 @@ public class ResequencerIntegrationTests {
 		assertNotNull(message4);
 		assertNotNull(message5);
 		assertNotNull(message6);
-		assertEquals((Integer)4, new IntegrationMessageHeaderAccessor(message4).getSequenceNumber());
-		assertEquals((Integer)5, new IntegrationMessageHeaderAccessor(message5).getSequenceNumber());
-		assertEquals((Integer)6, new IntegrationMessageHeaderAccessor(message6).getSequenceNumber());
+		assertEquals((Integer) 4, new IntegrationMessageHeaderAccessor(message4).getSequenceNumber());
+		assertEquals((Integer) 5, new IntegrationMessageHeaderAccessor(message5).getSequenceNumber());
+		assertEquals((Integer) 6, new IntegrationMessageHeaderAccessor(message6).getSequenceNumber());
 
 
 		assertEquals(0, store.getMessageGroup("A").getMessages().size());
 	}
 
 	@Test
-	public void validateUnboundedResequencerDeep(){
-		ApplicationContext context = new ClassPathXmlApplicationContext("ResequencerIntegrationTest-context.xml",  ResequencerIntegrationTests.class);
-		MessageChannel inputChannel = context .getBean("resequencerDeepInput", MessageChannel.class);
-		QueueChannel outputChannel = context .getBean("outputChannel", QueueChannel.class);
+	public void validateUnboundedResequencerDeep() {
+		MessageChannel inputChannel = context.getBean("resequencerDeepInput", MessageChannel.class);
+		QueueChannel outputChannel = context.getBean("outputChannel", QueueChannel.class);
 		EventDrivenConsumer edc = context.getBean("resequencerDeep", EventDrivenConsumer.class);
 		ResequencingMessageHandler handler = TestUtils.getPropertyValue(edc, "handler", ResequencingMessageHandler.class);
 		MessageGroupStore store = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
@@ -113,4 +123,16 @@ public class ResequencerIntegrationTests {
 		assertNotNull(outputChannel.receive(0));
 		assertEquals(0, store.getMessageGroup("A").getMessages().size());
 	}
+
+	@Test
+	public void testResequencerRefServiceActivator() {
+		MessageChannel inputChannel = context.getBean("inputChannel", MessageChannel.class);
+		QueueChannel outputChannel = context.getBean("outputChannel", QueueChannel.class);
+		Message<?> message1 = MessageBuilder.withPayload("1").setCorrelationId("A").setSequenceNumber(1).build();
+		inputChannel.send(message1);
+		message1 = outputChannel.receive(0);
+		assertNotNull(message1);
+		assertEquals((Integer) 1, new IntegrationMessageHeaderAccessor(message1).getSequenceNumber());
+	}
+
 }