From 4c0c3677d009ee3bbc1d9da574592d935f8707c7 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sat, 3 May 2008 22:23:40 +0000 Subject: [PATCH] The element is now available for message-driven JMS adapters. The is strictly for polling adapters. Provided basic foundation for converting other request-reply adapters into "gateways". This includes the following namespace changes: 'send-timeout' is now 'request-timeout', 'receive-timeout' is now 'reply-timeout', and the 'channel' element is now 'request-channel'. The Source interface now includes the receive() method itself. PollableSource and SubscribableSource interfaces have both been removed. The Subscribable interface was added (e.g. for SynchronousChannel), but "gateway" adapter types will be configured directly with a request channel (and optionally a response channel). --- .../java/META-INF/spring-integration.parsers | 1 + .../adapter/MessageHandlingSourceAdapter.java | 41 ++---- ...stractRequestReplySourceAdapterParser.java | 18 +-- .../spring-integration-adapters-1.0.xsd | 126 +++++++++++------- .../integration/adapter/file/FileSource.java | 4 +- .../integration/adapter/ftp/FtpSource.java | 4 +- .../jms/JmsMessageDrivenSourceAdapter.java | 60 ++------- .../adapter/jms/JmsPollableSource.java | 4 +- .../jms/config/JmsAdapterParserUtils.java | 40 ++++++ .../adapter/jms/config/JmsGatewayParser.java | 107 +++++++++++++++ .../jms/config/JmsSourceAdapterParser.java | 92 ------------- .../adapter/stream/ByteStreamSource.java | 4 +- .../adapter/stream/CharacterStreamSource.java | 4 +- .../HttpInvokerSourceAdapterParserTests.java | 28 ++-- .../httpInvokerSourceAdapterParserTests.xml | 9 +- .../config/JmsSourceAdapterParserTests.java | 8 +- ...terWithConnectionFactoryAndDestination.xml | 9 +- ...ithConnectionFactoryAndDestinationName.xml | 9 +- ...DrivenAdapterWithConnectionFactoryOnly.xml | 6 +- ...venAdapterWithDefaultConnectionFactory.xml | 6 +- ...rivenAdapterWithEmptyConnectionFactory.xml | 7 +- ...ssageDrivenAdapterWithMessageConverter.xml | 11 +- ...terWithConnectionFactoryAndDestination.xml | 3 +- ...ithConnectionFactoryAndDestinationName.xml | 3 +- ...ollingAdapterWithConnectionFactoryOnly.xml | 2 +- ...DestinationAndDefaultConnectionFactory.xml | 2 +- ...inationNameAndDefaultConnectionFactory.xml | 2 +- .../pollingAdapterWithDestinationNameOnly.xml | 2 +- .../pollingAdapterWithDestinationOnly.xml | 2 +- .../config/pollingAdapterWithJmsTemplate.xml | 2 +- .../jms/config/pollingJmsSourceEndpoint.xml | 2 +- .../config/RmiSourceAdapterParserTests.java | 19 ++- .../config/rmiSourceAdapterParserTests.xml | 12 +- .../adapter/MethodInvokingSource.java | 4 +- .../channel/RequestReplyTemplate.java | 14 +- .../config/SourceEndpointParser.java | 11 +- .../dispatcher/DefaultPollingDispatcher.java | 6 +- .../dispatcher/PollingDispatcher.java | 4 +- .../dispatcher/SimpleDispatcher.java | 4 +- .../dispatcher/SynchronousChannel.java | 12 +- .../endpoint/AbstractSourceEndpoint.java | 6 +- .../endpoint/PollingSourceEndpoint.java | 4 +- .../endpoint/SimpleSourceEndpoint.java | 32 ----- .../gateway/GatewayProxyFactoryBean.java | 16 ++- .../integration/gateway/MessagingGateway.java | 48 +++++-- .../integration/message/BlockingSource.java | 2 +- .../integration/message/PollableSource.java | 31 ----- .../integration/message/Source.java | 7 +- ...scribableSource.java => Subscribable.java} | 4 +- .../integration/bus/MessageBusTests.java | 9 +- .../dispatcher/SynchronousChannelTests.java | 6 +- .../endpoint/PollingSourceEndpointTests.java | 4 +- .../gateway/GatewayProxyFactoryBeanTests.java | 2 +- 53 files changed, 460 insertions(+), 415 deletions(-) create mode 100644 spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsGatewayParser.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/SimpleSourceEndpoint.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java rename spring-integration-core/src/main/java/org/springframework/integration/message/{SubscribableSource.java => Subscribable.java} (89%) 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");