diff --git a/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers b/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers index ac1e1f6beb..d282842c05 100644 --- a/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers +++ b/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers @@ -7,6 +7,7 @@ httpinvoker-source=org.springframework.integration.adapter.httpinvoker.config.Ht httpinvoker-target=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerTargetAdapterParser jms-source=org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser jms-target=org.springframework.integration.adapter.jms.config.JmsTargetParser +jms-gateway=org.springframework.integration.adapter.jms.config.JmsGatewayParser mail-target=org.springframework.integration.adapter.mail.config.MailTargetParser rmi-source=org.springframework.integration.adapter.rmi.config.RmiSourceAdapterParser rmi-target=org.springframework.integration.adapter.rmi.config.RmiTargetAdapterParser \ No newline at end of file diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java index 4b9dbc34d3..8edd41db2b 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java @@ -37,16 +37,12 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin private final Log logger = LogFactory.getLog(this.getClass()); - private final MessageChannel channel; + private final MessageChannel requestChannel; - private volatile RequestReplyTemplate requestReplyTemplate; + private final RequestReplyTemplate requestReplyTemplate = new RequestReplyTemplate(); private volatile boolean expectReply = true; - private volatile long sendTimeout = -1; - - private volatile long receiveTimeout = -1; - protected final Object lifecycleMonitor = new Object(); private volatile boolean initialized; @@ -55,12 +51,13 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin /** * Create an adapter that sends to the provided channel. * - * @param channel the channel where messages will be sent, must not be + * @param requestChannel the channel where messages will be sent, must not be * null. */ - public MessageHandlingSourceAdapter(MessageChannel channel) { - Assert.notNull(channel, "channel must not be null"); - this.channel = channel; + public MessageHandlingSourceAdapter(MessageChannel requestChannel) { + Assert.notNull(requestChannel, "request channel must not be null"); + this.requestChannel = requestChannel; + this.requestReplyTemplate.setRequestChannel(requestChannel); } @@ -72,16 +69,16 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin this.expectReply = expectReply; } - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; + public void setRequestTimeout(long requestTimeout) { + this.requestReplyTemplate.setRequestTimeout(requestTimeout); } - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; + public void setReplyTimeout(long replyTimeout) { + this.requestReplyTemplate.setReplyTimeout(replyTimeout); } protected MessageChannel getChannel() { - return this.channel; + return this.requestChannel; } public final void afterPropertiesSet() throws Exception { @@ -89,9 +86,6 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin if (this.initialized) { return; } - if (this.requestReplyTemplate == null) { - this.requestReplyTemplate = this.createRequestReplyTemplate(); - } } this.initialize(); this.initialized = true; @@ -103,13 +97,6 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin protected void initialize() throws Exception { } - private RequestReplyTemplate createRequestReplyTemplate() { - RequestReplyTemplate template = new RequestReplyTemplate(this.channel); - template.setRequestTimeout(this.sendTimeout); - template.setReplyTimeout(this.receiveTimeout); - return template; - } - public final Message handle(Message message) { if (!this.initialized) { try { @@ -120,9 +107,9 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin } } if (!this.expectReply) { - boolean sent = (this.sendTimeout < 0) ? this.channel.send(message) : this.channel.send(message, this.sendTimeout); + boolean sent = this.requestReplyTemplate.send(message); if (!sent && logger.isWarnEnabled()) { - logger.warn("failed to send message to channel within timeout of " + this.sendTimeout + " milliseconds"); + logger.warn("failed to send message to channel within timeout"); } return null; } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java index efab9e8cc6..df122c7437 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java @@ -50,24 +50,24 @@ public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSi @Override protected boolean isEligibleAttribute(String attributeName) { - return !attributeName.equals("name") && !attributeName.equals("channel") && super.isEligibleAttribute(attributeName); + return !attributeName.equals("name") && !attributeName.equals("request-channel") && super.isEligibleAttribute(attributeName); } @Override protected void postProcess(BeanDefinitionBuilder builder, Element element) { - String channelRef = element.getAttribute("channel"); + String channelRef = element.getAttribute("request-channel"); if (!StringUtils.hasText(channelRef)) { - throw new ConfigurationException("a 'channel' reference is required"); + throw new ConfigurationException("a 'request-channel' reference is required"); } builder.addConstructorArgReference(channelRef); builder.addPropertyValue("expectReply", element.getAttribute("expect-reply").equals("true")); - String sendTimeout = element.getAttribute("send-timeout"); - if (StringUtils.hasText(sendTimeout)) { - builder.addPropertyValue("sendTimeout", Long.parseLong(sendTimeout)); + String requestTimeout = element.getAttribute("request-timeout"); + if (StringUtils.hasText(requestTimeout)) { + builder.addPropertyValue("requestTimeout", Long.parseLong(requestTimeout)); } - String receiveTimeout = element.getAttribute("receive-timeout"); - if (StringUtils.hasText(receiveTimeout)) { - builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout)); + String replyTimeout = element.getAttribute("reply-timeout"); + if (StringUtils.hasText(replyTimeout)) { + builder.addPropertyValue("replyTimeout", Long.parseLong(replyTimeout)); } this.doPostProcess(builder, element); } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd index 6cab322f6d..a0cb69d78d 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd @@ -58,54 +58,83 @@ - + + + + Defines a JMS-based source channel adapter. + + + + + + + + Defines a JMS-based gateway adapter. + + - - - Defines a jms-based source channel adapter. - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + - - - - + + + Defines a target that sends JMS Messages. - - - - - - - - + + + + + + Common configuration for inbound JMS-based adapters. + + + + + + + + + + + + + + + + + + + + + + + + + Common configuration for JMS-based adapters. + + + + + + + + + @@ -114,7 +143,7 @@ - + @@ -138,7 +167,7 @@ - + Defines an httpinvoker-based source channel adapter. @@ -201,18 +230,19 @@ - + - Defines common configuration for request-reply source adapters. + Defines common configuration for gateway adapters. - - - + + + + \ No newline at end of file diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java index c2d5a25163..c1dcf6a565 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -23,7 +23,7 @@ import java.io.FilenameFilter; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.util.Assert; /** @@ -31,7 +31,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class FileSource implements PollableSource, InitializingBean { +public class FileSource implements Source, InitializingBean { private final File directory; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java index 58a6a9fc97..f6aff0eef2 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java @@ -35,7 +35,7 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageCreator; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -45,7 +45,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Mark Fisher */ -public class FtpSource implements PollableSource, MessageDeliveryAware { +public class FtpSource implements Source, MessageDeliveryAware { private final static String DEFAULT_HOST = "localhost"; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java index a2ad6d9ed2..aa0c83a11c 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java @@ -25,10 +25,11 @@ import org.springframework.context.Lifecycle; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.SubscribableSource; -import org.springframework.integration.message.Target; +import org.springframework.integration.channel.RequestReplyTemplate; +import org.springframework.integration.gateway.MessagingGateway; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.adapter.MessageListenerAdapter; import org.springframework.jms.support.converter.MessageConverter; import org.springframework.jms.support.converter.SimpleMessageConverter; import org.springframework.util.Assert; @@ -38,9 +39,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecycle, DisposableBean { - - private volatile MessageChannel channel; +public class JmsMessageDrivenSourceAdapter extends MessagingGateway implements Lifecycle, DisposableBean { private volatile AbstractMessageListenerContainer container; @@ -58,8 +57,6 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy private volatile int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE; - private volatile long receiveTimeout = 1000; - private volatile int concurrentConsumers = 1; private volatile int maxConcurrentConsumers = 1; @@ -68,37 +65,8 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy private volatile int idleTaskExecutionLimit = 1; - private volatile long sendTimeout = -1; + private boolean expectReply = false; - private volatile boolean initialized; - - private final Object lifecycleMonitor = new Object(); - - - public boolean subscribe(Target target) { - if (target instanceof MessageChannel) { - this.setChannel((MessageChannel) target); - return true; - } - return false; - } - - public boolean unsubscribe(Target target) { - if (target.equals(this.channel)) { - this.stop(); - this.channel = null; - return true; - } - return false; - } - - public void setChannel(MessageChannel channel) { - this.channel = channel; - } - - public MessageChannel getChannel() { - return this.channel; - } public void setContainer(AbstractMessageListenerContainer container) { this.container = container; @@ -121,10 +89,6 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy this.messageConverter = messageConverter; } - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - public void setTaskExecutor(TaskExecutor taskExecutor) { this.taskExecutor = taskExecutor; } @@ -137,15 +101,18 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy this.sessionAcknowledgeMode = sessionAcknowledgeMode; } + public void setExpectReply(boolean expectReply) { + this.expectReply = expectReply; + } + private void initialize() { - if (this.channel == null) { - throw new ConfigurationException("channel must not be null"); - } if (this.container == null) { this.container = createDefaultContainer(); } - ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel(), this.messageConverter); - listener.setTimeout(this.sendTimeout); + MessageListenerAdapter listener = new MessageListenerAdapter(); + listener.setDelegate(this); + listener.setDefaultListenerMethod(this.expectReply ? "request" : "send"); + listener.setMessageConverter(this.messageConverter); this.container.setMessageListener(listener); if (!this.container.isActive()) { this.container.afterPropertiesSet(); @@ -169,7 +136,6 @@ public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecy if (this.destinationName != null) { dmlc.setDestinationName(this.destinationName); } - dmlc.setReceiveTimeout(this.receiveTimeout); dmlc.setSessionTransacted(this.sessionTransacted); dmlc.setSessionAcknowledgeMode(this.sessionAcknowledgeMode); dmlc.setAutoStartup(false); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java index a85db97517..31004438b3 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java @@ -21,7 +21,7 @@ import javax.jms.Destination; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.jms.core.JmsTemplate; /** @@ -32,7 +32,7 @@ import org.springframework.jms.core.JmsTemplate; * * @author Mark Fisher */ -public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implements PollableSource { +public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implements Source { public JmsPollableSource(JmsTemplate jmsTemplate) { super(jmsTemplate); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsAdapterParserUtils.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsAdapterParserUtils.java index 09ab7026e1..86c8a9e6c0 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsAdapterParserUtils.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsAdapterParserUtils.java @@ -16,6 +16,8 @@ package org.springframework.integration.adapter.jms.config; +import javax.jms.Session; + import org.w3c.dom.Element; import org.springframework.beans.factory.BeanCreationException; @@ -44,6 +46,20 @@ public abstract class JmsAdapterParserUtils { public static final String DESTINATION_NAME_PROPERTY = "destinationName"; + public static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter"; + + public static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter"; + + private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge"; + + private static final String ACKNOWLEDGE_AUTO = "auto"; + + private static final String ACKNOWLEDGE_CLIENT = "client"; + + private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok"; + + private static final String ACKNOWLEDGE_TRANSACTED = "transacted"; + public static String determineConnectionFactoryBeanName(Element element) { String connectionFactoryBeanName = "connectionFactory"; @@ -57,4 +73,28 @@ public abstract class JmsAdapterParserUtils { return connectionFactoryBeanName; } + public static Integer parseAcknowledgeMode(Element element) { + String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE); + if (StringUtils.hasText(acknowledge)) { + int acknowledgeMode = Session.AUTO_ACKNOWLEDGE; + if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) { + acknowledgeMode = Session.SESSION_TRANSACTED; + } + else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) { + acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE; + } + else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) { + acknowledgeMode = Session.CLIENT_ACKNOWLEDGE; + } + else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) { + throw new BeanCreationException("Invalid JMS 'acknowledge' setting: " + + "only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported."); + } + return acknowledgeMode; + } + else { + return null; + } + } + } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsGatewayParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsGatewayParser.java new file mode 100644 index 0000000000..eb8409dba7 --- /dev/null +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsGatewayParser.java @@ -0,0 +1,107 @@ +/* + * Copyright 2002-2008 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.adapter.jms.config; + +import javax.jms.Session; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.adapter.jms.JmsMessageDrivenSourceAdapter; +import org.springframework.util.StringUtils; + +/** + * Parser for the <jms-gateway> element. + * + * @author Mark Fisher + */ +public class JmsGatewayParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return JmsMessageDrivenSourceAdapter.class; + } + + @Override + protected boolean shouldGenerateId() { + return false; + } + + @Override + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); + String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); + String messageConverter = element.getAttribute(JmsAdapterParserUtils.MESSAGE_CONVERTER_ATTRIBUTE); + if (StringUtils.hasText(element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE))) { + throw new BeanCreationException(JmsMessageDrivenSourceAdapter.class.getSimpleName() + + " does not accept a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + + "' reference. One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided."); + } + if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { + builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY, + JmsAdapterParserUtils.determineConnectionFactoryBeanName(element)); + if (StringUtils.hasText(destination)) { + builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination); + } + else { + builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName); + } + } + else { + throw new BeanCreationException("One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + + "' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided."); + } + if (StringUtils.hasText(messageConverter)) { + builder.addPropertyReference(JmsAdapterParserUtils.MESSAGE_CONVERTER_PROPERTY, messageConverter); + } + Integer acknowledgeMode = JmsAdapterParserUtils.parseAcknowledgeMode(element); + if (acknowledgeMode != null) { + if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) { + builder.addPropertyValue("sessionTransacted", Boolean.TRUE); + } + else { + builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); + } + } + String requestChannel = element.getAttribute("request-channel"); + if (StringUtils.hasText(requestChannel)) { + builder.addPropertyReference("requestChannel", requestChannel); + } + String requestTimeout = element.getAttribute("request-timeout"); + if (StringUtils.hasText(requestTimeout)) { + builder.addPropertyValue("requestTimeout", Long.parseLong(requestTimeout)); + } + String replyChannel = element.getAttribute("reply-channel"); + if (StringUtils.hasText(replyChannel)) { + builder.addPropertyReference("replyChannel", replyChannel); + } + String replyTimeout = element.getAttribute("reply-timeout"); + if (StringUtils.hasText(replyTimeout)) { + builder.addPropertyValue("replyTimeout", Long.parseLong(replyTimeout)); + } + } + +} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java index b39ede5b87..e214d008b2 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java @@ -36,21 +36,6 @@ import org.springframework.util.StringUtils; */ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { - private static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter"; - - private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter"; - - private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge"; - - private static final String ACKNOWLEDGE_AUTO = "auto"; - - private static final String ACKNOWLEDGE_CLIENT = "client"; - - private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok"; - - private static final String ACKNOWLEDGE_TRANSACTED = "transacted"; - - protected boolean shouldGenerateId() { return false; } @@ -61,20 +46,7 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { @Override protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) { - if ("true".equals(element.getAttribute("message-driven"))) { - return parseMessageDrivenSource(element, parserContext); - } - return parsePollableSource(element, parserContext); - } - - private AbstractBeanDefinition parsePollableSource(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsPollableSource.class); - if (StringUtils.hasText(element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE))) { - throw new BeanCreationException( - "The '" + MESSAGE_CONVERTER_ATTRIBUTE + "' attribute is not supported for a polling JMS adapter. " + - ". Consider providing a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + - "' reference where the template contains a 'messageConverter' property instead."); - } String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE); String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); @@ -107,68 +79,4 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { return builder.getBeanDefinition(); } - private AbstractBeanDefinition parseMessageDrivenSource(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsMessageDrivenSourceAdapter.class); - String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); - String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); - String messageConverter = element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE); - if (StringUtils.hasText(element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE))) { - throw new BeanCreationException(JmsMessageDrivenSourceAdapter.class.getSimpleName() + - " does not accept a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + - "' reference. One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + - JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided."); - } - if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { - builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY, - JmsAdapterParserUtils.determineConnectionFactoryBeanName(element)); - if (StringUtils.hasText(destination)) { - builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination); - } - else { - builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName); - } - } - else { - throw new BeanCreationException("One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + - "' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided."); - } - if (StringUtils.hasText(messageConverter)) { - builder.addPropertyReference(MESSAGE_CONVERTER_PROPERTY, messageConverter); - } - Integer acknowledgeMode = parseAcknowledgeMode(element); - if (acknowledgeMode != null) { - if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) { - builder.addPropertyValue("sessionTransacted", Boolean.TRUE); - } - else { - builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); - } - } - return builder.getBeanDefinition(); - } - - private Integer parseAcknowledgeMode(Element element) { - String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE); - if (StringUtils.hasText(acknowledge)) { - int acknowledgeMode = Session.AUTO_ACKNOWLEDGE; - if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) { - acknowledgeMode = Session.SESSION_TRANSACTED; - } - else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) { - acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE; - } - else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) { - acknowledgeMode = Session.CLIENT_ACKNOWLEDGE; - } - else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) { - throw new BeanCreationException("Invalid jms-source 'acknowledge' setting: " + - "only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported."); - } - return acknowledgeMode; - } - else { - return null; - } - } - } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java index 6010667bfb..c4fd3152b2 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java @@ -23,14 +23,14 @@ import java.io.InputStream; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; /** * A pollable source for receiving bytes from an {@link InputStream}. * * @author Mark Fisher */ -public class ByteStreamSource implements PollableSource { +public class ByteStreamSource implements Source { private BufferedInputStream stream; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java index a87dd1c6cf..c274ebb0df 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -24,7 +24,7 @@ import java.io.UnsupportedEncodingException; import org.springframework.integration.ConfigurationException; import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.integration.message.StringMessage; import org.springframework.util.Assert; @@ -33,7 +33,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class CharacterStreamSource implements PollableSource { +public class CharacterStreamSource implements Source { private final BufferedReader reader; diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/HttpInvokerSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/HttpInvokerSourceAdapterParserTests.java index e1b5961315..4d9c591f95 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/HttpInvokerSourceAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/HttpInvokerSourceAdapterParserTests.java @@ -25,6 +25,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.adapter.httpinvoker.HttpInvokerSourceAdapter; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.RequestReplyTemplate; /** * @author Mark Fisher @@ -38,10 +39,13 @@ public class HttpInvokerSourceAdapterParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); HttpInvokerSourceAdapter adapter = (HttpInvokerSourceAdapter) context.getBean("adapterWithDefaults"); DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); - assertEquals(channel, accessor.getPropertyValue("channel")); + assertEquals(channel, accessor.getPropertyValue("requestChannel")); assertEquals(true, accessor.getPropertyValue("expectReply")); - assertEquals(-1L, accessor.getPropertyValue("sendTimeout")); - assertEquals(-1L, accessor.getPropertyValue("receiveTimeout")); + RequestReplyTemplate template = (RequestReplyTemplate) + accessor.getPropertyValue("requestReplyTemplate"); + DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); + assertEquals(-1L, templateAccessor.getPropertyValue("requestTimeout")); + assertEquals(-1L, templateAccessor.getPropertyValue("replyTimeout")); } @Test @@ -51,10 +55,13 @@ public class HttpInvokerSourceAdapterParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); HttpInvokerSourceAdapter adapter = (HttpInvokerSourceAdapter) context.getBean("/adapter/with/name"); DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); - assertEquals(channel, accessor.getPropertyValue("channel")); + assertEquals(channel, accessor.getPropertyValue("requestChannel")); assertEquals(true, accessor.getPropertyValue("expectReply")); - assertEquals(-1L, accessor.getPropertyValue("sendTimeout")); - assertEquals(-1L, accessor.getPropertyValue("receiveTimeout")); + RequestReplyTemplate template = (RequestReplyTemplate) + accessor.getPropertyValue("requestReplyTemplate"); + DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); + assertEquals(-1L, templateAccessor.getPropertyValue("requestTimeout")); + assertEquals(-1L, templateAccessor.getPropertyValue("replyTimeout")); } @Test @@ -64,10 +71,13 @@ public class HttpInvokerSourceAdapterParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); HttpInvokerSourceAdapter adapter = (HttpInvokerSourceAdapter) context.getBean("adapterWithCustomProperties"); DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); - assertEquals(channel, accessor.getPropertyValue("channel")); + assertEquals(channel, accessor.getPropertyValue("requestChannel")); assertEquals(false, accessor.getPropertyValue("expectReply")); - assertEquals(123L, accessor.getPropertyValue("sendTimeout")); - assertEquals(456L, accessor.getPropertyValue("receiveTimeout")); + RequestReplyTemplate template = (RequestReplyTemplate) + accessor.getPropertyValue("requestReplyTemplate"); + DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); + assertEquals(123L, templateAccessor.getPropertyValue("requestTimeout")); + assertEquals(456L, templateAccessor.getPropertyValue("replyTimeout")); } } diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/httpInvokerSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/httpInvokerSourceAdapterParserTests.xml index 3e48f7275a..8b6234a560 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/httpInvokerSourceAdapterParserTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/config/httpInvokerSourceAdapterParserTests.xml @@ -11,11 +11,12 @@ - + - + - + \ No newline at end of file diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java index 86670ac7f4..3e2925d70b 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java @@ -81,7 +81,7 @@ public class JmsSourceAdapterParserTests { "messageDrivenAdapterWithConnectionFactoryAndDestination.xml", this.getClass()); MessageChannel channel = new QueueChannel(1); JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); - source.setChannel(channel); + source.setRequestChannel(channel); context.start(); Message message = channel.receive(3000); assertNotNull("message should not be null", message); @@ -95,7 +95,7 @@ public class JmsSourceAdapterParserTests { "messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); MessageChannel channel = new QueueChannel(1); JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); - source.setChannel(channel); + source.setRequestChannel(channel); context.start(); assertEquals(JmsMessageDrivenSourceAdapter.class, source.getClass()); Message message = channel.receive(3000); @@ -110,7 +110,7 @@ public class JmsSourceAdapterParserTests { "messageDrivenAdapterWithMessageConverter.xml", this.getClass()); MessageChannel channel = new QueueChannel(1); JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); - source.setChannel(channel); + source.setRequestChannel(channel); context.start(); Message message = channel.receive(3000); assertNotNull("message should not be null", message); @@ -194,7 +194,7 @@ public class JmsSourceAdapterParserTests { "messageDrivenAdapterWithDefaultConnectionFactory.xml", this.getClass()); MessageChannel channel = new QueueChannel(1); JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); - source.setChannel(channel); + source.setRequestChannel(channel); context.start(); Message message = channel.receive(3000); assertNotNull("message should not be null", message); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestination.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestination.xml index 835be17332..9b1b563a41 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestination.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestination.xml @@ -7,9 +7,12 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml index 7c2b6788dd..f26c5b0884 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml @@ -9,9 +9,12 @@ - + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryOnly.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryOnly.xml index ad0fdc151e..58fda50c4b 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryOnly.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryOnly.xml @@ -9,7 +9,11 @@ - + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithDefaultConnectionFactory.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithDefaultConnectionFactory.xml index 163d070f6d..00da02ba5b 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithDefaultConnectionFactory.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithDefaultConnectionFactory.xml @@ -9,7 +9,11 @@ - + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithEmptyConnectionFactory.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithEmptyConnectionFactory.xml index 53ea30c514..ce1b2b4f37 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithEmptyConnectionFactory.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithEmptyConnectionFactory.xml @@ -7,6 +7,11 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml index f0424eadb4..0e741f7a9d 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml @@ -9,10 +9,13 @@ - + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestination.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestination.xml index bc40568c27..22a477f1e8 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestination.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestination.xml @@ -9,8 +9,7 @@ + destination="testDestination"/> diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestinationName.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestinationName.xml index 051e5d30d8..4f02f499ec 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestinationName.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestinationName.xml @@ -9,8 +9,7 @@ + destination-name="testDestinationName"/> diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryOnly.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryOnly.xml index 83d597cf91..cc84238006 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryOnly.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryOnly.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationAndDefaultConnectionFactory.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationAndDefaultConnectionFactory.xml index d2defb8d24..3f60385461 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationAndDefaultConnectionFactory.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationAndDefaultConnectionFactory.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml index 9004616760..d28cbb76c0 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameOnly.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameOnly.xml index 29721ed65d..3ecb8db83f 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameOnly.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameOnly.xml @@ -7,6 +7,6 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationOnly.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationOnly.xml index ef8c96767e..0b23445fc7 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationOnly.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationOnly.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithJmsTemplate.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithJmsTemplate.xml index 62ba77bb4e..9a96444d8a 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithJmsTemplate.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithJmsTemplate.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml index ae19d9e71a..ec942b4ef3 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml @@ -15,7 +15,7 @@ - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java index e13ba3ffb5..b90542a291 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java @@ -26,6 +26,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.adapter.rmi.RmiSourceAdapter; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.RequestReplyTemplate; /** * @author Mark Fisher @@ -39,10 +40,13 @@ public class RmiSourceAdapterParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithDefaults"); DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); - assertEquals(channel, accessor.getPropertyValue("channel")); + assertEquals(channel, accessor.getPropertyValue("requestChannel")); assertEquals(true, accessor.getPropertyValue("expectReply")); - assertEquals(-1L, accessor.getPropertyValue("sendTimeout")); - assertEquals(-1L, accessor.getPropertyValue("receiveTimeout")); + RequestReplyTemplate template = (RequestReplyTemplate) + accessor.getPropertyValue("requestReplyTemplate"); + DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); + assertEquals(-1L, templateAccessor.getPropertyValue("requestTimeout")); + assertEquals(-1L, templateAccessor.getPropertyValue("replyTimeout")); } @Test @@ -52,10 +56,13 @@ public class RmiSourceAdapterParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithCustomProperties"); DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); - assertEquals(channel, accessor.getPropertyValue("channel")); + assertEquals(channel, accessor.getPropertyValue("requestChannel")); assertEquals(false, accessor.getPropertyValue("expectReply")); - assertEquals(123L, accessor.getPropertyValue("sendTimeout")); - assertEquals(456L, accessor.getPropertyValue("receiveTimeout")); + RequestReplyTemplate template = (RequestReplyTemplate) + accessor.getPropertyValue("requestReplyTemplate"); + DirectFieldAccessor templateAccessor = new DirectFieldAccessor(template); + assertEquals(123L, templateAccessor.getPropertyValue("requestTimeout")); + assertEquals(456L, templateAccessor.getPropertyValue("replyTimeout")); } @Test diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml index 5c9b00404d..9e52cc2d9b 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml @@ -11,16 +11,16 @@ - + - + - + - + - + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java index 5cce87c8fb..3bcbdded22 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java @@ -23,7 +23,7 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.handler.HandlerMethodInvoker; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.integration.util.MethodValidator; import org.springframework.util.Assert; @@ -33,7 +33,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MethodInvokingSource implements PollableSource, InitializingBean { +public class MethodInvokingSource implements Source, InitializingBean { private T object; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java index 9e7d5129e6..1d14d7f9ca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java @@ -18,11 +18,9 @@ package org.springframework.integration.channel; import java.util.List; -import org.springframework.beans.BeansException; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; import org.springframework.integration.ConfigurationException; -import org.springframework.integration.config.MessageBusParser; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.MessageBusAware; import org.springframework.integration.endpoint.EndpointRegistry; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.handler.ReplyHandler; @@ -38,7 +36,7 @@ import org.springframework.integration.scheduling.Subscription; * * @author Mark Fisher */ -public class RequestReplyTemplate implements ApplicationContextAware { +public class RequestReplyTemplate implements MessageBusAware { private MessageChannel requestChannel; @@ -122,9 +120,9 @@ public class RequestReplyTemplate implements ApplicationContextAware { this.endpointRegistry = endpointRegistry; } - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - if (applicationContext.containsBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)) { - this.setEndpointRegistry((EndpointRegistry) applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); + public void setMessageBus(MessageBus messageBus) { + if (this.endpointRegistry == null) { + this.setEndpointRegistry(messageBus); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourceEndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourceEndpointParser.java index 103ca2a213..b26edc00e9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourceEndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourceEndpointParser.java @@ -23,7 +23,6 @@ import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.endpoint.PollingSourceEndpoint; -import org.springframework.integration.endpoint.SimpleSourceEndpoint; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; import org.springframework.util.StringUtils; @@ -37,10 +36,7 @@ import org.springframework.util.xml.DomUtils; public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser { protected final Class getBeanClass(Element element) { - if (this.getScheduleElement(element) != null) { - return PollingSourceEndpoint.class; - } - return SimpleSourceEndpoint.class; + return PollingSourceEndpoint.class; } protected boolean shouldGenerateId() { @@ -68,9 +64,10 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser { builder.addConstructorArgReference(source); builder.addConstructorArgReference(output); Element scheduleElement = this.getScheduleElement(element); - if (scheduleElement != null) { - builder.addConstructorArgValue(this.parseSchedule(scheduleElement)); + if (scheduleElement == null) { + throw new ConfigurationException("The sub-element is required for a ."); } + builder.addConstructorArgValue(this.parseSchedule(scheduleElement)); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java index 68606d792f..60cc4e639e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java @@ -25,7 +25,7 @@ import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.util.Assert; /** @@ -46,14 +46,14 @@ import org.springframework.util.Assert; */ public class DefaultPollingDispatcher extends SimpleDispatcher implements PollingDispatcher { - private final PollableSource source; + private final Source source; public DefaultPollingDispatcher(MessageChannel channel) { this(channel, channel.getDispatcherPolicy()); } - public DefaultPollingDispatcher(PollableSource source, DispatcherPolicy dispatcherPolicy) { + public DefaultPollingDispatcher(Source source, DispatcherPolicy dispatcherPolicy) { super(dispatcherPolicy); Assert.notNull(source, "source must not be null"); this.source = source; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java index c1848ab70d..0a3cbf8abd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java @@ -17,11 +17,11 @@ package org.springframework.integration.dispatcher; import org.springframework.integration.message.Poller; -import org.springframework.integration.message.SubscribableSource; +import org.springframework.integration.message.Subscribable; /** * @author Mark Fisher */ -public interface PollingDispatcher extends Poller, MessageDispatcher, SubscribableSource { +public interface PollingDispatcher extends Poller, MessageDispatcher, Subscribable { } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java index 233645776e..2bc2605510 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java @@ -30,7 +30,7 @@ import org.springframework.integration.handler.MessageHandlerRejectedExecutionEx import org.springframework.integration.message.BlockingTarget; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.SubscribableSource; +import org.springframework.integration.message.Subscribable; import org.springframework.integration.message.Target; /** @@ -38,7 +38,7 @@ import org.springframework.integration.message.Target; * * @author Mark Fisher */ -public class SimpleDispatcher implements MessageDispatcher, SubscribableSource { +public class SimpleDispatcher implements MessageDispatcher, Subscribable { protected final Log logger = LogFactory.getLog(this.getClass()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java index a99067891a..5c1d7e2adc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java @@ -26,8 +26,8 @@ import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.message.SubscribableSource; +import org.springframework.integration.message.Source; +import org.springframework.integration.message.Subscribable; import org.springframework.integration.message.Target; import org.springframework.integration.message.selector.MessageSelector; @@ -45,12 +45,12 @@ import org.springframework.integration.message.selector.MessageSelector; * @author Dave Syer * @author Mark Fisher */ -public class SynchronousChannel extends AbstractMessageChannel implements SubscribableSource { +public class SynchronousChannel extends AbstractMessageChannel implements Subscribable { private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder(); - private volatile PollableSource source; + private volatile Source source; private final SimpleDispatcher dispatcher; @@ -61,14 +61,14 @@ public class SynchronousChannel extends AbstractMessageChannel implements Subscr this(null); } - public SynchronousChannel(PollableSource source) { + public SynchronousChannel(Source source) { super(defaultDispatcherPolicy()); this.source = source; this.dispatcher = new SimpleDispatcher(this.getDispatcherPolicy()); } - public void setSource(PollableSource source) { + public void setSource(Source source) { this.source = source; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java index 61054b263b..54cf1c9865 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java @@ -32,14 +32,14 @@ public abstract class AbstractSourceEndpoint implements SourceEndpoint { protected final Log logger = LogFactory.getLog(this.getClass()); - private final Source source; + private final Source source; private final MessageChannel channel; private volatile String name; - public AbstractSourceEndpoint(Source source, MessageChannel channel) { + public AbstractSourceEndpoint(Source source, MessageChannel channel) { Assert.notNull(source, "source must not be null"); Assert.notNull(channel, "channel must not be null"); this.source = source; @@ -47,7 +47,7 @@ public abstract class AbstractSourceEndpoint implements SourceEndpoint { } - public Source getSource() { + public Source getSource() { return this.source; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java index d17afe459c..d268448f5d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java @@ -29,7 +29,7 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.dispatcher.DefaultPollingDispatcher; import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.dispatcher.PollingDispatcherTask; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; @@ -60,7 +60,7 @@ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements Mes private final Object taskMonitor = new Object(); - public PollingSourceEndpoint(PollableSource source, MessageChannel channel, PollingSchedule schedule) { + public PollingSourceEndpoint(Source source, MessageChannel channel, PollingSchedule schedule) { super(source, channel); Assert.notNull(schedule, "schedule must not be null"); this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SimpleSourceEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SimpleSourceEndpoint.java deleted file mode 100644 index 21c6a6dd46..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SimpleSourceEndpoint.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * Copyright 2002-2008 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.endpoint; - -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.SubscribableSource; - -/** - * @author Mark Fisher - */ -public class SimpleSourceEndpoint extends AbstractSourceEndpoint { - - public SimpleSourceEndpoint(SubscribableSource source, MessageChannel channel) { - super(source, channel); - source.subscribe(channel); - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index b919c8a9aa..59dd434362 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -23,8 +23,12 @@ import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.SimpleTypeConverter; import org.springframework.beans.TypeConverter; import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.message.Message; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -35,7 +39,8 @@ import org.springframework.util.ClassUtils; * * @author Mark Fisher */ -public class GatewayProxyFactoryBean extends MessagingGateway implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware { +public class GatewayProxyFactoryBean extends MessagingGateway + implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware, BeanFactoryAware { private Class serviceInterface; @@ -59,6 +64,11 @@ public class GatewayProxyFactoryBean extends MessagingGateway implements Factory this.beanClassLoader = beanClassLoader; } + public void setBeanFactory(BeanFactory beanFactory) { + this.getRequestReplyTemplate().setMessageBus( + (MessageBus) beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); + } + public void afterPropertiesSet() { this.serviceProxy = new ProxyFactory(this.serviceInterface, this).getProxy(this.beanClassLoader); } @@ -84,7 +94,7 @@ public class GatewayProxyFactoryBean extends MessagingGateway implements Factory if (shouldReturnMessage) { return this.receive(); } - response = this.invoke(); + response = this.receive(); } else { Object payload = (paramCount == 1) ? invocation.getArguments()[0] : invocation.getArguments(); @@ -92,7 +102,7 @@ public class GatewayProxyFactoryBean extends MessagingGateway implements Factory this.send(payload); return null; } - response = this.invoke(payload, !shouldReturnMessage); + response = shouldReturnMessage ? this.sendAndReceiveMessage(payload) : this.sendAndReceive(payload); } return (response != null) ? this.typeConverter.convertIfNecessary(response, returnType) : null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java index d61b4b7003..094bf57f42 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java @@ -33,19 +33,17 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessagingGateway extends RequestReplyTemplate { +public class MessagingGateway { + + private final RequestReplyTemplate requestReplyTemplate = new RequestReplyTemplate(); private MessageCreator messageCreator = new DefaultMessageCreator(); private MessageMapper messageMapper = new DefaultMessageMapper(); - public MessagingGateway(MessageChannel requestChannel, MessageChannel replyChannel) { - super(requestChannel, replyChannel); - } - public MessagingGateway(MessageChannel requestChannel) { - super(requestChannel); + this.requestReplyTemplate.setRequestChannel(requestChannel); } public MessagingGateway() { @@ -53,6 +51,22 @@ public class MessagingGateway extends RequestReplyTemplate { } + public void setRequestChannel(MessageChannel requestChannel) { + this.requestReplyTemplate.setRequestChannel(requestChannel); + } + + public void setReplyChannel(MessageChannel replyChannel) { + this.requestReplyTemplate.setReplyChannel(replyChannel); + } + + public void setRequestTimeout(long requestTimeout) { + this.requestReplyTemplate.setRequestTimeout(requestTimeout); + } + + public void setReplyTimeout(long replyTimeout) { + this.requestReplyTemplate.setReplyTimeout(replyTimeout); + } + public void setMessageCreator(MessageCreator messageCreator) { Assert.notNull(messageCreator, "messageCreator must not be null"); this.messageCreator = messageCreator; @@ -63,30 +77,38 @@ public class MessagingGateway extends RequestReplyTemplate { this.messageMapper = messageMapper; } + protected RequestReplyTemplate getRequestReplyTemplate() { + return this.requestReplyTemplate; + } + public void send(Object object) { Message message = (object instanceof Message) ? (Message) object : this.messageCreator.createMessage(object); if (message != null) { - this.send(message); + this.requestReplyTemplate.send(message); } } - public Object invoke() { - Message message = this.receive(); + public Object receive() { + Message message = this.requestReplyTemplate.receive(); return (message != null) ? this.messageMapper.mapMessage(message) : null; } - public Object invoke(Object object) { - return this.invoke(object, true); + public Object sendAndReceive(Object object) { + return this.sendAndReceive(object, true); } - public Object invoke(Object object, boolean shouldMapMessage) { + public Message sendAndReceiveMessage(Object object) { + return (Message) this.sendAndReceive(object, false); + } + + private Object sendAndReceive(Object object, boolean shouldMapMessage) { Message request = (object instanceof Message) ? (Message) object : this.messageCreator.createMessage(object); if (request == null) { return null; } - Message reply = this.request(request); + Message reply = this.requestReplyTemplate.request(request); if (!shouldMapMessage) { return reply; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java index 83e726e702..1333e74e17 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/BlockingSource.java @@ -21,7 +21,7 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public interface BlockingSource extends PollableSource { +public interface BlockingSource extends Source { /** * Receive a message, blocking indefinitely if necessary. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java deleted file mode 100644 index f8d39416d1..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright 2002-2008 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.message; - -/** - * Interface for any external message source that can be polled. - * - * @author Mark Fisher - */ -public interface PollableSource extends Source { - - /** - * Retrieve a message from this source or null if no message is available. - */ - Message receive(); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/Source.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Source.java index 4357a8c6d4..cb18538344 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/Source.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/Source.java @@ -21,6 +21,11 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public interface Source { +public interface Source { + + /** + * Retrieve a message from this source or null if no message is available. + */ + Message receive(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Subscribable.java similarity index 89% rename from spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java rename to spring-integration-core/src/main/java/org/springframework/integration/message/Subscribable.java index dee66a53cc..ce4fc44543 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/SubscribableSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/Subscribable.java @@ -17,11 +17,11 @@ package org.springframework.integration.message; /** - * Interface for any message source that accepts subscribers. + * Interface for any component that accepts subscribers. * * @author Mark Fisher */ -public interface SubscribableSource extends Source { +public interface Subscribable { /** * Register a {@link Target} as a subscriber to this source. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 94165074cc..35a912a58a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -25,6 +25,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.Test; + import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.DispatcherPolicy; @@ -38,7 +39,7 @@ import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Subscription; @@ -219,9 +220,7 @@ public class MessageBusTests { @Test(expected = BeanCreationException.class) public void testMultipleMessageBusBeans() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", - this.getClass()); - + new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", this.getClass()); } @Test @@ -258,7 +257,7 @@ public class MessageBusTests { assertTrue(messageBusAwareBean.getMessageBus() == context.getBean("bus")); } - private static class FailingSource implements PollableSource { + private static class FailingSource implements Source { private CountDownLatch latch; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java index 7e530ea095..528fb4c750 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java @@ -29,7 +29,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.Target; @@ -96,7 +96,7 @@ public class SynchronousChannelTests { @Test public void testReceive() { - SynchronousChannel channel = new SynchronousChannel(new PollableSource() { + SynchronousChannel channel = new SynchronousChannel(new Source() { public Message receive() { return new StringMessage("foo"); } @@ -175,7 +175,7 @@ public class SynchronousChannelTests { } - private static class MessageReturningTestSource implements PollableSource { + private static class MessageReturningTestSource implements Source { private final String messageText; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java index cea7d90800..3e25192f8e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java @@ -35,7 +35,7 @@ import org.springframework.aop.MethodBeforeAdvice; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.Source; import org.springframework.integration.scheduling.PollingSchedule; /** @@ -429,7 +429,7 @@ public class PollingSourceEndpointTests { } - private static class TestSource implements PollableSource { + private static class TestSource implements Source { private String message; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java index a28f15399e..c031f0c7ce 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java @@ -37,8 +37,8 @@ public class GatewayProxyFactoryBeanTests { final MessageChannel requestChannel = new QueueChannel(); startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); - proxyFactory.setServiceInterface(TestService.class); proxyFactory.setRequestChannel(requestChannel); + proxyFactory.setServiceInterface(TestService.class); proxyFactory.afterPropertiesSet(); TestService service = (TestService) proxyFactory.getObject(); String result = service.requestReply("foo");