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 40bc9b3c16..352e771b7c 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 @@ -1,6 +1,6 @@ file-source=org.springframework.integration.adapter.file.config.FileSourceAdapterParser file-target=org.springframework.integration.adapter.file.config.FileTargetAdapterParser -ftp-source=org.springframework.integration.adapter.ftp.config.FtpSourceAdapterParser +ftp-source=org.springframework.integration.adapter.ftp.config.FtpSourceParser httpinvoker-source=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerSourceAdapterParser httpinvoker-target=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerTargetAdapterParser jms-source=org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser 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 d22c720c4d..732b40436c 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 @@ -25,6 +25,7 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.RequestReplyTemplate; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; +import org.springframework.util.Assert; /** * A source adapter that implements the {@link MessageHandler} interface. It may @@ -32,10 +33,12 @@ import org.springframework.integration.message.Message; * * @author Mark Fisher */ -public class MessageHandlingSourceAdapter extends AbstractSourceAdapter implements MessageHandler, InitializingBean { +public class MessageHandlingSourceAdapter implements MessageHandler, InitializingBean { private final Log logger = LogFactory.getLog(this.getClass()); + private final MessageChannel channel; + private volatile RequestReplyTemplate requestReplyTemplate; private volatile boolean expectReply = true; @@ -56,7 +59,8 @@ public class MessageHandlingSourceAdapter extends AbstractSourceAdapter implemen * null. */ public MessageHandlingSourceAdapter(MessageChannel channel) { - super(channel); + Assert.notNull(channel, "channel must not be null"); + this.channel = channel; } @@ -68,10 +72,18 @@ public class MessageHandlingSourceAdapter extends AbstractSourceAdapter implemen this.expectReply = expectReply; } + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + public void setReceiveTimeout(long receiveTimeout) { this.receiveTimeout = receiveTimeout; } + protected MessageChannel getChannel() { + return this.channel; + } + public final void afterPropertiesSet() throws Exception { synchronized (this.lifecycleMonitor) { if (this.initialized) { @@ -92,7 +104,7 @@ public class MessageHandlingSourceAdapter extends AbstractSourceAdapter implemen } private RequestReplyTemplate createRequestReplyTemplate() { - RequestReplyTemplate template = new RequestReplyTemplate(this.getChannel()); + RequestReplyTemplate template = new RequestReplyTemplate(this.channel); template.setDefaultSendTimeout(this.sendTimeout); template.setDefaultReceiveTimeout(this.receiveTimeout); return template; @@ -108,7 +120,8 @@ public class MessageHandlingSourceAdapter extends AbstractSourceAdapter implemen } } if (!this.expectReply) { - if (!this.sendToChannel(message) && logger.isWarnEnabled()) { + boolean sent = (this.sendTimeout < 0) ? this.channel.send(message) : this.channel.send(message, this.sendTimeout); + if (!sent && logger.isWarnEnabled()) { logger.warn("failed to send message to channel within timeout of " + this.sendTimeout + " milliseconds"); } return null; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractPollingSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractPollingSourceAdapterParser.java deleted file mode 100644 index f4bd4099f9..0000000000 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractPollingSourceAdapterParser.java +++ /dev/null @@ -1,121 +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.adapter.config; - -import org.w3c.dom.Element; - -import org.springframework.beans.factory.BeanDefinitionStoreException; -import org.springframework.beans.factory.support.AbstractBeanDefinition; -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; -import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; -import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.adapter.PollingSourceAdapter; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Schedule; -import org.springframework.util.StringUtils; - -/** - * Base parser for polling adapters. - * - * @author Mark Fisher - */ -public abstract class AbstractPollingSourceAdapterParser extends AbstractSingleBeanDefinitionParser { - - protected final Class getBeanClass(Element element) { - return PollingSourceAdapter.class; - } - - protected boolean shouldGenerateId() { - return false; - } - - protected boolean shouldGenerateIdAsFallback() { - return true; - } - - protected boolean isEligibleAttribute(String attributeName) { - return !ID_ATTRIBUTE.equals(attributeName) && !"channel".equals(attributeName) && !"poll-period".equals(attributeName) - && !shouldSkipAttribute(attributeName); - } - - protected boolean shouldSkipAttribute(String attributeName) { - return false; - } - - protected void doPostProcess(BeanDefinitionBuilder beanDefinition, Element element) { - } - - @Override - protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - String channel = element.getAttribute("channel"); - if (!StringUtils.hasText(channel)) { - throw new ConfigurationException("'channel' is required"); - } - SourceParser sourceParser = new SourceParser(); - sourceParser.parse(element, parserContext); - builder.addConstructorArgReference(sourceParser.generatedName); - builder.addConstructorArgReference(channel); - builder.addConstructorArgValue(this.parseSchedule(element)); - } - - /** - * Subclasses may override this method to control the creation of the {@link Schedule}. The default - * implementation creates a {@link PollingSchedule} instance based on the provided "poll-period" attribute. - */ - protected Schedule parseSchedule(Element element) { - String period = element.getAttribute("poll-period"); - if (!StringUtils.hasText(period)) { - throw new ConfigurationException("'poll-period' is required"); - } - PollingSchedule schedule = new PollingSchedule(Long.valueOf(period)); - return schedule; - } - - protected abstract Class> getSourceBeanClass(Element element); - - - private class SourceParser extends AbstractSimpleBeanDefinitionParser { - - private String generatedName; - - @Override - protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) - throws BeanDefinitionStoreException { - this.generatedName = parserContext.getReaderContext().generateBeanName(definition); - return this.generatedName; - } - - @Override - protected Class getBeanClass(Element element) { - return getSourceBeanClass(element); - } - - @Override - protected boolean isEligibleAttribute(String attributeName) { - return AbstractPollingSourceAdapterParser.this.isEligibleAttribute(attributeName); - } - - @Override - protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { - doPostProcess(beanDefinition, 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 7bb54af5fb..ea771e85a6 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 @@ -25,8 +25,6 @@ - - @@ -57,8 +55,6 @@ - - @@ -71,12 +67,11 @@ + - - diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java index 5e324aa0bc..d0b3e7b3d7 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java @@ -21,7 +21,7 @@ import java.util.List; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; -import org.springframework.integration.adapter.AbstractSourceAdapter; +import org.springframework.integration.channel.ChannelPublisher; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.util.Assert; @@ -33,7 +33,7 @@ import org.springframework.util.CollectionUtils; * * @author Mark Fisher */ -public class ApplicationEventSourceAdapter extends AbstractSourceAdapter implements ApplicationListener { +public class ApplicationEventSourceAdapter extends ChannelPublisher implements ApplicationListener { private List> eventTypes = new ArrayList>(); @@ -67,7 +67,7 @@ public class ApplicationEventSourceAdapter extends AbstractSourceAdapter impleme } private boolean sendMessage(ApplicationEvent event) { - return this.sendToChannel(new GenericMessage(event)); + return this.publish(new GenericMessage(event)); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java index fd317bb409..b72d038810 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java @@ -19,30 +19,29 @@ package org.springframework.integration.adapter.file.config; import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.integration.adapter.config.AbstractPollingSourceAdapterParser; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; import org.springframework.integration.adapter.file.FileSource; -import org.springframework.integration.message.PollableSource; /** * Parser for the <file-source/> element. * * @author Mark Fisher */ -public class FileSourceAdapterParser extends AbstractPollingSourceAdapterParser { +public class FileSourceAdapterParser extends AbstractSimpleBeanDefinitionParser { @Override - protected Class> getSourceBeanClass(Element element) { + protected Class getBeanClass(Element element) { return FileSource.class; } @Override - protected boolean shouldSkipAttribute(String attributeName) { - return "directory".equals(attributeName); + protected boolean isEligibleAttribute(String attributeName) { + return (!"directory".equals(attributeName)) && super.isEligibleAttribute(attributeName); } @Override - protected void doPostProcess(BeanDefinitionBuilder builder, Element element) { - builder.addConstructorArgValue(element.getAttribute("directory")); + protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { + beanDefinition.addConstructorArgValue(element.getAttribute("directory")); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java similarity index 72% rename from spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java rename to spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java index 324102bd13..f02df80485 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java @@ -18,20 +18,18 @@ package org.springframework.integration.adapter.ftp.config; import org.w3c.dom.Element; -import org.springframework.integration.adapter.config.AbstractPollingSourceAdapterParser; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; import org.springframework.integration.adapter.ftp.FtpSource; -import org.springframework.integration.message.PollableSource; /** * Parser for the <ftp-source/> element. * - * @author Marius Bogoevici * @author Mark Fisher */ -public class FtpSourceAdapterParser extends AbstractPollingSourceAdapterParser { +public class FtpSourceParser extends AbstractSimpleBeanDefinitionParser { @Override - protected Class> getSourceBeanClass(Element element) { + protected Class getBeanClass(Element element) { return FtpSource.class; } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java index 3bb24ac4be..6d03f0fed8 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java @@ -18,12 +18,12 @@ package org.springframework.integration.adapter.jms; import javax.jms.MessageListener; -import org.springframework.integration.ConfigurationException; +import org.springframework.integration.channel.ChannelPublisher; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessagingException; import org.springframework.jms.support.converter.MessageConverter; -import org.springframework.util.Assert; /** * JMS {@link MessageListener} implementation that converts the received JMS @@ -31,38 +31,23 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class ChannelPublishingJmsListener implements MessageListener { - - private final MessageChannel channel; +public class ChannelPublishingJmsListener extends ChannelPublisher implements MessageListener { private final MessageConverter converter; - private volatile long timeout = -1; - public ChannelPublishingJmsListener(MessageChannel channel, MessageConverter converter) { - Assert.notNull(channel, "'channel' must not be null"); - this.channel = channel; + super(channel); this.converter = (converter != null && converter instanceof HeaderMappingMessageConverter) ? converter : new HeaderMappingMessageConverter(converter); } - public void setTimeout(long timeout) { - this.timeout = timeout; - } - public void onMessage(javax.jms.Message jmsMessage) { - if (this.channel == null) { - throw new ConfigurationException("'channel' must not be null"); - } try { Message messageToSend = (Message) this.converter.fromMessage(jmsMessage); - if (this.timeout < 0) { - this.channel.send(messageToSend); - } - else { - this.channel.send(messageToSend, timeout); + if (!this.publish(messageToSend)){ + throw new MessageDeliveryException(messageToSend, "failed to send Message to channel: " + this.getChannel()); } } catch (Exception e) { 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 49d6701d41..a2ad6d9ed2 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 @@ -21,12 +21,12 @@ import javax.jms.Destination; import javax.jms.Session; import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.ConfigurationException; -import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.SubscribableSource; +import org.springframework.integration.message.Target; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.support.converter.MessageConverter; @@ -38,7 +38,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class JmsMessageDrivenSourceAdapter implements SourceAdapter, Lifecycle, InitializingBean, DisposableBean { +public class JmsMessageDrivenSourceAdapter implements SubscribableSource, Lifecycle, DisposableBean { private volatile MessageChannel channel; @@ -70,6 +70,27 @@ public class JmsMessageDrivenSourceAdapter implements SourceAdapter, Lifecycle, private volatile long sendTimeout = -1; + 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; @@ -116,21 +137,19 @@ public class JmsMessageDrivenSourceAdapter implements SourceAdapter, Lifecycle, this.sessionAcknowledgeMode = sessionAcknowledgeMode; } - public void afterPropertiesSet() { + private void initialize() { if (this.channel == null) { throw new ConfigurationException("channel must not be null"); } - this.initContainer(); - } - - private void initContainer() { if (this.container == null) { this.container = createDefaultContainer(); } ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel(), this.messageConverter); listener.setTimeout(this.sendTimeout); this.container.setMessageListener(listener); - this.container.afterPropertiesSet(); + if (!this.container.isActive()) { + this.container.afterPropertiesSet(); + } } private AbstractMessageListenerContainer createDefaultContainer() { @@ -161,19 +180,24 @@ public class JmsMessageDrivenSourceAdapter implements SourceAdapter, Lifecycle, } public boolean isRunning() { - return container.isRunning(); + return (this.container != null && this.container.isRunning()); } public void start() { - container.start(); + this.initialize(); + this.container.start(); } public void stop() { - container.stop(); + if (this.container != null) { + this.container.stop(); + } } public void destroy() { - container.destroy(); + if (this.container != null) { + this.container.destroy(); + } } } 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 b81816ce8d..b39ede5b87 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 @@ -21,16 +21,12 @@ import javax.jms.Session; import org.w3c.dom.Element; import org.springframework.beans.factory.BeanCreationException; -import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.jms.JmsMessageDrivenSourceAdapter; import org.springframework.integration.adapter.jms.JmsPollableSource; -import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.util.StringUtils; /** @@ -40,8 +36,6 @@ import org.springframework.util.StringUtils; */ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { - private static final String POLL_PERIOD_ATTRIBUTE = "poll-period"; - private static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter"; private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter"; @@ -67,18 +61,14 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { @Override protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) { - if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) { - return parsePollingSourceAdapter(element, parserContext); + if ("true".equals(element.getAttribute("message-driven"))) { + return parseMessageDrivenSource(element, parserContext); } - return parseMessageDrivenSourceAdapter(element, parserContext); + return parsePollableSource(element, parserContext); } - private AbstractBeanDefinition parsePollingSourceAdapter(Element element, ParserContext parserContext) { - BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(JmsPollableSource.class); - String pollPeriod = element.getAttribute(POLL_PERIOD_ATTRIBUTE); - if (!StringUtils.hasText(pollPeriod)) { - throw new BeanCreationException("'" + POLL_PERIOD_ATTRIBUTE + "' is required for a polling JMS adapter"); - } + 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. " + @@ -98,15 +88,15 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { "', '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "', or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' should be provided."); } - sourceBuilder.addConstructorArgReference(jmsTemplate); + builder.addConstructorArgReference(jmsTemplate); } else if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { - sourceBuilder.addConstructorArgReference(JmsAdapterParserUtils.determineConnectionFactoryBeanName(element)); + builder.addConstructorArgReference(JmsAdapterParserUtils.determineConnectionFactoryBeanName(element)); if (StringUtils.hasText(destination)) { - sourceBuilder.addConstructorArgReference(destination); + builder.addConstructorArgReference(destination); } else if (StringUtils.hasText(destinationName)) { - sourceBuilder.addConstructorArgValue(destinationName); + builder.addConstructorArgValue(destinationName); } } else { @@ -114,20 +104,10 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' attributes must be provided for a polling JMS adapter"); } - String channel = element.getAttribute(JmsAdapterParserUtils.CHANNEL_ATTRIBUTE); - PollingSchedule schedule = new PollingSchedule(Long.valueOf(pollPeriod)); - BeanDefinition sourceDef = sourceBuilder.getBeanDefinition(); - String sourceBeanName = parserContext.getReaderContext().generateBeanName(sourceDef); - BeanComponentDefinition sourceComponent = new BeanComponentDefinition(sourceDef, sourceBeanName); - parserContext.registerBeanComponent(sourceComponent); - BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(PollingSourceAdapter.class); - adapterBuilder.addConstructorArgReference(sourceBeanName); - adapterBuilder.addConstructorArgReference(channel); - adapterBuilder.addConstructorArgValue(schedule); - return adapterBuilder.getBeanDefinition(); + return builder.getBeanDefinition(); } - private AbstractBeanDefinition parseMessageDrivenSourceAdapter(Element element, ParserContext parserContext) { + 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); @@ -164,8 +144,6 @@ public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); } } - String channel = element.getAttribute(JmsAdapterParserUtils.CHANNEL_ATTRIBUTE); - builder.addPropertyReference(JmsAdapterParserUtils.CHANNEL_PROPERTY, channel); return builder.getBeanDefinition(); } diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java index 7f29f8878f..9ddded6c0a 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java @@ -25,10 +25,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.file.FileSource; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -38,15 +35,8 @@ public class FileSourceAdapterParserTests { @Test public void testFileSourceAdapterParser() { ApplicationContext context = new ClassPathXmlApplicationContext("fileSourceAdapterParserTests.xml", this.getClass()); - PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); - DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); - PollingSchedule schedule = (PollingSchedule) adapterAccessor.getPropertyValue("schedule"); - assertEquals(1234, schedule.getPeriod()); - MessageChannel channel = (MessageChannel) context.getBean("testChannel"); - assertEquals(channel, adapterAccessor.getPropertyValue("channel")); - Object source = adapterAccessor.getPropertyValue("source"); - assertEquals(FileSource.class, source.getClass()); - DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(source); + FileSource fileSource = (FileSource) context.getBean("fileSource"); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(fileSource); File directory = (File) sourceAccessor.getPropertyValue("directory"); assertEquals(System.getProperty("java.io.tmpdir"), directory.getAbsolutePath()); } diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml index 37eec4f644..aff5d2137c 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml @@ -10,11 +10,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/ftp/config/FtpSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java similarity index 61% rename from spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java rename to spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java index f9526903e2..907e2172a1 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java @@ -18,29 +18,31 @@ package org.springframework.integration.adapter.ftp.config; import static org.junit.Assert.assertEquals; +import java.io.File; + import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.ftp.FtpSource; -import org.springframework.integration.scheduling.PollingSchedule; /** - * @author Marius Bogoevici * @author Mark Fisher */ -public class FtpSourceAdapterParserTests { +public class FtpSourceParserTests { @Test public void testFtpSourceAdapterParser() { - ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceAdapterParserTests.xml", this.getClass()); - PollingSourceAdapter ftpAdapter = (PollingSourceAdapter) context.getBean("ftpAdapter"); - DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(ftpAdapter); - assertEquals(FtpSource.class, adapterAccessor.getPropertyValue("source").getClass()); - assertEquals(context.getBean("testChannel"), adapterAccessor.getPropertyValue("channel")); - assertEquals(12345L, ((PollingSchedule) adapterAccessor.getPropertyValue("schedule")).getPeriod()); + ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); + FtpSource ftpSource = (FtpSource) context.getBean("ftpSource"); + DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + assertEquals("testHost", accessor.getPropertyValue("host")); + assertEquals(2121, accessor.getPropertyValue("port")); + assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); + assertEquals("testUser", accessor.getPropertyValue("username")); + assertEquals("testPassword", accessor.getPropertyValue("password")); } } diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml similarity index 64% rename from spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml rename to spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml index 09efea0447..0259dbe45f 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml @@ -7,16 +7,12 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - - - - - + username="testUser" + password="testPassword"/> + 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 d5a498b167..86670ac7f4 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 @@ -29,9 +29,11 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.jms.JmsMessageDrivenSourceAdapter; +import org.springframework.integration.adapter.jms.JmsPollableSource; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.PollingSourceEndpoint; import org.springframework.integration.message.Message; import org.springframework.jms.support.converter.MessageConversionException; import org.springframework.jms.support.converter.MessageConverter; @@ -45,25 +47,18 @@ public class JmsSourceAdapterParserTests { public void testPollingAdapterWithJmsTemplate() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithJmsTemplate.xml", this.getClass()); - context.start(); - PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); - adapter.run(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); - Message message = channel.receive(500); + JmsPollableSource source = (JmsPollableSource) context.getBean("jmsSource"); + Message message = source.receive(); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); - context.stop(); } @Test public void testPollingAdapterWithConnectionFactoryAndDestination() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithConnectionFactoryAndDestination.xml", this.getClass()); - context.start(); - PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); - adapter.run(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); - Message message = channel.receive(500); + JmsPollableSource source = (JmsPollableSource) context.getBean("jmsSource"); + Message message = source.receive(); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); context.stop(); @@ -73,11 +68,8 @@ public class JmsSourceAdapterParserTests { public void testPollingAdapterWithConnectionFactoryAndDestinationName() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); - context.start(); - PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); - adapter.run(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); - Message message = channel.receive(500); + JmsPollableSource source = (JmsPollableSource) context.getBean("jmsSource"); + Message message = source.receive(); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); context.stop(); @@ -87,10 +79,10 @@ public class JmsSourceAdapterParserTests { public void testMessageDrivenAdapterWithConnectionFactoryAndDestination() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageDrivenAdapterWithConnectionFactoryAndDestination.xml", this.getClass()); + MessageChannel channel = new QueueChannel(1); + JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); + source.setChannel(channel); context.start(); - JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter"); - assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("channel"); Message message = channel.receive(3000); assertNotNull("message should not be null", message); assertEquals("message-driven-test", message.getPayload()); @@ -101,10 +93,11 @@ public class JmsSourceAdapterParserTests { public void testMessageDrivenAdapterWithConnectionFactoryAndDestinationName() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); + MessageChannel channel = new QueueChannel(1); + JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); + source.setChannel(channel); context.start(); - JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter"); - assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("channel"); + assertEquals(JmsMessageDrivenSourceAdapter.class, source.getClass()); Message message = channel.receive(3000); assertNotNull("message should not be null", message); assertEquals("message-driven-test", message.getPayload()); @@ -115,9 +108,10 @@ public class JmsSourceAdapterParserTests { public void testMessageDrivenAdapterWithMessageConverter() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageDrivenAdapterWithMessageConverter.xml", this.getClass()); - JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter"); - assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("channel"); + MessageChannel channel = new QueueChannel(1); + JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); + source.setChannel(channel); + context.start(); Message message = channel.receive(3000); assertNotNull("message should not be null", message); assertEquals("converted-test-message", message.getPayload()); @@ -150,11 +144,8 @@ public class JmsSourceAdapterParserTests { public void testPollingAdapterWithDestinationAndDefaultConnectionFactory() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithDestinationAndDefaultConnectionFactory.xml", this.getClass()); - context.start(); - PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); - adapter.run(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); - Message message = channel.receive(500); + JmsPollableSource source = (JmsPollableSource) context.getBean("jmsSource"); + Message message = source.receive(); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); context.stop(); @@ -169,25 +160,10 @@ public class JmsSourceAdapterParserTests { public void testPollingAdapterWithDestinationNameAndDefaultConnectionFactory() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml", this.getClass()); - context.start(); - PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); - adapter.run(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); - Message message = channel.receive(500); + JmsPollableSource source = (JmsPollableSource) context.getBean("jmsSource"); + Message message = source.receive(); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); - context.stop(); - } - - @Test(expected=BeanDefinitionStoreException.class) - public void testPollingAdapterWithoutPollPeriod() { - try { - new ClassPathXmlApplicationContext("pollingAdapterWithoutPollPeriod.xml", this.getClass()); - } - catch (RuntimeException e) { - assertEquals(BeanCreationException.class, e.getCause().getClass()); - throw e; - } } @Test(expected=BeanDefinitionStoreException.class) @@ -216,16 +192,30 @@ public class JmsSourceAdapterParserTests { public void testMessageDrivenAdapterWithDefaultConnectionFactory() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageDrivenAdapterWithDefaultConnectionFactory.xml", this.getClass()); + MessageChannel channel = new QueueChannel(1); + JmsMessageDrivenSourceAdapter source = (JmsMessageDrivenSourceAdapter) context.getBean("jmsSource"); + source.setChannel(channel); context.start(); - JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter"); - assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("channel"); Message message = channel.receive(3000); assertNotNull("message should not be null", message); assertEquals("message-driven-test", message.getPayload()); context.stop(); } + @Test + public void testPollingJmsSourceEndpoint() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "pollingJmsSourceEndpoint.xml", this.getClass()); + context.start(); + PollingSourceEndpoint endpoint = (PollingSourceEndpoint) context.getBean("endpoint"); + assertEquals(JmsPollableSource.class, endpoint.getSource().getClass()); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message message = channel.receive(3000); + assertNotNull("message should not be null", message); + assertEquals("polling-test", message.getPayload()); + context.stop(); + } + public static class TestMessageConverter implements MessageConverter { 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 46a17462bf..835be17332 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,14 +7,9 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - - - - - + destination="testDestination"/> 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 a6345786c1..7c2b6788dd 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,12 +9,9 @@ - - - + destination-name="testDestinationName"/> 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 d34c7b0c1b..ad0fdc151e 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,9 +9,7 @@ - - - + 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 2668576511..163d070f6d 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,9 +9,7 @@ - - - + 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 d8da4d065e..53ea30c514 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,10 +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/messageDrivenAdapterWithMessageConverter.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml index 07e533fa71..f0424eadb4 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,12 +9,9 @@ - - - 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 fcf5346be5..bc40568c27 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 @@ -7,15 +7,10 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - - - - - + message-driven="false"/> 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 5f35aff9d4..051e5d30d8 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 @@ -7,15 +7,10 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - - - - - + message-driven="false"/> 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 da7fd24980..83d597cf91 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,11 +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 ac20415664..d2defb8d24 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,11 +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 781b5796a4..9004616760 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,11 +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 e38376d143..29721ed65d 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,10 +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 4e2388542e..ef8c96767e 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,11 +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 69751ee5ea..62ba77bb4e 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,11 +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/pollingAdapterWithoutPollPeriod.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml similarity index 82% rename from spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithoutPollPeriod.xml rename to spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml index 81ae4a6423..ae19d9e71a 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithoutPollPeriod.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/pollingJmsSourceEndpoint.xml @@ -11,7 +11,11 @@ - + + + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java index f9fa14f2d2..0aea3193c0 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java @@ -23,9 +23,9 @@ import java.io.ByteArrayInputStream; import org.junit.Test; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.PollingSourceEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.scheduling.PollingSchedule; @@ -42,8 +42,8 @@ public class ByteStreamSourceAdapterTests { ByteStreamSource source = new ByteStreamSource(stream); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.run(); Message message1 = channel.receive(500); byte[] payload = (byte[]) message1.getPayload(); assertEquals(3, payload.length); @@ -52,7 +52,7 @@ public class ByteStreamSourceAdapterTests { assertEquals(3, payload[2]); Message message2 = channel.receive(0); assertNull(message2); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(0); assertNull(message3); } @@ -66,9 +66,9 @@ public class ByteStreamSourceAdapterTests { source.setBytesPerMessage(8); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(5); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); Message message1 = channel.receive(500); assertEquals(8, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); @@ -84,16 +84,16 @@ public class ByteStreamSourceAdapterTests { source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(1); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(1); + endpoint.run(); Message message1 = channel.receive(0); byte[] bytes1 = (byte[]) message1.getPayload(); assertEquals(4, bytes1.length); assertEquals(0, bytes1[0]); Message message2 = channel.receive(0); assertNull(message2); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(0); byte[] bytes3 = (byte[]) message3.getPayload(); assertEquals(4, bytes3.length); @@ -109,9 +109,9 @@ public class ByteStreamSourceAdapterTests { source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(5); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); Message message1 = channel.receive(0); byte[] bytes1 = (byte[]) message1.getPayload(); assertEquals(4, bytes1.length); @@ -133,14 +133,14 @@ public class ByteStreamSourceAdapterTests { source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(1); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(1); + endpoint.run(); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(0); assertEquals(2, ((byte[]) message3.getPayload()).length); } @@ -155,14 +155,14 @@ public class ByteStreamSourceAdapterTests { source.setShouldTruncate(false); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(1); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(1); + endpoint.run(); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(0); assertEquals(4, ((byte[]) message3.getPayload()).length); assertEquals(0, ((byte[]) message3.getPayload())[3]); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java index ca72cecbc7..aecd6e80ab 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java @@ -23,9 +23,9 @@ import java.io.StringReader; import org.junit.Test; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.PollingSourceEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.scheduling.PollingSchedule; @@ -41,13 +41,13 @@ public class CharacterStreamSourceAdapterTests { CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.run(); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(0); assertNull(message3); } @@ -59,9 +59,9 @@ public class CharacterStreamSourceAdapterTests { CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(5); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); @@ -76,14 +76,14 @@ public class CharacterStreamSourceAdapterTests { CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(1); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(1); + endpoint.run(); Message message1 = channel.receive(0); assertEquals("test1", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(0); assertEquals("test2", message3.getPayload()); } @@ -96,9 +96,9 @@ public class CharacterStreamSourceAdapterTests { CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(5000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(5); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); Message message1 = channel.receive(500); assertEquals("test1", message1.getPayload()); Message message2 = channel.receive(500); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 3ae43bfc49..048880abee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -36,7 +36,6 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.Lifecycle; import org.springframework.context.event.ContextRefreshedEvent; import org.springframework.integration.ConfigurationException; -import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.DefaultChannelRegistry; @@ -47,6 +46,7 @@ import org.springframework.integration.endpoint.DefaultEndpointRegistry; import org.springframework.integration.endpoint.EndpointRegistry; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.SourceEndpoint; import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Target; @@ -79,7 +79,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio private final Map subscriptionManagers = new ConcurrentHashMap(); - private final List lifecycleSourceAdapters = new CopyOnWriteArrayList(); + private final List lifecycleEndpoints = new CopyOnWriteArrayList(); private volatile MessagingTaskScheduler taskScheduler; @@ -161,15 +161,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } } - @SuppressWarnings("unchecked") - private void registerSourceAdapters(ApplicationContext context) { - Map sourceAdapterBeans = - (Map) context.getBeansOfType(SourceAdapter.class); - for (Map.Entry entry : sourceAdapterBeans.entrySet()) { - this.registerSourceAdapter(entry.getKey(), entry.getValue()); - } - } - public void initialize() { synchronized (this.lifecycleMonitor) { if (this.initialized || this.initializing) { @@ -259,12 +250,11 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (endpoint instanceof ChannelRegistryAware) { ((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry); } - if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null - && endpoint instanceof TargetEndpoint) { - ((TargetEndpoint) endpoint).setConcurrencyPolicy(this.defaultConcurrencyPolicy); - } if (endpoint instanceof TargetEndpoint) { - ((TargetEndpoint) endpoint).afterPropertiesSet(); + this.registerTargetEndpoint(name, (TargetEndpoint) endpoint); + } + else if (endpoint instanceof SourceEndpoint) { + this.registerSourceEndpoint(name, (SourceEndpoint) endpoint); } this.endpointRegistry.registerEndpoint(name, endpoint); if (this.isRunning()) { @@ -275,18 +265,27 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } } + private void registerTargetEndpoint(String name, TargetEndpoint endpoint) { + if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null) { + endpoint.setConcurrencyPolicy(this.defaultConcurrencyPolicy); + } + endpoint.afterPropertiesSet(); + } + public MessageEndpoint unregisterEndpoint(String name) { MessageEndpoint endpoint = this.endpointRegistry.unregisterEndpoint(name); if (endpoint == null) { return null; } - Collection managers = this.subscriptionManagers.values(); - boolean removed = false; - for (SubscriptionManager manager : managers) { - removed = (removed || manager.removeTarget(endpoint)); - } - if (removed) { - return endpoint; + if (endpoint instanceof TargetEndpoint) { + Collection managers = this.subscriptionManagers.values(); + boolean removed = false; + for (SubscriptionManager manager : managers) { + removed = (removed || manager.removeTarget((TargetEndpoint) endpoint)); + } + if (removed) { + return endpoint; + } } return null; } @@ -310,6 +309,12 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } private void activateEndpoint(MessageEndpoint endpoint) { + if (endpoint instanceof TargetEndpoint) { + this.activateTargetEndpoint((TargetEndpoint) endpoint); + } + } + + private void activateTargetEndpoint(TargetEndpoint endpoint) { Subscription subscription = endpoint.getSubscription(); if (subscription == null) { throw new ConfigurationException("Unable to register endpoint '" + @@ -360,17 +365,17 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio } } - public void registerSourceAdapter(String name, SourceAdapter adapter) { + private void registerSourceEndpoint(String name, SourceEndpoint endpoint) { if (!this.initialized) { this.initialize(); } - if (adapter instanceof MessagingTask) { - this.taskScheduler.schedule((MessagingTask) adapter); + if (endpoint instanceof MessagingTask) { + this.taskScheduler.schedule((MessagingTask) endpoint); } - if (adapter instanceof Lifecycle) { - this.lifecycleSourceAdapters.add((Lifecycle) adapter); + if (endpoint instanceof Lifecycle) { + this.lifecycleEndpoints.add((Lifecycle) endpoint); if (this.isRunning()) { - ((Lifecycle) adapter).start(); + ((Lifecycle) endpoint).start(); } } if (logger.isInfoEnabled()) { @@ -415,10 +420,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio logger.info("started subscription manager '" + manager + "'"); } } - for (Lifecycle adapter : this.lifecycleSourceAdapters) { - adapter.start(); + for (Lifecycle endpoint : this.lifecycleEndpoints) { + endpoint.start(); if (logger.isInfoEnabled()) { - logger.info("started source adapter '" + adapter + "'"); + logger.info("started endpoint '" + endpoint + "'"); } } } @@ -436,10 +441,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio synchronized (this.lifecycleMonitor) { this.running = false; this.taskScheduler.stop(); - for (Lifecycle adapter : this.lifecycleSourceAdapters) { - adapter.stop(); + for (Lifecycle endpoint : this.lifecycleEndpoints) { + endpoint.stop(); if (logger.isInfoEnabled()) { - logger.info("stopped source adapter '" + adapter + "'"); + logger.info("stopped endpoint '" + endpoint + "'"); } } for (SubscriptionManager manager : this.subscriptionManagers.values()) { @@ -458,7 +463,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (event instanceof ContextRefreshedEvent) { ApplicationContext context = ((ContextRefreshedEvent) event).getApplicationContext(); this.registerEndpoints(context); - this.registerSourceAdapters(context); if (this.autoStartup) { this.start(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPublisher.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPublisher.java new file mode 100644 index 0000000000..62fb11b0ec --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPublisher.java @@ -0,0 +1,77 @@ +/* + * 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.channel; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.message.Message; +import org.springframework.util.Assert; + +/** + * Sends to a channel and provides a configurable timeout. Convenient for either + * subclassing or delegation from components that need to publish to a channel. + * + * @author Mark Fisher + */ +public class ChannelPublisher { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private volatile MessageChannel channel; + + private volatile long timeout = 0; + + + public ChannelPublisher() { + } + + public ChannelPublisher(MessageChannel channel) { + this.setChannel(channel); + } + + + public void setChannel(MessageChannel channel) { + Assert.notNull(channel, "channel must not be null"); + this.channel = channel; + } + + public void setTimeout(long timeout) { + this.timeout = timeout; + } + + protected MessageChannel getChannel() { + return this.channel; + } + + public boolean publish(Message message) { + if (this.channel == null) { + if (logger.isWarnEnabled()) { + logger.warn("unable to send message, no channel available"); + } + return false; + } + if (message == null) { + if (logger.isWarnEnabled()) { + logger.warn("null messages are not supported"); + } + return false; + } + return (this.timeout < 0) ? this.channel.send(message) : this.channel.send(message, this.timeout); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index eabae330dd..6d9e48ec72 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -27,8 +27,8 @@ import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.adapter.MethodInvokingSource; import org.springframework.integration.adapter.MethodInvokingTarget; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.endpoint.PollingSourceEndpoint; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Subscription; import org.springframework.util.StringUtils; @@ -75,7 +75,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser { RootBeanDefinition adapterDef = null; RootBeanDefinition invokerDef = null; if (this.isInbound) { - adapterDef = new RootBeanDefinition(PollingSourceAdapter.class); + adapterDef = new RootBeanDefinition(PollingSourceEndpoint.class); invokerDef = new RootBeanDefinition(MethodInvokingSource.class); invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref)); invokerDef.getPropertyValues().addPropertyValue("method", method); 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 3ee613657c..6bbcc71740 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 @@ -51,6 +51,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("priority-channel", new ChannelParser()); registerBeanDefinitionParser("source-adapter", new ChannelAdapterParser(true)); registerBeanDefinitionParser("target-adapter", new ChannelAdapterParser(false)); + registerBeanDefinitionParser("source-endpoint", new SourceEndpointParser()); registerBeanDefinitionParser("endpoint", new EndpointParser()); registerBeanDefinitionParser("handler", new HandlerParser()); registerBeanDefinitionParser("handler-chain", new HandlerParser()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 7b1e5effa4..4b6de86f96 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -36,7 +36,6 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.ConfigurationException; import org.springframework.integration.adapter.MethodInvokingSource; import org.springframework.integration.adapter.MethodInvokingTarget; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.CompletionStrategy; import org.springframework.integration.annotation.Concurrency; @@ -51,6 +50,7 @@ import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.endpoint.PollingSourceEndpoint; import org.springframework.integration.handler.AbstractMessageHandlerAdapter; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; @@ -163,10 +163,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor PollingSchedule schedule = new PollingSchedule(period); schedule.setInitialDelay(initialDelay); schedule.setFixedRate(fixedRate); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); + PollingSourceEndpoint sourceEndpoint = new PollingSourceEndpoint(source, channel, schedule); String channelName = beanName + "-inputChannel"; messageBus.registerChannel(channelName, channel); - messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); + messageBus.registerEndpoint(beanName + "-sourceEndpoint", sourceEndpoint); Subscription subscription = new Subscription(channel); endpoint.setSubscription(subscription); } 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 new file mode 100644 index 0000000000..103ca2a213 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourceEndpointParser.java @@ -0,0 +1,93 @@ +/* + * 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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +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; +import org.springframework.util.xml.DomUtils; + +/** + * Parser for the element. + * + * @author Mark Fisher + */ +public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser { + + protected final Class getBeanClass(Element element) { + if (this.getScheduleElement(element) != null) { + return PollingSourceEndpoint.class; + } + return SimpleSourceEndpoint.class; + } + + protected boolean shouldGenerateId() { + return false; + } + + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + protected boolean isEligibleAttribute(String name) { + return (!"source".equals(name) && !"channel".equals(name) && super.isEligibleAttribute(name)); + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + String source = element.getAttribute("source"); + if (!StringUtils.hasText(source)) { + throw new ConfigurationException("'source' is required"); + } + String output = element.getAttribute("channel"); + if (!StringUtils.hasText(output)) { + throw new ConfigurationException("'channel' is required"); + } + builder.addConstructorArgReference(source); + builder.addConstructorArgReference(output); + Element scheduleElement = this.getScheduleElement(element); + if (scheduleElement != null) { + builder.addConstructorArgValue(this.parseSchedule(scheduleElement)); + } + } + + /** + * Subclasses may override this method to control the creation of the {@link Schedule}. The default + * implementation creates a {@link PollingSchedule} instance based on the provided "period" attribute. + */ + protected Schedule parseSchedule(Element element) { + String period = element.getAttribute("period"); + if (!StringUtils.hasText(period)) { + throw new ConfigurationException("The 'period' attribute is required for the 'schedule' element."); + } + PollingSchedule schedule = new PollingSchedule(Long.valueOf(period)); + return schedule; + } + + private Element getScheduleElement(Element element) { + return DomUtils.getChildElementByTagName(element, "schedule"); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 487bd286d6..8e682b716e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -88,6 +88,25 @@ + + + + + Defines a source endpoint. + + + + + + + + + + + + + + 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/endpoint/AbstractSourceEndpoint.java similarity index 55% rename from spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java index b1c9f6c52b..61054b263b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java @@ -14,41 +14,53 @@ * limitations under the License. */ -package org.springframework.integration.adapter; +package org.springframework.integration.endpoint; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.Message; +import org.springframework.integration.message.Source; import org.springframework.util.Assert; /** + * Base class for {@link SourceEndpoint} implementations. + * * @author Mark Fisher */ -public abstract class AbstractSourceAdapter implements SourceAdapter { +public abstract class AbstractSourceEndpoint implements SourceEndpoint { + + protected final Log logger = LogFactory.getLog(this.getClass()); + + private final Source source; private final MessageChannel channel; - private volatile long sendTimeout = -1; + private volatile String name; - public AbstractSourceAdapter(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; this.channel = channel; } + public Source getSource() { + return this.source; + } + + public void setBeanName(String name) { + this.name = name; + } + + public String getName() { + return this.name; + } + protected MessageChannel getChannel() { return this.channel; } - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - protected boolean sendToChannel(Message message) { - if (message == null) { - throw new IllegalArgumentException("message must not be null"); - } - return (this.sendTimeout < 0) ? this.channel.send(message) : this.channel.send(message, this.sendTimeout); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 194ebc295b..c26f7b647f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,23 +16,15 @@ package org.springframework.integration.endpoint; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.message.Target; -import org.springframework.integration.scheduling.Subscription; +import org.springframework.beans.factory.BeanNameAware; /** * Base interface for message endpoints. * * @author Mark Fisher */ -public interface MessageEndpoint extends Target, ChannelRegistryAware, InitializingBean, Lifecycle { +public interface MessageEndpoint extends BeanNameAware { String getName(); - Subscription getSubscription(); - - ConcurrencyPolicy getConcurrencyPolicy(); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java similarity index 65% rename from spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java index d923d19c56..79c1690afc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java @@ -14,17 +14,12 @@ * limitations under the License. */ -package org.springframework.integration.adapter; +package org.springframework.integration.endpoint; import java.util.ArrayList; import java.util.List; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessageDeliveryException; @@ -40,31 +35,26 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class PollingSourceAdapter extends AbstractSourceAdapter implements MessagingTask, InitializingBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final PollableSource source; +public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask { private final PollingSchedule schedule; + private volatile long sendTimeout = 0; + private volatile int maxMessagesPerTask = 1; - private volatile boolean initialized; - - /** - * Create a new adapter for the given source. - */ - public PollingSourceAdapter(PollableSource source, MessageChannel channel, PollingSchedule schedule) { - super(channel); - Assert.notNull(source, "source must not be null"); + public PollingSourceEndpoint(PollableSource source, MessageChannel channel, PollingSchedule schedule) { + super(source, channel); Assert.notNull(schedule, "schedule must not be null"); - this.source = source; this.schedule = schedule; } + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + public void setMaxMessagesPerTask(int maxMessagesPerTask) { Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least one"); this.maxMessagesPerTask = maxMessagesPerTask; @@ -74,18 +64,11 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements Messa return this.schedule; } - public void afterPropertiesSet() { - if (this.getChannel() instanceof SynchronousChannel) { - ((SynchronousChannel) this.getChannel()).setSource(this.source); - } - this.initialized = true; - } - public List> poll(int limit) { List> results = new ArrayList>(); int count = 0; while (count < limit) { - Message message = this.source.receive(); + Message message = ((PollableSource) this.getSource()).receive(); if (message == null) { break; } @@ -96,16 +79,16 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements Messa } protected boolean sendMessage(Message message) { - if (!this.initialized) { - this.afterPropertiesSet(); + if (message == null) { + throw new IllegalArgumentException("message must not be null"); } - boolean sent = super.sendToChannel(message); - if (this.source instanceof MessageDeliveryAware) { + boolean sent = (this.sendTimeout < 0) ? this.getChannel().send(message) : this.getChannel().send(message, this.sendTimeout); + if (this.getSource() instanceof MessageDeliveryAware) { if (sent) { - ((MessageDeliveryAware) this.source).onSend(message); + ((MessageDeliveryAware) this.getSource()).onSend(message); } else { - ((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message")); + ((MessageDeliveryAware) this.getSource()).onFailure(new MessageDeliveryException(message, "failed to send message")); } } return sent; 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 new file mode 100644 index 0000000000..21c6a6dd46 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SimpleSourceEndpoint.java @@ -0,0 +1,32 @@ +/* + * 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/adapter/SourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java similarity index 74% rename from spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java index 58e164b3cf..7ce7c276d5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java @@ -14,13 +14,17 @@ * limitations under the License. */ -package org.springframework.integration.adapter; +package org.springframework.integration.endpoint; + +import org.springframework.integration.message.Source; /** - * Base interface for source adapters. + * Base interface for source endpoints. * * @author Mark Fisher */ -public interface SourceAdapter { +public interface SourceEndpoint extends MessageEndpoint { + + Source getSource(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index 9a98f97765..3bd9be284c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -28,8 +28,9 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.handler.MessageHandlerNotRunningException; @@ -47,7 +48,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class TargetEndpoint implements MessageEndpoint, BeanNameAware { +public class TargetEndpoint implements Target, MessageEndpoint, ChannelRegistryAware, InitializingBean, Lifecycle { protected final Log logger = LogFactory.getLog(this.getClass()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml index 565d60a5e5..fb3d83e235 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml @@ -10,7 +10,7 @@ - + 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 9543bedf36..d78523005d 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 @@ -29,13 +29,12 @@ import org.junit.Test; import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.ConfigurationException; -import org.springframework.integration.adapter.PollingSourceAdapter; -import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.RendezvousChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.PollingSourceEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; @@ -174,8 +173,8 @@ public class MessageBusTests { public void testErrorChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new MessageBus(); CountDownLatch latch = new CountDownLatch(1); - SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000)); - bus.registerSourceAdapter("testAdapter", sourceAdapter); + PollingSourceEndpoint sourceEndpoint = new PollingSourceEndpoint(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000)); + bus.registerEndpoint("testEndpoint", sourceEndpoint); bus.start(); latch.await(1000, TimeUnit.MILLISECONDS); Message message = bus.getErrorChannel().receive(100); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java index 22a8bdfa20..36b3596c0d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java @@ -32,7 +32,6 @@ import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; import org.springframework.integration.handler.TestHandlers; @@ -100,7 +99,7 @@ public class SubscriptionManagerTests { MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); QueueChannel channel = new QueueChannel(); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - MessageEndpoint inactiveEndpoint = createEndpoint(handler1, true); + HandlerEndpoint inactiveEndpoint = createEndpoint(handler1, true); manager.addTarget(inactiveEndpoint); manager.addTarget(createEndpoint(handler2, true)); manager.addTarget(createEndpoint(handler3, true)); @@ -124,7 +123,7 @@ public class SubscriptionManagerTests { MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); - MessageEndpoint inactiveEndpoint = createEndpoint(handler2, true); + HandlerEndpoint inactiveEndpoint = createEndpoint(handler2, true); manager.addTarget(createEndpoint(handler1, true)); manager.addTarget(inactiveEndpoint); manager.addTarget(createEndpoint(handler3, true)); @@ -450,7 +449,7 @@ public class SubscriptionManagerTests { } - private static MessageEndpoint createEndpoint(MessageHandler handler, boolean asynchronous) { + private static HandlerEndpoint createEndpoint(MessageHandler handler, boolean asynchronous) { HandlerEndpoint endpoint = new HandlerEndpoint(handler); if (asynchronous) { endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index a8ab759c10..47eb2495f0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -28,7 +28,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.scheduling.Subscription; @@ -115,7 +115,7 @@ public class MessageBusParserTests { public void testDefaultConcurrency() { ApplicationContext context = new ClassPathXmlApplicationContext( "messageBusWithDefaultConcurrencyTests.xml", this.getClass()); - MessageEndpoint endpoint1 = (MessageEndpoint) context.getBean("endpoint1"); + TargetEndpoint endpoint1 = (TargetEndpoint) context.getBean("endpoint1"); assertEquals(4, endpoint1.getConcurrencyPolicy().getCoreSize()); assertEquals(7, endpoint1.getConcurrencyPolicy().getMaxSize()); } @@ -124,7 +124,7 @@ public class MessageBusParserTests { public void testExplicitConcurrencyTakesPrecedence() { ApplicationContext context = new ClassPathXmlApplicationContext( "messageBusWithDefaultConcurrencyTests.xml", this.getClass()); - MessageEndpoint endpoint2 = (MessageEndpoint) context.getBean("endpoint2"); + TargetEndpoint endpoint2 = (TargetEndpoint) context.getBean("endpoint2"); assertEquals(14, endpoint2.getConcurrencyPolicy().getCoreSize()); assertEquals(17, endpoint2.getConcurrencyPolicy().getMaxSize()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java similarity index 87% rename from spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java index 46a528ac14..b687e221a5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.adapter; +package org.springframework.integration.endpoint; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -33,15 +33,15 @@ import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher */ -public class PollingSourceAdapterTests { +public class PollingSourceEndpointTests { @Test public void testPolledSourceSendsToChannel() { TestSource source = new TestSource("testing", 1); QueueChannel channel = new QueueChannel(); PollingSchedule schedule = new PollingSchedule(100); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.run(); Message message = channel.receive(1000); assertNotNull("message should not be null", message); assertEquals("testing.1", message.getPayload()); @@ -53,16 +53,16 @@ public class PollingSourceAdapterTests { QueueChannel channel = new QueueChannel(1); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setSendTimeout(10); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setSendTimeout(10); + endpoint.run(); Message message1 = channel.receive(1000); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); Message message2 = channel.receive(0); assertNull("second message should be null", message2); source.resetCounter(); - adapter.run(); + endpoint.run(); Message message3 = channel.receive(100); assertNotNull("third message should not be null", message3); assertEquals("testing.1", message3.getPayload()); @@ -74,9 +74,9 @@ public class PollingSourceAdapterTests { QueueChannel channel = new QueueChannel(); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); - PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); - adapter.setMaxMessagesPerTask(5); - adapter.run(); + PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule); + endpoint.setMaxMessagesPerTask(5); + endpoint.run(); Message message1 = channel.receive(0); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); diff --git a/spring-integration-samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo.xml b/spring-integration-samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo.xml index 2bf9be1813..0a2d77bd62 100644 --- a/spring-integration-samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo.xml +++ b/spring-integration-samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo.xml @@ -1,29 +1,31 @@ - - + - + - + - + - + + + - + - + - + - + + +