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 a938bbf069..d22c720c4d 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,7 +25,6 @@ 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 @@ -33,12 +32,10 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessageHandlingSourceAdapter implements SourceAdapter, MessageHandler, InitializingBean { +public class MessageHandlingSourceAdapter extends AbstractSourceAdapter implements MessageHandler, InitializingBean { private final Log logger = LogFactory.getLog(this.getClass()); - private volatile MessageChannel channel; - private volatile RequestReplyTemplate requestReplyTemplate; private volatile boolean expectReply = true; @@ -59,28 +56,9 @@ public class MessageHandlingSourceAdapter implements SourceAdapter, MessageHandl * null. */ public MessageHandlingSourceAdapter(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - this.channel = channel; + super(channel); } - /** - * No-arg constructor for configuration via setters. Note that upon - * initialization, this adapter will throw an exception if a - * {@link MessageChannel} has not been provided. - * - * @see #setChannel(MessageChannel) - */ - public MessageHandlingSourceAdapter() { - } - - - public void setChannel(MessageChannel channel) { - this.channel = channel; - } - - protected MessageChannel getChannel() { - return this.channel; - } /** * Specify whether the handle method should be expected to return a reply. @@ -90,19 +68,11 @@ public class MessageHandlingSourceAdapter implements SourceAdapter, MessageHandl this.expectReply = expectReply; } - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - public void setReceiveTimeout(long receiveTimeout) { this.receiveTimeout = receiveTimeout; } public final void afterPropertiesSet() throws Exception { - if (this.channel == null) { - throw new ConfigurationException("The 'channel' property of '" + this.getClass().getName() - + "' must not be null."); - } synchronized (this.lifecycleMonitor) { if (this.initialized) { return; @@ -122,7 +92,7 @@ public class MessageHandlingSourceAdapter implements SourceAdapter, MessageHandl } private RequestReplyTemplate createRequestReplyTemplate() { - RequestReplyTemplate template = new RequestReplyTemplate(this.channel); + RequestReplyTemplate template = new RequestReplyTemplate(this.getChannel()); template.setDefaultSendTimeout(this.sendTimeout); template.setDefaultReceiveTimeout(this.receiveTimeout); return template; @@ -138,9 +108,8 @@ public class MessageHandlingSourceAdapter implements SourceAdapter, MessageHandl } } if (!this.expectReply) { - if (!this.channel.send(message, this.sendTimeout) && logger.isWarnEnabled()) { - logger.warn("failed to send message to channel '" + this.channel + "' within timeout of " - + this.sendTimeout + " milliseconds"); + if (!this.sendToChannel(message) && 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 new file mode 100644 index 0000000000..f4bd4099f9 --- /dev/null +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractPollingSourceAdapterParser.java @@ -0,0 +1,121 @@ +/* + * 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/AbstractRequestReplySourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java index f98d3553f4..efab9e8cc6 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java @@ -50,7 +50,7 @@ public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSi @Override protected boolean isEligibleAttribute(String attributeName) { - return !attributeName.equals("name") && super.isEligibleAttribute(attributeName); + return !attributeName.equals("name") && !attributeName.equals("channel") && super.isEligibleAttribute(attributeName); } @Override @@ -59,7 +59,7 @@ public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSi if (!StringUtils.hasText(channelRef)) { throw new ConfigurationException("a 'channel' reference is required"); } - builder.addPropertyReference("channel", channelRef); + builder.addConstructorArgReference(channelRef); builder.addPropertyValue("expectReply", element.getAttribute("expect-reply").equals("true")); String sendTimeout = element.getAttribute("send-timeout"); if (StringUtils.hasText(sendTimeout)) { 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 8bb81bd4eb..7bb54af5fb 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd @@ -58,7 +58,7 @@ - + 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 11ca0f269e..5e324aa0bc 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 @@ -22,22 +22,27 @@ 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.MessageChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; /** - * A source adapter for passing Spring + * A message source for passing Spring * {@link ApplicationEvent ApplicationEvents} within messages. * * @author Mark Fisher */ -public class ApplicationEventSourceAdapter extends AbstractSourceAdapter implements - ApplicationListener { +public class ApplicationEventSourceAdapter extends AbstractSourceAdapter implements ApplicationListener { private List> eventTypes = new ArrayList>(); + public ApplicationEventSourceAdapter(MessageChannel channel) { + super(channel); + } + + /** * Set the list of event types (classes that extend ApplicationEvent) that * this adapter should send to the message channel. By default, all event @@ -61,8 +66,8 @@ public class ApplicationEventSourceAdapter extends AbstractSourceAdapter(event)); + private boolean sendMessage(ApplicationEvent event) { + return this.sendToChannel(new GenericMessage(event)); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java similarity index 90% rename from spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java rename to spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java index e0c7fd8f1a..9bcd8c52c7 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -20,7 +20,7 @@ import java.io.File; import java.io.FileFilter; import java.io.FilenameFilter; -import org.springframework.integration.adapter.PollingSourceAdapter; +import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.PollableSource; @@ -31,7 +31,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class FileSourceAdapter extends PollingSourceAdapter implements PollableSource { +public class FileSource implements PollableSource, InitializingBean { private final File directory; @@ -46,7 +46,7 @@ public class FileSourceAdapter extends PollingSourceAdapter implements P private volatile FilenameFilter filenameFilter; - public FileSourceAdapter(File directory) { + public FileSource(File directory) { Assert.notNull(directory, "directory must not be null"); this.directory = directory; } @@ -72,9 +72,7 @@ public class FileSourceAdapter extends PollingSourceAdapter implements P this.fileNameGenerator = fileNameGenerator; } - @Override - protected void initialize() { - this.setSource(this); + public void afterPropertiesSet() { if (this.isTextBased()) { this.mapper = new TextFileMapper(this.directory); } @@ -84,7 +82,6 @@ public class FileSourceAdapter extends PollingSourceAdapter implements P if (this.fileNameGenerator != null) { this.mapper.setFileNameGenerator(this.fileNameGenerator); } - super.initialize(); } public Message poll() { 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 91c3a65cae..fd317bb409 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,35 +19,30 @@ package org.springframework.integration.adapter.file.config; import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; -import org.springframework.integration.adapter.file.FileSourceAdapter; +import org.springframework.integration.adapter.config.AbstractPollingSourceAdapterParser; +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 AbstractSingleBeanDefinitionParser { +public class FileSourceAdapterParser extends AbstractPollingSourceAdapterParser { - protected Class getBeanClass(Element element) { - return FileSourceAdapter.class; + @Override + protected Class> getSourceBeanClass(Element element) { + return FileSource.class; } - protected boolean shouldGenerateId() { - return false; + @Override + protected boolean shouldSkipAttribute(String attributeName) { + return "directory".equals(attributeName); } - protected boolean shouldGenerateIdAsFallback() { - return true; - } - - protected void doParse(Element element, BeanDefinitionBuilder builder) { - String directory = element.getAttribute("directory"); - String channel = element.getAttribute("channel"); - String pollPeriod = element.getAttribute("poll-period"); - builder.addConstructorArgValue(directory); - builder.addPropertyReference("channel", channel); - builder.addPropertyValue("period", pollPeriod); + @Override + protected void doPostProcess(BeanDefinitionBuilder builder, Element element) { + builder.addConstructorArgValue(element.getAttribute("directory")); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java similarity index 93% rename from spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java rename to spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java index 795947fe3c..545b043b41 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java @@ -28,11 +28,11 @@ import org.apache.commons.net.ftp.FTP; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; -import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.file.ByteArrayFileMapper; import org.springframework.integration.adapter.file.FileNameGenerator; import org.springframework.integration.adapter.file.TextFileMapper; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessageMapper; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.PollableSource; @@ -45,7 +45,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Mark Fisher */ -public class FtpSourceAdapter extends PollingSourceAdapter implements PollableSource { +public class FtpSource implements PollableSource, MessageDeliveryAware { private final static String DEFAULT_HOST = "localhost"; @@ -104,37 +104,22 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Po } public boolean isTextBased() { - return textBased; + return this.textBased; } public void setTextBased(boolean textBased) { this.textBased = textBased; } - @Override - protected void initialize() { - this.setSource(this); + public void afterPropertiesSet() { if (this.isTextBased()) { this.mapper = new TextFileMapper(this.localWorkingDirectory); } else { this.mapper = new ByteArrayFileMapper(this.localWorkingDirectory); } - super.initialize(); } - @Override - protected void onSend(Message message) { - String filename = message.getHeader().getProperty(FileNameGenerator.FILENAME_PROPERTY_KEY); - if (StringUtils.hasText(filename)) { - this.directoryContentManager.fileProcessed(filename); - } - else if (this.logger.isWarnEnabled()) { - logger.warn("No filename in Message header, cannot send notification of processing."); - } - } - - public final Message poll() { try { this.establishConnection(); @@ -195,4 +180,20 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Po } } + public void onSend(Message message) { + String filename = message.getHeader().getProperty(FileNameGenerator.FILENAME_PROPERTY_KEY); + if (StringUtils.hasText(filename)) { + this.directoryContentManager.fileProcessed(filename); + } + else if (this.logger.isWarnEnabled()) { + logger.warn("No filename in Message header, cannot send notification of processing."); + } + } + + public void onFailure(MessagingException exception) { + if (this.logger.isWarnEnabled()) { + logger.warn("FtpSource received failure notifcation", exception); + } + } + } 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/FtpSourceAdapterParser.java index c7cfe13c8b..324102bd13 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/FtpSourceAdapterParser.java @@ -18,45 +18,21 @@ package org.springframework.integration.adapter.ftp.config; import org.w3c.dom.Element; -import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; -import org.springframework.core.Conventions; -import org.springframework.integration.adapter.ftp.FtpSourceAdapter; -import org.springframework.util.StringUtils; +import org.springframework.integration.adapter.config.AbstractPollingSourceAdapterParser; +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 AbstractSimpleBeanDefinitionParser { - - private static final String CHANNEL_ATTRIBUTE = "channel"; - - - protected Class getBeanClass(Element element) { - return FtpSourceAdapter.class; - } - - protected boolean shouldGenerateId() { - return false; - } - - protected boolean shouldGenerateIdAsFallback() { - return true; - } +public class FtpSourceAdapterParser extends AbstractPollingSourceAdapterParser { @Override - protected boolean isEligibleAttribute(String attributeName) { - return !CHANNEL_ATTRIBUTE.equals(attributeName) && super.isEligibleAttribute(attributeName); - } - - protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { - String channelRef = element.getAttribute(CHANNEL_ATTRIBUTE); - if (StringUtils.hasText(channelRef)) { - beanDefinition.addPropertyReference( - Conventions.attributeNameToPropertyName(CHANNEL_ATTRIBUTE), channelRef); - } + protected Class> getSourceBeanClass(Element element) { + return FtpSource.class; } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java index be0d776a51..6f9ee8bbf4 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java @@ -64,13 +64,8 @@ public class HttpInvokerSourceAdapter extends MessageHandlingSourceAdapter imple private volatile HttpInvokerServiceExporter exporter; - public HttpInvokerSourceAdapter() { - super(); - } - public HttpInvokerSourceAdapter(MessageChannel channel) { - this(); - this.setChannel(channel); + super(channel); } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java index 1589a6c674..2bdfdaafc2 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java @@ -46,13 +46,16 @@ public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implement super(connectionFactory, destinationName); } - public JmsPollableSource() { - super(); - } - public Message poll() { - return new GenericMessage(this.getJmsTemplate().receiveAndConvert()); + Object receivedObject = this.getJmsTemplate().receiveAndConvert(); + if (receivedObject == null) { + return null; + } + if (receivedObject instanceof Message) { + return (Message) receivedObject; + } + return new GenericMessage(receivedObject); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java deleted file mode 100644 index 445bd02345..0000000000 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java +++ /dev/null @@ -1,44 +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.jms; - -import javax.jms.ConnectionFactory; -import javax.jms.Destination; - -import org.springframework.integration.adapter.PollingSourceAdapter; -import org.springframework.jms.core.JmsTemplate; - -/** - * A convenience adapter that wraps a {@link JmsPollableSource}. - * - * @author Mark Fisher - */ -public class JmsPollingSourceAdapter extends PollingSourceAdapter { - - public JmsPollingSourceAdapter(JmsTemplate jmsTemplate) { - super(new JmsPollableSource(jmsTemplate)); - } - - public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, Destination destination) { - super(new JmsPollableSource(connectionFactory, destination)); - } - - public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, String destinationName) { - super(new JmsPollableSource(connectionFactory, destinationName)); - } - -} diff --git a/spring-integration-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 ab3bf26cad..b81816ce8d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -21,10 +21,16 @@ 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.AbstractSingleBeanDefinitionParser; +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.JmsPollingSourceAdapter; +import org.springframework.integration.adapter.jms.JmsPollableSource; +import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.util.StringUtils; /** @@ -32,12 +38,10 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher */ -public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser { +public class JmsSourceAdapterParser extends AbstractBeanDefinitionParser { private static final String POLL_PERIOD_ATTRIBUTE = "poll-period"; - private static final String POLL_PERIOD_PROPERTY = "period"; - private static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter"; private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter"; @@ -53,13 +57,6 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser { private static final String ACKNOWLEDGE_TRANSACTED = "transacted"; - protected Class getBeanClass(Element element) { - if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) { - return JmsPollingSourceAdapter.class; - } - return JmsMessageDrivenSourceAdapter.class; - } - protected boolean shouldGenerateId() { return false; } @@ -68,31 +65,26 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser { return true; } - protected void doParse(Element element, BeanDefinitionBuilder builder) { - if (builder.getBeanDefinition().getBeanClass().equals(JmsPollingSourceAdapter.class)) { - parsePollingSourceAdapter(element, builder); + @Override + protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) { + if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) { + return parsePollingSourceAdapter(element, parserContext); } - else { - parseMessageDrivenSourceAdapter(element, builder); - } - String channel = element.getAttribute(JmsAdapterParserUtils.CHANNEL_ATTRIBUTE); - builder.addPropertyReference(JmsAdapterParserUtils.CHANNEL_PROPERTY, channel); + return parseMessageDrivenSourceAdapter(element, parserContext); } - private void parsePollingSourceAdapter(Element element, BeanDefinitionBuilder builder) { + 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 " + JmsPollingSourceAdapter.class.getSimpleName()); + throw new BeanCreationException("'" + POLL_PERIOD_ATTRIBUTE + "' is required for a polling JMS adapter"); } if (StringUtils.hasText(element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE))) { throw new BeanCreationException( - "The '" + MESSAGE_CONVERTER_ATTRIBUTE + "' attribute is not supported for a " + - JmsPollingSourceAdapter.class.getSimpleName() + ". Consider providing a '" + - JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + + "The '" + MESSAGE_CONVERTER_ATTRIBUTE + "' attribute is not supported for a polling JMS adapter. " + + ". Consider providing a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + "' reference where the template contains a 'messageConverter' property instead."); } - builder.addPropertyValue(POLL_PERIOD_PROPERTY, pollPeriod); String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE); String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); @@ -106,26 +98,37 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser { "', '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "', or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' should be provided."); } - builder.addConstructorArgReference(jmsTemplate); + sourceBuilder.addConstructorArgReference(jmsTemplate); } else if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) { - builder.addConstructorArgReference(JmsAdapterParserUtils.determineConnectionFactoryBeanName(element)); + sourceBuilder.addConstructorArgReference(JmsAdapterParserUtils.determineConnectionFactoryBeanName(element)); if (StringUtils.hasText(destination)) { - builder.addConstructorArgReference(destination); + sourceBuilder.addConstructorArgReference(destination); } else if (StringUtils.hasText(destinationName)) { - builder.addConstructorArgValue(destinationName); + sourceBuilder.addConstructorArgValue(destinationName); } } else { - throw new BeanCreationException("either a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + - "' or one of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + - JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' attributes " + - "must be provided for a " + JmsPollingSourceAdapter.class.getSimpleName()); + throw new BeanCreationException("either a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + "' or one of '" + + 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(); } - private void parseMessageDrivenSourceAdapter(Element element, BeanDefinitionBuilder builder) { + private AbstractBeanDefinition parseMessageDrivenSourceAdapter(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsMessageDrivenSourceAdapter.class); String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); String messageConverter = element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE); @@ -161,6 +164,9 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser { builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); } } + String channel = element.getAttribute(JmsAdapterParserUtils.CHANNEL_ATTRIBUTE); + builder.addPropertyReference(JmsAdapterParserUtils.CHANNEL_PROPERTY, channel); + return builder.getBeanDefinition(); } private Integer parseAcknowledgeMode(Element element) { diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java index 8fbbe3301b..a80da96682 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java @@ -43,13 +43,8 @@ public class RmiSourceAdapter extends MessageHandlingSourceAdapter { private volatile RemoteInvocationExecutor remoteInvocationExecutor; - public RmiSourceAdapter() { - super(); - } - public RmiSourceAdapter(MessageChannel channel) { - this(); - this.setChannel(channel); + super(channel); } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapter.java deleted file mode 100644 index 97fa46fcfe..0000000000 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapter.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.adapter.stream; - -import java.io.InputStream; - -import org.springframework.integration.adapter.PollingSourceAdapter; - -/** - * A polling source adapter that wraps a {@link ByteStreamSource}. - * - * @author Mark Fisher - */ -public class ByteStreamSourceAdapter extends PollingSourceAdapter { - - public ByteStreamSourceAdapter(InputStream stream) { - super(new ByteStreamSource(stream)); - } - - - public void setBytesPerMessage(int bytesPerMessage) { - ((ByteStreamSource) this.getSource()).setBytesPerMessage(bytesPerMessage); - } - - public void setShouldTruncate(boolean shouldTruncate) { - ((ByteStreamSource) this.getSource()).setShouldTruncate(shouldTruncate); - } - -} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java index 8457ce51e6..1f24502c9d 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -18,8 +18,10 @@ package org.springframework.integration.adapter.stream; import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; import java.io.Reader; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.PollableSource; import org.springframework.integration.message.StringMessage; @@ -71,4 +73,9 @@ public class CharacterStreamSource implements PollableSource { } } + + public static final CharacterStreamSource stdin() { + return new CharacterStreamSource(new InputStreamReader(System.in)); + } + } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java deleted file mode 100644 index 82bc48f867..0000000000 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java +++ /dev/null @@ -1,47 +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.stream; - -import java.io.InputStreamReader; -import java.io.Reader; - -import org.springframework.integration.adapter.PollingSourceAdapter; -import org.springframework.integration.channel.MessageChannel; - -/** - * A polling source adapter that wraps a {@link CharacterStreamSource}. - * - * @author Mark Fisher - */ -public class CharacterStreamSourceAdapter extends PollingSourceAdapter { - - public CharacterStreamSourceAdapter(Reader reader) { - super(new CharacterStreamSource(reader)); - } - - - /** - * Factory method that creates an adapter for stdin (System.in). - */ - public static CharacterStreamSourceAdapter stdinAdapter(MessageChannel channel) { - CharacterStreamSourceAdapter adapter = - new CharacterStreamSourceAdapter(new InputStreamReader(System.in)); - adapter.setChannel(channel); - return adapter; - } - -} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java index c561ee8a4c..e5b7267ab3 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -43,8 +43,7 @@ public class ApplicationEventSourceAdapterTests { @Test public void testAnyApplicationEventSentByDefault() { MessageChannel channel = new SimpleChannel(); - ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(); - adapter.setChannel(channel); + ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(channel); Message message1 = channel.receive(0); assertNull(message1); adapter.onApplicationEvent(new TestApplicationEvent1()); @@ -60,11 +59,10 @@ public class ApplicationEventSourceAdapterTests { @Test public void testOnlyConfiguredEventTypesAreSent() { MessageChannel channel = new SimpleChannel(); - ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(); + ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(channel); List> eventTypes = new ArrayList>(); eventTypes.add(TestApplicationEvent1.class); adapter.setEventTypes(eventTypes); - adapter.setChannel(channel); Message message1 = channel.receive(0); assertNull(message1); adapter.onApplicationEvent(new TestApplicationEvent1()); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml index 429ff0da6d..59161a80f9 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml @@ -9,7 +9,7 @@ - + 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 f4a1461d8c..7f29f8878f 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,7 +25,8 @@ 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.file.FileSourceAdapter; +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; @@ -37,14 +38,17 @@ public class FileSourceAdapterParserTests { @Test public void testFileSourceAdapterParser() { ApplicationContext context = new ClassPathXmlApplicationContext("fileSourceAdapterParserTests.xml", this.getClass()); - FileSourceAdapter adapter = (FileSourceAdapter) context.getBean("adapter"); - DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); - PollingSchedule schedule = (PollingSchedule) accessor.getPropertyValue("schedule"); + PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); + PollingSchedule schedule = (PollingSchedule) adapterAccessor.getPropertyValue("schedule"); assertEquals(1234, schedule.getPeriod()); - File directory = (File) accessor.getPropertyValue("directory"); - assertEquals(System.getProperty("java.io.tmpdir"), directory.getAbsolutePath()); MessageChannel channel = (MessageChannel) context.getBean("testChannel"); - assertEquals(channel, accessor.getPropertyValue("channel")); + assertEquals(channel, adapterAccessor.getPropertyValue("channel")); + Object source = adapterAccessor.getPropertyValue("source"); + assertEquals(FileSource.class, source.getClass()); + DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(source); + 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/ftp/config/FtpSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java index dfec16c294..f9526903e2 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/FtpSourceAdapterParserTests.java @@ -23,21 +23,24 @@ 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.ftp.FtpSourceAdapter; +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 { @Test public void testFtpSourceAdapterParser() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceAdapterParserTests.xml", this.getClass()); - FtpSourceAdapter ftpAdapter = (FtpSourceAdapter) context.getBean("ftpAdapter"); - DirectFieldAccessor ftpPollingAdapterAccessor = new DirectFieldAccessor(ftpAdapter); - assertEquals(context.getBean("testChannel"), ftpPollingAdapterAccessor.getPropertyValue("channel")); - assertEquals(12345L, ((PollingSchedule) ftpPollingAdapterAccessor.getPropertyValue("schedule")).getPeriod()); + 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()); } } 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/ftpSourceAdapterParserTests.xml index beaae3e5a9..09efea0447 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/ftpSourceAdapterParserTests.xml @@ -12,7 +12,7 @@ message = channel.receive(500); assertNotNull("message should not be null", message); @@ -59,8 +59,8 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithConnectionFactoryAndDestination.xml", this.getClass()); context.start(); - JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.processMessages(); + PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); + adapter.run(); MessageChannel channel = (MessageChannel) context.getBean("channel"); Message message = channel.receive(500); assertNotNull("message should not be null", message); @@ -73,8 +73,8 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); context.start(); - JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.processMessages(); + PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); + adapter.run(); MessageChannel channel = (MessageChannel) context.getBean("channel"); Message message = channel.receive(500); assertNotNull("message should not be null", message); @@ -150,8 +150,8 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithDestinationAndDefaultConnectionFactory.xml", this.getClass()); context.start(); - JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.processMessages(); + PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); + adapter.run(); MessageChannel channel = (MessageChannel) context.getBean("channel"); Message message = channel.receive(500); assertNotNull("message should not be null", message); @@ -169,8 +169,8 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithDestinationNameAndDefaultConnectionFactory.xml", this.getClass()); context.start(); - JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.processMessages(); + PollingSourceAdapter adapter = (PollingSourceAdapter) context.getBean("adapter"); + adapter.run(); MessageChannel channel = (MessageChannel) context.getBean("channel"); Message message = channel.receive(500); assertNotNull("message should not be null", message); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java index 8e8a299783..3b75380108 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -41,8 +41,7 @@ public class RmiTargetAdapterParserTests { @Before public void exportRemoteHandler() throws Exception { testChannel.setBeanName("testChannel"); - RmiSourceAdapter sourceAdapter = new RmiSourceAdapter(); - sourceAdapter.setChannel(testChannel); + RmiSourceAdapter sourceAdapter = new RmiSourceAdapter(testChannel); sourceAdapter.setExpectReply(false); sourceAdapter.afterPropertiesSet(); } 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 f9c9c6d2b1..417298b166 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -23,9 +23,11 @@ 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.SimpleChannel; import org.springframework.integration.message.Message; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -37,12 +39,11 @@ public class ByteStreamSourceAdapterTests { byte[] bytes = new byte[] {1,2,3}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); - ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); - adapter.setChannel(channel); - adapter.setInitialDelay(10000); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + ByteStreamSource source = new ByteStreamSource(stream); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); + adapter.run(); Message message1 = channel.receive(500); byte[] payload = (byte[]) message1.getPayload(); assertEquals(3, payload.length); @@ -51,7 +52,7 @@ public class ByteStreamSourceAdapterTests { assertEquals(3, payload[2]); Message message2 = channel.receive(0); assertNull(message2); - adapter.processMessages(); + adapter.run(); Message message3 = channel.receive(0); assertNull(message3); } @@ -61,14 +62,13 @@ public class ByteStreamSourceAdapterTests { byte[] bytes = new byte[] {0,1,2,3,4,5,6,7}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); - ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); - adapter.setChannel(channel); - adapter.setInitialDelay(10000); - adapter.setBytesPerMessage(8); + ByteStreamSource source = new ByteStreamSource(stream); + source.setBytesPerMessage(8); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(5); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + adapter.run(); Message message1 = channel.receive(500); assertEquals(8, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); @@ -80,21 +80,20 @@ public class ByteStreamSourceAdapterTests { byte[] bytes = new byte[] {0,1,2,3,4,5,6,7}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); - ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); - adapter.setBytesPerMessage(4); - adapter.setInitialDelay(10000); + ByteStreamSource source = new ByteStreamSource(stream); + source.setBytesPerMessage(4); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(1); - adapter.setChannel(channel); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + adapter.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.processMessages(); + adapter.run(); Message message3 = channel.receive(0); byte[] bytes3 = (byte[]) message3.getPayload(); assertEquals(4, bytes3.length); @@ -106,14 +105,13 @@ public class ByteStreamSourceAdapterTests { byte[] bytes = new byte[] {0,1,2,3,4,5,6,7}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); - ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); - adapter.setInitialDelay(10000); - adapter.setChannel(channel); - adapter.setBytesPerMessage(4); + ByteStreamSource source = new ByteStreamSource(stream); + source.setBytesPerMessage(4); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(5); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(2, count); + adapter.run(); Message message1 = channel.receive(0); byte[] bytes1 = (byte[]) message1.getPayload(); assertEquals(4, bytes1.length); @@ -131,19 +129,18 @@ public class ByteStreamSourceAdapterTests { byte[] bytes = new byte[] {0,1,2,3,4,5}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); - ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); - adapter.setInitialDelay(10000); - adapter.setBytesPerMessage(4); + ByteStreamSource source = new ByteStreamSource(stream); + source.setBytesPerMessage(4); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(1); - adapter.setChannel(channel); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + adapter.run(); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - adapter.processMessages(); + adapter.run(); Message message3 = channel.receive(0); assertEquals(2, ((byte[]) message3.getPayload()).length); } @@ -153,20 +150,19 @@ public class ByteStreamSourceAdapterTests { byte[] bytes = new byte[] {0,1,2,3,4,5}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); - ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); - adapter.setInitialDelay(10000); - adapter.setBytesPerMessage(4); - adapter.setShouldTruncate(false); + ByteStreamSource source = new ByteStreamSource(stream); + source.setBytesPerMessage(4); + source.setShouldTruncate(false); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(1); - adapter.setChannel(channel); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + adapter.run(); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - adapter.processMessages(); + adapter.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 e7e4855884..2e61778a01 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,11 @@ 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.SimpleChannel; import org.springframework.integration.message.Message; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -36,17 +38,16 @@ public class CharacterStreamSourceAdapterTests { public void testEndOfStream() { StringReader reader = new StringReader("test"); MessageChannel channel = new SimpleChannel(); - CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader); - adapter.setChannel(channel); - adapter.setInitialDelay(10000); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + CharacterStreamSource source = new CharacterStreamSource(reader); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); + adapter.run(); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - adapter.processMessages(); + adapter.run(); Message message3 = channel.receive(0); assertNull(message3); } @@ -55,13 +56,12 @@ public class CharacterStreamSourceAdapterTests { public void testEndOfStreamWithMaxMessagesPerTask() { StringReader reader = new StringReader("test"); MessageChannel channel = new SimpleChannel(); - CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader); - adapter.setInitialDelay(10000); - adapter.setChannel(channel); + CharacterStreamSource source = new CharacterStreamSource(reader); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(5); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + adapter.run(); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); @@ -73,18 +73,17 @@ public class CharacterStreamSourceAdapterTests { String s = "test1" + System.getProperty("line.separator") + "test2"; StringReader reader = new StringReader(s); MessageChannel channel = new SimpleChannel(); - CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader); - adapter.setInitialDelay(10000); + CharacterStreamSource source = new CharacterStreamSource(reader); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(1); - adapter.setChannel(channel); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(1, count); + adapter.run(); Message message1 = channel.receive(0); assertEquals("test1", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - adapter.processMessages(); + adapter.run(); Message message3 = channel.receive(0); assertEquals("test2", message3.getPayload()); } @@ -94,13 +93,12 @@ public class CharacterStreamSourceAdapterTests { String s = "test1" + System.getProperty("line.separator") + "test2"; StringReader reader = new StringReader(s); MessageChannel channel = new SimpleChannel(); - CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader); - adapter.setChannel(channel); - adapter.setInitialDelay(5000); + CharacterStreamSource source = new CharacterStreamSource(reader); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(5000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(5); - adapter.start(); - int count = adapter.processMessages(); - assertEquals(2, count); + adapter.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/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 40312a17ec..8d9605c7c2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -18,132 +18,74 @@ package org.springframework.integration.adapter; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.Executors; -import org.springframework.context.Lifecycle; -import org.springframework.integration.ConfigurationException; +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.MessageMapper; +import org.springframework.integration.message.MessageDeliveryAware; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.PollableSource; import org.springframework.integration.scheduling.MessagingTask; -import org.springframework.integration.scheduling.MessagingTaskScheduler; -import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Schedule; -import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.util.Assert; /** - * A channel adapter that retrieves objects from a {@link PollableSource}, - * delegates to a {@link MessageMapper} to create messages from those objects, + * A channel adapter that retrieves messages from a {@link PollableSource} * and then sends the resulting messages to the provided {@link MessageChannel}. * * @author Mark Fisher */ -public class PollingSourceAdapter extends AbstractSourceAdapter implements MessagingTaskSchedulerAware, Lifecycle { +public class PollingSourceAdapter extends AbstractSourceAdapter implements MessagingTask, InitializingBean { - private volatile PollableSource source; + private final Log logger = LogFactory.getLog(this.getClass()); - private volatile PollingSchedule schedule = new PollingSchedule(1000); + private final PollableSource source; - private volatile MessagingTaskScheduler scheduler; + private final PollingSchedule schedule; private volatile int maxMessagesPerTask = 1; - private volatile boolean running; - - private final Object lifecycleMonitor = new Object(); + private volatile boolean initialized; /** * Create a new adapter for the given source. */ - public PollingSourceAdapter(PollableSource source) { - this.setSource(source); - } - - /** - * No-arg constructor for providing source after construction. - */ - public PollingSourceAdapter() { - } - - - public void setSource(PollableSource source) { - Assert.notNull(source, "'source' must not be null"); + public PollingSourceAdapter(PollableSource source, MessageChannel channel, PollingSchedule schedule) { + super(channel); + Assert.notNull(source, "source must not be null"); + Assert.notNull(schedule, "schedule must not be null"); this.source = source; + this.schedule = schedule; } - public void setInitialDelay(long intialDelay) { - Assert.isTrue(intialDelay >= 0, "'intialDelay' must not be negative"); - this.schedule.setInitialDelay(intialDelay); - } - - public void setPeriod(long period) { - this.schedule.setPeriod(period); - } public void setMaxMessagesPerTask(int maxMessagesPerTask) { Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least one"); this.maxMessagesPerTask = maxMessagesPerTask; } - public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) { - Assert.notNull(scheduler, "scheduler must not be null"); - this.scheduler = scheduler; + public Schedule getSchedule() { + return this.schedule; } - protected PollableSource getSource() { - return this.source; - } - - public boolean isRunning() { - return this.running; - } - - @Override - protected void initialize() { - if (this.source == null) { - throw new ConfigurationException("source must not be null"); - } + public void afterPropertiesSet() { if (this.getChannel() instanceof SynchronousChannel) { ((SynchronousChannel) this.getChannel()).setSource(this.source); } + this.initialized = true; } - public void start() { - synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - return; - } - if (!this.isInitialized()) { - this.afterPropertiesSet(); - } - if (this.scheduler == null) { - if (logger.isInfoEnabled()) { - logger.info("no task scheduler has been provided, will create one"); - } - this.scheduler = new SimpleMessagingTaskScheduler(Executors.newSingleThreadScheduledExecutor()); - } - this.running = true; - } - if (!this.scheduler.isRunning()) { - this.scheduler.start(); - } - this.scheduler.schedule(new PollingSourceAdapterTask()); - } - - public void stop() { - this.running = false; - } - - public List> poll(int limit) { - List> results = new ArrayList>(); + public List> poll(int limit) { + List> results = new ArrayList>(); int count = 0; while (count < limit) { - Message message = this.source.poll(); + Message message = this.source.poll(); if (message == null) { break; } @@ -153,44 +95,35 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements return results; } - public int processMessages() { - if (!this.isRunning()) { - if (logger.isDebugEnabled()) { - logger.debug("source adapter not polling since it has not yet been started"); - } - return 0; + protected boolean sendMessage(Message message) { + if (!this.initialized) { + this.afterPropertiesSet(); } - int messagesProcessed = 0; - List> messages = this.poll(this.maxMessagesPerTask); - for (Message message : messages) { - if (this.sendToChannel(message)) { - messagesProcessed++; - this.onSend(message); + boolean sent = super.sendToChannel(message); + if (this.source instanceof MessageDeliveryAware) { + if (sent) { + ((MessageDeliveryAware) this.source).onSend(message); } else { - return messagesProcessed; + ((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message")); } } - return messagesProcessed; + return sent; } - /** - * Callback method invoked after a message is sent to the channel. - *

- * Subclasses may override. The default implementation does nothing. - */ - protected void onSend(Message sentMessage) { - } - - - private class PollingSourceAdapterTask implements MessagingTask { - - public void run() { - processMessages(); + public void run() { + int messagesProcessed = 0; + List> messages = this.poll(this.maxMessagesPerTask); + for (Message message : messages) { + if (this.sendMessage(message)) { + messagesProcessed++; + } + else { + break; + } } - - public Schedule getSchedule() { - return schedule; + if (logger.isDebugEnabled()) { + logger.debug("polling source task processed " + messagesProcessed + " messages"); } } 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/adapter/SourceAdapter.java index 5fe72ad111..58e164b3cf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -16,8 +16,6 @@ package org.springframework.integration.adapter; -import org.springframework.integration.channel.MessageChannel; - /** * Base interface for source adapters. * @@ -25,6 +23,4 @@ import org.springframework.integration.channel.MessageChannel; */ public interface SourceAdapter { - void setChannel(MessageChannel channel); - } 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 6c096cd754..342b0e5fda 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 @@ -50,8 +50,8 @@ import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.MessagingException; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; +import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.MessagingTaskScheduler; -import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.integration.scheduling.Subscription; @@ -355,8 +355,8 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (!this.initialized) { this.initialize(); } - if (adapter instanceof MessagingTaskSchedulerAware) { - ((MessagingTaskSchedulerAware) adapter).setMessagingTaskScheduler(this.taskScheduler); + if (adapter instanceof MessagingTask) { + this.taskScheduler.schedule((MessagingTask) adapter); } if (adapter instanceof Lifecycle) { this.lifecycleSourceAdapters.add((Lifecycle) adapter); 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 9ccaac21b6..a2f32b0e5b 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 @@ -30,6 +30,7 @@ import org.springframework.integration.adapter.MethodInvokingSource; import org.springframework.integration.adapter.MethodInvokingTarget; import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Subscription; import org.springframework.util.StringUtils; @@ -77,21 +78,22 @@ public class ChannelAdapterParser implements BeanDefinitionParser { if (this.isInbound) { adapterDef = new RootBeanDefinition(PollingSourceAdapter.class); invokerDef = new RootBeanDefinition(MethodInvokingSource.class); + String invokerBeanName = this.configureAndRegisterInvoker(invokerDef, ref, method, parserContext); String period = element.getAttribute(PERIOD_ATTRIBUTE); - if (StringUtils.hasText(period)) { - adapterDef.getPropertyValues().addPropertyValue("period", period); + if (!StringUtils.hasText(period)) { + throw new ConfigurationException("'period' is required"); } - adapterDef.getPropertyValues().addPropertyValue("channel", new RuntimeBeanReference(channel)); + PollingSchedule schedule = new PollingSchedule(Integer.valueOf(period)); + adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName)); + adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(channel)); + adapterDef.getConstructorArgumentValues().addGenericArgumentValue(schedule); } else { adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class); invokerDef = new RootBeanDefinition(MethodInvokingTarget.class); + String invokerBeanName = this.configureAndRegisterInvoker(invokerDef, ref, method, parserContext); + adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName)); } - invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref)); - invokerDef.getPropertyValues().addPropertyValue("method", method); - String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef); - parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName)); - adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName)); adapterDef.setSource(parserContext.extractSource(element)); String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { @@ -112,4 +114,12 @@ public class ChannelAdapterParser implements BeanDefinitionParser { return adapterDef; } + private String configureAndRegisterInvoker(RootBeanDefinition invokerDef, String objectRef, String methodName, ParserContext parserContext) { + invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef)); + invokerDef.getPropertyValues().addPropertyValue("method", methodName); + String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName)); + return invokerBeanName; + } + } 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 80ef182b8b..c0ed1a23c6 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 @@ -49,8 +49,7 @@ import org.springframework.integration.annotation.Router; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.AbstractMessageHandlerAdapter; @@ -161,17 +160,15 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor MethodInvokingSource source = new MethodInvokingSource(); source.setObject(bean); source.setMethod(method.getName()); - PollingSourceAdapter adapter = new PollingSourceAdapter(source); - MessageChannel channel = new SimpleChannel(); - adapter.setChannel(channel); - adapter.setPeriod(period); - String channelName = beanName + "-inputChannel"; - messageBus.registerChannel(channelName, channel); - messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); + SynchronousChannel channel = new SynchronousChannel(); PollingSchedule schedule = new PollingSchedule(period); schedule.setInitialDelay(initialDelay); schedule.setFixedRate(fixedRate); - Subscription subscription = new Subscription(channel, schedule); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); + String channelName = beanName + "-inputChannel"; + messageBus.registerChannel(channelName, channel); + messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); + Subscription subscription = new Subscription(channel); endpoint.setSubscription(subscription); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java similarity index 55% rename from spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java rename to spring-integration-core/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java index 092308eb5c..21768e09a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageDeliveryAware.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -14,16 +14,24 @@ * limitations under the License. */ -package org.springframework.integration.scheduling; +package org.springframework.integration.message; /** - * Callback interface for components that require the - * {@link MessagingTaskScheduler}. + * Interface that provides callback definitions for components that require + * message delivery status notifications. * * @author Mark Fisher */ -public interface MessagingTaskSchedulerAware { +public interface MessageDeliveryAware { - void setMessagingTaskScheduler(MessagingTaskScheduler scheduler); + /** + * Callback method invoked after a message is sent successfully. + */ + void onSend(Message sentMessage); + + /** + * Callback method invoked after a message delivery failure. + */ + void onFailure(MessagingException exception); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java index 748f6842c7..eb50874665 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/PollableSource.java @@ -23,6 +23,9 @@ package org.springframework.integration.message; */ public interface PollableSource { + /** + * Retrieve a message from this source or null if no message is available. + */ Message poll(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java index 1f09dcf4e7..9930f2ff0d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -23,6 +23,9 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.integration.util.ErrorHandler; import org.springframework.util.Assert; @@ -34,14 +37,14 @@ import org.springframework.util.Assert; */ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler { + private final Log logger = LogFactory.getLog(this.getClass()); + private final ScheduledExecutorService executor; private volatile ErrorHandler errorHandler; private final Set pendingTasks = new CopyOnWriteArraySet(); - private volatile boolean starting; - private volatile boolean running; private final Object lifecycleMonitor = new Object(); @@ -67,17 +70,21 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler public void start() { synchronized (this.lifecycleMonitor) { - if (this.running || this.starting) { + if (this.running) { return; } - this.starting = true; + this.running = true; + for (Runnable task : this.pendingTasks) { + if (logger.isDebugEnabled()) { + logger.debug("scheduling task: " + task); + } + this.schedule(task); + } + this.pendingTasks.clear(); + if (logger.isInfoEnabled()) { + logger.info("task scheduler started successfully"); + } } - for (Runnable task : this.pendingTasks) { - this.schedule(task); - } - this.pendingTasks.clear(); - this.running = true; - this.starting = false; } public void stop() { @@ -91,7 +98,7 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler @Override public ScheduledFuture schedule(Runnable task) { - if (!this.isRunning()) { + if (!this.running) { this.pendingTasks.add(task); return null; } @@ -118,9 +125,9 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler private class MessagingTaskRunner implements Runnable { - private Runnable task; + private final Runnable task; - private boolean shouldRepeat; + private volatile boolean shouldRepeat; public MessagingTaskRunner(Runnable task) { 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/adapter/PollingSourceAdapterTests.java index eda141e952..04f1553cea 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -28,6 +28,7 @@ import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.PollableSource; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -38,10 +39,9 @@ public class PollingSourceAdapterTests { public void testPolledSourceSendsToChannel() { TestSource source = new TestSource("testing", 1); SimpleChannel channel = new SimpleChannel(); - PollingSourceAdapter adapter = new PollingSourceAdapter(source); - adapter.setChannel(channel); - adapter.setPeriod(100); - adapter.start(); + PollingSchedule schedule = new PollingSchedule(100); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); + adapter.run(); Message message = channel.receive(1000); assertNotNull("message should not be null", message); assertEquals("testing.1", message.getPayload()); @@ -51,22 +51,18 @@ public class PollingSourceAdapterTests { public void testSendTimeout() { TestSource source = new TestSource("testing", 1); SimpleChannel channel = new SimpleChannel(1); - PollingSourceAdapter adapter = new PollingSourceAdapter(source); - adapter.setChannel(channel); - adapter.setInitialDelay(10000); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setSendTimeout(10); - adapter.start(); - adapter.processMessages(); - adapter.processMessages(); - adapter.stop(); + adapter.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.start(); - adapter.processMessages(); + adapter.run(); Message message3 = channel.receive(100); assertNotNull("third message should not be null", message3); assertEquals("testing.1", message3.getPayload()); @@ -76,12 +72,11 @@ public class PollingSourceAdapterTests { public void testMultipleMessagesPerPoll() { TestSource source = new TestSource("testing", 3); SimpleChannel channel = new SimpleChannel(); - PollingSourceAdapter adapter = new PollingSourceAdapter(source); - adapter.setChannel(channel); - adapter.setInitialDelay(10000); + PollingSchedule schedule = new PollingSchedule(1000); + schedule.setInitialDelay(10000); + PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.setMaxMessagesPerTask(5); - adapter.start(); - adapter.processMessages(); + adapter.run(); Message message1 = channel.receive(0); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); 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 87d9349f94..3f1386e04d 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 @@ -19,7 +19,12 @@ - + + + + + + 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 0b90c1085e..f240a5bcb4 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 @@ -42,6 +42,7 @@ import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.PollableSource; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.integration.scheduling.Subscription; /** @@ -172,8 +173,7 @@ public class MessageBusTests { public void testErrorChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new MessageBus(); CountDownLatch latch = new CountDownLatch(1); - SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch)); - sourceAdapter.setChannel(new SimpleChannel()); + SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch), new SimpleChannel(), new PollingSchedule(1000)); bus.registerSourceAdapter("testAdapter", sourceAdapter); bus.start(); latch.await(1000, TimeUnit.MILLISECONDS);