diff --git a/spring-integration-core/.classpath b/spring-integration-core/.classpath index 9181e43c30..c565e26fdc 100644 --- a/spring-integration-core/.classpath +++ b/spring-integration-core/.classpath @@ -9,11 +9,11 @@ - + - + - + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java index 293b952199..d5145a8e01 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -75,6 +75,12 @@ public class AbstractSourceAdapter implements SourceAdapter, InitializingBean } protected boolean sendToChannel(T object) { + if (object == null) { + if (logger.isDebugEnabled()) { + logger.debug("adapter attempted to send a null object"); + } + return false; + } Message message = null; if (object instanceof Message) { message = (Message) object; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java index b9ddf2216a..01f66a0cec 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java @@ -29,12 +29,12 @@ import org.springframework.integration.adapter.file.FileSourceAdapter; */ public class FileSourceAdapterParser extends AbstractSingleBeanDefinitionParser { - protected Class getBeanClass(Element element) { + protected Class getBeanClass(Element element) { return FileSourceAdapter.class; } protected boolean shouldGenerateId() { - return true; + return false; } protected void doParse(Element element, BeanDefinitionBuilder builder) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParser.java index b430cd0b6b..288b2973c2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParser.java @@ -29,12 +29,12 @@ import org.springframework.integration.adapter.file.FileTargetAdapter; */ public class FileTargetAdapterParser extends AbstractSingleBeanDefinitionParser { - protected Class getBeanClass(Element element) { + protected Class getBeanClass(Element element) { return FileTargetAdapter.class; } protected boolean shouldGenerateId() { - return true; + return false; } protected void doParse(Element element, BeanDefinitionBuilder builder) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java index ab105b9b99..4ee647cd68 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java @@ -22,7 +22,7 @@ import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; -import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.DisposableBean; import org.springframework.context.Lifecycle; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.MessageHandlingException; @@ -41,7 +41,7 @@ import org.springframework.util.Assert; * @author Mark Fisher */ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter implements MessageListener, Lifecycle, - InitializingBean { + DisposableBean { private AbstractJmsListeningContainer container; @@ -133,6 +133,10 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter container.stop(); } + public void destroy() { + container.destroy(); + } + public void onMessage(Message message) { try { this.sendToChannel(messageConverter.fromMessage(message)); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java index 946584f8c7..c2cbf0a1bf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java @@ -29,9 +29,9 @@ import org.springframework.jms.core.JmsTemplate; /** * A source for receiving JMS Messages with a polling listener. This source is - * only recommended for very low message volume. Otherwise, an event-driven - * option that uses one of Spring's MessageListener containers is highly - * recommended. + * only recommended for very low message volume. Otherwise, the + * {@link JmsMessageDrivenSourceAdapter} that uses Spring's MessageListener + * container support is highly recommended. * * @author Mark Fisher */ @@ -41,6 +41,8 @@ public class JmsPollableSource implements PollableSource, InitializingBe private Destination destination; + private String destinationName; + private JmsTemplate jmsTemplate; @@ -54,6 +56,12 @@ public class JmsPollableSource implements PollableSource, InitializingBe this.initJmsTemplate(); } + public JmsPollableSource(ConnectionFactory connectionFactory, String destinationName) { + this.connectionFactory = connectionFactory; + this.destinationName = destinationName; + this.initJmsTemplate(); + } + /** * No-arg constructor provided for convenience when configuring with * setters. Note that the initialization callback will validate. @@ -69,15 +77,19 @@ public class JmsPollableSource implements PollableSource, InitializingBe this.destination = destination; } + public void setDestinationName(String destinationName) { + this.destinationName = destinationName; + } + public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public void afterPropertiesSet() { if (this.jmsTemplate == null) { - if (this.connectionFactory == null || this.destination == null) { + if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) { throw new MessagingConfigurationException("Either a 'jmsTemplate' or " - + "*both* 'connectionFactory' and 'destination' are required."); + + "both 'connectionFactory' and 'destination' (or 'destinationName') are required."); } this.initJmsTemplate(); } @@ -86,7 +98,15 @@ public class JmsPollableSource implements PollableSource, InitializingBe private void initJmsTemplate() { this.jmsTemplate = new JmsTemplate(); this.jmsTemplate.setConnectionFactory(this.connectionFactory); - this.jmsTemplate.setDefaultDestination(this.destination); + if (this.destination != null) { + this.jmsTemplate.setDefaultDestination(this.destination); + } + else if (this.destinationName != null) { + this.jmsTemplate.setDefaultDestinationName(this.destinationName); + } + else { + throw new MessagingConfigurationException("either 'destination' or 'destinationName' is required"); + } } public Collection poll(int limit) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java index 326b6a9b8b..32e18ea39b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java @@ -29,12 +29,16 @@ import org.springframework.jms.core.JmsTemplate; */ public class JmsPollingSourceAdapter extends PollingSourceAdapter { - public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, Destination destination) { - super(new JmsPollableSource(connectionFactory, destination)); - } - public JmsPollingSourceAdapter(JmsTemplate jmsTemplate) { super(new JmsPollableSource(jmsTemplate)); } + public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, Destination destination) { + super(new JmsPollableSource(connectionFactory, destination)); + } + + public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, String destinationName) { + super(new JmsPollableSource(connectionFactory, destinationName)); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java new file mode 100644 index 0000000000..27011411db --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java @@ -0,0 +1,141 @@ +/* + * Copyright 2002-2007 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 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.integration.adapter.jms.JmsMessageDrivenSourceAdapter; +import org.springframework.integration.adapter.jms.JmsPollingSourceAdapter; +import org.springframework.util.StringUtils; + +/** + * Parser for the <jms-source/> element. + * + * @author Mark Fisher + */ +public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser { + + private static final String JMS_TEMPLATE_ATTRIBUTE = "jms-template"; + + private static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory"; + + private static final String CONNECTION_FACTORY_PROPERTY = "connectionFactory"; + + private static final String DESTINATION_ATTRIBUTE = "destination"; + + private static final String DESTINATION_PROPERTY = "destination"; + + private static final String DESTINATION_NAME_ATTRIBUTE = "destination-name"; + + private static final String DESTINATION_NAME_PROPERTY = "destinationName"; + + private static final String CHANNEL_ATTRIBUTE = "channel"; + + private static final String CHANNEL_PROPERTY = "channel"; + + private static final String POLL_PERIOD_ATTRIBUTE = "poll-period"; + + private static final String POLL_PERIOD_PROPERTY = "period"; + + + protected Class getBeanClass(Element element) { + if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) { + return JmsPollingSourceAdapter.class; + } + return JmsMessageDrivenSourceAdapter.class; + } + + protected boolean shouldGenerateId() { + return false; + } + + protected void doParse(Element element, BeanDefinitionBuilder builder) { + if (builder.getBeanDefinition().getBeanClass().equals(JmsPollingSourceAdapter.class)) { + parsePollingSourceAdapter(element, builder); + } + else { + parseMessageDrivenSourceAdapter(element, builder); + } + String channel = element.getAttribute(CHANNEL_ATTRIBUTE); + builder.addPropertyReference(CHANNEL_PROPERTY, channel); + } + + private void parsePollingSourceAdapter(Element element, BeanDefinitionBuilder builder) { + String pollPeriod = element.getAttribute(POLL_PERIOD_ATTRIBUTE); + if (!StringUtils.hasText(pollPeriod)) { + throw new BeanCreationException("'" + POLL_PERIOD_ATTRIBUTE + + "' is required for a " + JmsPollingSourceAdapter.class.getSimpleName()); + } + builder.addPropertyValue(POLL_PERIOD_PROPERTY, pollPeriod); + String jmsTemplate = element.getAttribute(JMS_TEMPLATE_ATTRIBUTE); + String connectionFactory = element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE); + String destination = element.getAttribute(DESTINATION_ATTRIBUTE); + String destinationName = element.getAttribute(DESTINATION_NAME_ATTRIBUTE); + if (StringUtils.hasText(jmsTemplate)) { + if (StringUtils.hasText(connectionFactory) || StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { + throw new BeanCreationException("when providing '" + JMS_TEMPLATE_ATTRIBUTE + + "', none of '" + CONNECTION_FACTORY_ATTRIBUTE + "', '" + DESTINATION_ATTRIBUTE + + "', or '" + DESTINATION_NAME_ATTRIBUTE + "' should be provided."); + } + builder.addConstructorArgReference(jmsTemplate); + } + else if (StringUtils.hasText(connectionFactory) && (StringUtils.hasText(destination) || StringUtils.hasText(destinationName))) { + builder.addConstructorArgReference(connectionFactory); + if (StringUtils.hasText(destination)) { + builder.addConstructorArgReference(destination); + } + else if (StringUtils.hasText(destinationName)) { + builder.addConstructorArg(destinationName); + } + } + else { + throw new BeanCreationException("either a '" + JMS_TEMPLATE_ATTRIBUTE + "' or both '" + + CONNECTION_FACTORY_ATTRIBUTE + "' and '" + DESTINATION_ATTRIBUTE + "' (or '" + + DESTINATION_NAME_ATTRIBUTE + "') attributes must be provided for a " + + JmsPollingSourceAdapter.class.getSimpleName()); + } + } + + private void parseMessageDrivenSourceAdapter(Element element, BeanDefinitionBuilder builder) { + String connectionFactory = element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE); + String destination = element.getAttribute(DESTINATION_ATTRIBUTE); + String destinationName = element.getAttribute(DESTINATION_NAME_ATTRIBUTE); + if (StringUtils.hasText(element.getAttribute(JMS_TEMPLATE_ATTRIBUTE))) { + throw new BeanCreationException(JmsMessageDrivenSourceAdapter.class.getSimpleName() + + " does not accept a '" + JMS_TEMPLATE_ATTRIBUTE + "' reference, both " + + "'" + CONNECTION_FACTORY_ATTRIBUTE + "' and '" + DESTINATION_ATTRIBUTE + + "' (or '" + DESTINATION_NAME_ATTRIBUTE + "') must be provided."); + } + if (StringUtils.hasText(connectionFactory) && (StringUtils.hasText(destination) || StringUtils.hasText(destinationName))) { + builder.addPropertyReference(CONNECTION_FACTORY_PROPERTY, connectionFactory); + if (StringUtils.hasText(destination)) { + builder.addPropertyReference(DESTINATION_PROPERTY, destination); + } + else { + builder.addPropertyValue(DESTINATION_NAME_PROPERTY, destinationName); + } + } + else { + throw new BeanCreationException("Both '" + CONNECTION_FACTORY_ATTRIBUTE + "' and '" + + DESTINATION_ATTRIBUTE + "' (or '" + DESTINATION_NAME_ATTRIBUTE + "') must be provided."); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java index 2190397258..ed20426eff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java @@ -53,7 +53,7 @@ public class JmsTargetAdapterParser extends AbstractSingleBeanDefinitionParser { } protected boolean shouldGenerateId() { - return true; + return false; } protected void doParse(Element element, BeanDefinitionBuilder builder) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index 4d17bb7bb6..c3fe2fce70 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -19,6 +19,7 @@ package org.springframework.integration.config; import org.springframework.beans.factory.xml.NamespaceHandlerSupport; import org.springframework.integration.adapter.file.config.FileSourceAdapterParser; import org.springframework.integration.adapter.file.config.FileTargetAdapterParser; +import org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser; import org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser; /** @@ -37,6 +38,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("endpoint", new EndpointParser()); registerBeanDefinitionParser("file-source", new FileSourceAdapterParser()); registerBeanDefinitionParser("file-target", new FileTargetAdapterParser()); + registerBeanDefinitionParser("jms-source", new JmsSourceAdapterParser()); registerBeanDefinitionParser("jms-target", new JmsTargetAdapterParser()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 2d527797f8..3a8b77ef6d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -141,6 +141,7 @@ Defines a file-based source channel adapter. + @@ -154,11 +155,29 @@ Defines a file-based target channel adapter. + + + + + + Defines a jms-based source channel adapter. + + + + + + + + + + + + @@ -166,6 +185,7 @@ Defines a jms-based target channel adapter. +