Added SourcePollingChannelAdapterFactoryBean. The 'poller' element is now required for inbound-channel-adapter elements unless a default poller has been configured.

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;
+		}
+	}
+
+}

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.

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");

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)