diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index eb2e0b37c6..8a7581bc7c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -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; } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java index 7759ee32f2..014813ce1e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java +++ b/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 implements FactoryBean, BeanFactoryAware { @@ -124,6 +125,12 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean { @@ -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) { } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java index 06c48e4c5c..656558d80f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/FilterFactoryBean.java +++ b/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); } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java index d8dc8937c6..e397edc05c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java +++ b/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(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java index 229da3e767..0079165a0f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ServiceActivatorFactoryBean.java +++ b/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."); + } + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java index e6762db259..04b23a88c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java +++ b/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 } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java index 9590700005..414c99daff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java +++ b/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 } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java index 060ec3a9db..c93d986bb8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/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() diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java new file mode 100644 index 0000000000..70f222a9d8 --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 97f99ff447..fd78644178 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/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 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<MessageChannel> 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 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 { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest-context.xml deleted file mode 100644 index 6b65e31ab3..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest-context.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - - - - - - - - - - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests-context.xml new file mode 100644 index 0000000000..4b5482f7b2 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests-context.xml @@ -0,0 +1,43 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests.java index 3d84340bd1..1318073b03 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTests.java +++ b/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()); + } + }