diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java new file mode 100644 index 0000000000..4114a54e7b --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -0,0 +1,139 @@ +/* + * 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.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.message.MessageSource; +import org.springframework.util.Assert; + +/** + * FactoryBean for creating a SourcePollingChannelAdapter instance. + * + * @author Mark Fisher + */ +public class SourcePollingChannelAdapterFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean { + + private volatile MessageSource source; + + private volatile MessageChannel outputChannel; + + private volatile PollerMetadata pollerMetadata; + + private volatile boolean autoStartup = true; + + private volatile String beanName; + + private volatile ConfigurableBeanFactory beanFactory; + + private volatile ClassLoader beanClassLoader; + + private volatile SourcePollingChannelAdapter adapter; + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + public void setSource(MessageSource source) { + this.source = source; + } + + public void setOutputChannel(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + } + + public void setPollerMetadata(PollerMetadata pollerMetadata) { + this.pollerMetadata = pollerMetadata; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setBeanFactory(BeanFactory beanFactory) { + Assert.isInstanceOf(ConfigurableBeanFactory.class, beanFactory, + "a ConfigurableBeanFactory is required"); + this.beanFactory = (ConfigurableBeanFactory) beanFactory; + } + + public void setBeanClassLoader(ClassLoader classLoader) { + this.beanClassLoader = classLoader; + } + + public void setBeanName(String beanName) { + this.beanName = beanName; + } + + public void afterPropertiesSet() throws Exception { + this.initializeAdapter(); + } + + public Object getObject() throws Exception { + if (this.adapter == null) { + this.initializeAdapter(); + } + return this.adapter; + } + + public Class getObjectType() { + return SourcePollingChannelAdapter.class; + } + + public boolean isSingleton() { + return true; + } + + private void initializeAdapter() { + synchronized (this.initializationMonitor) { + if (this.initialized) { + return; + } + Assert.notNull(this.source, "source is required"); + Assert.notNull(this.outputChannel, "outputChannel is required"); + SourcePollingChannelAdapter spca = new SourcePollingChannelAdapter(); + spca.setSource(this.source); + spca.setOutputChannel(this.outputChannel); + if (this.pollerMetadata == null) { + this.pollerMetadata = IntegrationContextUtils.getDefaultPollerMetadata(this.beanFactory); + Assert.notNull(this.pollerMetadata, "No poller has been defined for channel-adapter '" + + this.beanName + "', and no default poller is available within the context."); + } + spca.setTrigger(this.pollerMetadata.getTrigger()); + spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll()); + spca.setTaskExecutor(this.pollerMetadata.getTaskExecutor()); + spca.setTransactionManager(this.pollerMetadata.getTransactionManager()); + spca.setTransactionDefinition(this.pollerMetadata.getTransactionDefinition()); + spca.setAutoStartup(this.autoStartup); + spca.setBeanName(this.beanName); + spca.setBeanFactory(this.beanFactory); + spca.setBeanClassLoader(this.beanClassLoader); + spca.afterPropertiesSet(); + this.adapter = spca; + this.initialized = true; + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java index 1f8c38290a..dea2cf4d8f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java @@ -21,9 +21,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.config.PollerMetadata; -import org.springframework.integration.endpoint.SourcePollingChannelAdapter; -import org.springframework.integration.scheduling.IntervalTrigger; +import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.util.Assert; import org.springframework.util.xml.DomUtils; @@ -34,35 +32,21 @@ import org.springframework.util.xml.DomUtils; */ public abstract class AbstractPollingInboundChannelAdapterParser extends AbstractChannelAdapterParser { - private volatile PollerMetadata defaultPollerMetadata; - - @Override protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { String source = this.parseSource(element, parserContext); Assert.hasText(source, "failed to parse source"); - BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SourcePollingChannelAdapter.class); + BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(SourcePollingChannelAdapterFactoryBean.class); adapterBuilder.addPropertyReference("source", source); adapterBuilder.addPropertyReference("outputChannel", channelName); Element pollerElement = DomUtils.getChildElementByTagName(element, "poller"); if (pollerElement != null) { IntegrationNamespaceUtils.configurePollerMetadata(pollerElement, adapterBuilder, parserContext); } - else { - adapterBuilder.addPropertyValue("pollerMetadata", this.getDefaultPollerMetadata()); - } IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "auto-startup"); return adapterBuilder.getBeanDefinition(); } - private synchronized PollerMetadata getDefaultPollerMetadata() { - if (this.defaultPollerMetadata == null) { - this.defaultPollerMetadata = new PollerMetadata(); - this.defaultPollerMetadata.setTrigger(new IntervalTrigger(1000)); - } - return this.defaultPollerMetadata; - } - /** * Subclasses must implement this method to parse the PollableSource instance * which the created Channel Adapter will poll. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index a2c0a92c1a..a668ed7b7a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -17,7 +17,6 @@ package org.springframework.integration.endpoint; import org.springframework.integration.channel.MessageChannelTemplate; -import org.springframework.integration.config.PollerMetadata; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageSource; @@ -60,14 +59,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint { this.channelTemplate.setSendTimeout(sendTimeout); } - public void setPollerMetadata(PollerMetadata pollerMetadata) { - this.setTrigger(pollerMetadata.getTrigger()); - this.setMaxMessagesPerPoll(pollerMetadata.getMaxMessagesPerPoll()); - this.setTaskExecutor(pollerMetadata.getTaskExecutor()); - this.setTransactionDefinition(pollerMetadata.getTransactionDefinition()); - this.setTransactionManager(pollerMetadata.getTransactionManager()); - } - @Override protected void onInit() { Assert.notNull(this.source, "source must not be null"); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java index e1a47776eb..b4fda1ff9d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java @@ -94,15 +94,17 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests @Test public void methodInvokingSource() { String beanName = "methodInvokingSource"; - PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel"); + PollableChannel channel = (PollableChannel) this.applicationContext.getBean("queueChannel"); TestBean testBean = (TestBean) this.applicationContext.getBean("testBean"); testBean.store("source test"); Object adapter = this.applicationContext.getBean(beanName); assertNotNull(adapter); assertTrue(adapter instanceof SourcePollingChannelAdapter); + ((SourcePollingChannelAdapter) adapter).start(); Message message = channel.receive(1000); assertNotNull(message); assertEquals("source test", testBean.getMessage()); + ((SourcePollingChannelAdapter) adapter).stop(); } @Test(expected = ChannelResolutionException.class)