INT-1708, INT-1709 added error-channel to the poller

spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java

@@ -130,7 +130,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 				this.pollerMetadata.setMaxMessagesPerPoll(1);
 			}
 			spca.setPollerMetadata(this.pollerMetadata);
-			
+			spca.setErrorHandler(this.pollerMetadata.getErrorHandler());
 			spca.setBeanClassLoader(this.beanClassLoader);
 			spca.setAutoStartup(this.autoStartup);
 			spca.setBeanName(this.beanName);

spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java

@@ -99,6 +99,14 @@ public class PollerParser extends AbstractBeanDefinitionParser {
 		configureAdviceChain(adviceChainElement, txElement, metadataBuilder, parserContext);
 		
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "task-executor");
+		String errorChannel = element.getAttribute("error-channel");
+		if (StringUtils.hasText(errorChannel)){
+			BeanDefinitionBuilder errorHandler = BeanDefinitionBuilder.genericBeanDefinition(
+					"org.springframework.integration.channel.MessagePublishingErrorHandler");
+			errorHandler.addPropertyReference("defaultErrorChannel", errorChannel);
+			
+			metadataBuilder.addPropertyValue("errorHandler", errorHandler.getBeanDefinition());
+		}
 		return metadataBuilder.getBeanDefinition();
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java

@@ -22,6 +22,7 @@ import java.util.concurrent.Executor;
 import org.aopalliance.aop.Advice;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.util.ErrorHandler;
 
 /**
  * @author Mark Fisher
@@ -36,6 +37,8 @@ public class PollerMetadata {
 	private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
 
 	private volatile long receiveTimeout = 1000;
+	
+	private volatile ErrorHandler errorHandler;
 
 	private List<Advice> adviceChain;
 
@@ -48,6 +51,14 @@ public class PollerMetadata {
 	public Trigger getTrigger() {
 		return this.trigger;
 	}
+	
+	public ErrorHandler getErrorHandler() {
+		return errorHandler;
+	}
+
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
 
 	/**
 	 * Set the maximum number of messages to receive for each poll.

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd

@@ -1242,6 +1242,15 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="error-channel" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+					Identifies channel that error messages will be sent to if a failure occurs in this
+					poller's invocation. To completely suppress Exceptions, provide a
+					reference to the "nullChannel" here.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 	</xsd:complexType>
 
 	<xsd:complexType name="intervalTriggerType">

spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannel-context.xml

@@ -9,22 +9,30 @@
 
 	<int:inbound-channel-adapter id="withErrorHeader" expression="5/0" channel="serviceChannel" auto-startup="false">
 		<int:poller fixed-rate="1000"/>
-		<int:header name="errorChannel" value="eChannel"/>
+		<int:header name="eChannel" value="eChannel"/>
 	</int:inbound-channel-adapter>
 	
 	<int:inbound-channel-adapter id="withErrorChannel" expression="5/0" 
-					channel="serviceChannel" auto-startup="false" error-channel="eChannel">
-		<int:poller fixed-rate="1000"/>
+					channel="serviceChannel" auto-startup="false">
+		<int:poller fixed-rate="1000" error-channel="eChannel"/>
 	</int:inbound-channel-adapter>
 	
 	<int:inbound-channel-adapter id="withErrorChannelAndHeader" expression="5/0" 
-					channel="serviceChannel" auto-startup="false" error-channel="eChannel">
-		<int:poller fixed-rate="1000"/>
+					channel="serviceChannel" auto-startup="false">
+		<int:poller fixed-rate="1000" error-channel="eChannel"/>
+		<int:header name="errorChannel" value="errChannel"/>
+	</int:inbound-channel-adapter>
+	
+	<int:inbound-channel-adapter id="withErrorChannelAndHeaderErrorOnSend" expression="5" 
+					channel="serviceChannel" auto-startup="false">
+		<int:poller fixed-rate="1000" error-channel="eChannel"/>
 		<int:header name="errorChannel" value="errChannel"/>
 	</int:inbound-channel-adapter>
 	
 	<int:channel id="serviceChannel"/>
 	
+	<int:service-activator input-channel="serviceChannel" expression="payload/0"/>
+	
 	<int:channel id="eChannel">
 		<int:queue/>
 	</int:channel>

spring-integration-core/src/test/java/org/springframework/integration/config/xml/PollerWithErrorChannelTests.java

@@ -16,34 +16,49 @@
 package org.springframework.integration.config.xml;
 
 import static junit.framework.Assert.assertNotNull;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import org.junit.Ignore;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.Message;
+import org.springframework.integration.core.MessageHandler;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.core.SubscribableChannel;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 
 /**
  * @author Oleg Zhurakousky
  *
  */
-public class PollerWithErrorChannel {
+public class PollerWithErrorChannelTests {
 
 	@Test
-	@Ignore
+	/*
+	 * Although adapter configuration specifies header-enricher pointing to the 'eChannel' as errorChannel
+	 * the ErrorMessage will still be forwarded to the 'errorChannel' since exception occurs on 
+	 * receive() and not on send()
+	 */
 	public void testWithErrorChannelAsHeader() throws Exception{
 		ApplicationContext ac = new ClassPathXmlApplicationContext("PollerWithErrorChannel-context.xml", this.getClass());
 		SourcePollingChannelAdapter adapter = ac.getBean("withErrorHeader", SourcePollingChannelAdapter.class);
+		
+		SubscribableChannel errorChannel = ac.getBean("errorChannel", SubscribableChannel.class);
+		MessageHandler handler = mock(MessageHandler.class);
+		errorChannel.subscribe(handler);
 		adapter.start();
-		PollableChannel errorChannel = ac.getBean("eChannel", PollableChannel.class);
-		assertNotNull(errorChannel.receive(1000));
+		Thread.sleep(1000);
+		verify(handler, atLeastOnce()).handleMessage(Mockito.any(Message.class));
 		adapter.stop();
 	}
 	
 	@Test
-	@Ignore
 	public void testWithErrorChannel() throws Exception{
 		ApplicationContext ac = new ClassPathXmlApplicationContext("PollerWithErrorChannel-context.xml", this.getClass());
 		SourcePollingChannelAdapter adapter = ac.getBean("withErrorChannel", SourcePollingChannelAdapter.class);
@@ -54,11 +69,21 @@ public class PollerWithErrorChannel {
 	}
 	
 	@Test
-	@Ignore
 	public void testWithErrorChannelAndHeader() throws Exception{
 		ApplicationContext ac = new ClassPathXmlApplicationContext("PollerWithErrorChannel-context.xml", this.getClass());
 		SourcePollingChannelAdapter adapter = ac.getBean("withErrorChannelAndHeader", SourcePollingChannelAdapter.class);
 		adapter.start();
+		PollableChannel errorChannel = ac.getBean("eChannel", PollableChannel.class);
+		assertNotNull(errorChannel.receive(1000));
+		adapter.stop();
+	}
+	
+	@Test 
+	// config the same as above but the error wil come from the send
+	public void testWithErrorChannelAndHeaderWithSendFailure() throws Exception{
+		ApplicationContext ac = new ClassPathXmlApplicationContext("PollerWithErrorChannel-context.xml", this.getClass());
+		SourcePollingChannelAdapter adapter = ac.getBean("withErrorChannelAndHeaderErrorOnSend", SourcePollingChannelAdapter.class);
+		adapter.start();
 		PollableChannel errorChannel = ac.getBean("errChannel", PollableChannel.class);
 		assertNotNull(errorChannel.receive(1000));
 		adapter.stop();