INT-1624 Add error-channel to event inbound channel adapter XSD, Parser, Test

spring-integration-event/.springBeans

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beansProjectDescription>
+	<version>1</version>
+	<pluginVersion><![CDATA[2.5.1.201011101000-RELEASE]]></pluginVersion>
+	<configSuffixes>
+		<configSuffix><![CDATA[xml]]></configSuffix>
+	</configSuffixes>
+	<enableImports><![CDATA[false]]></enableImports>
+	<configs>
+	</configs>
+	<configSets>
+	</configSets>
+</beansProjectDescription>

spring-integration-event/src/main/java/org/springframework/integration/event/config/EventInboundChannelAdapterParser.java

@@ -36,6 +36,7 @@ public class EventInboundChannelAdapterParser extends AbstractChannelAdapterPars
 		BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.rootBeanDefinition(
 				"org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(adapterBuilder, element, "channel", "outputChannel");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(adapterBuilder, element, "error-channel", "errorChannel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "event-types");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "payload-expression");
 		return adapterBuilder.getBeanDefinition();

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

@@ -38,6 +38,21 @@
             		</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="error-channel" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type
+								type="org.springframework.integration.core.MessageChannel" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						If a (synchronous) downstream exception is thrown and an error-channel is specified,
+						the MessageHandlingException will be sent to this channel. Otherwise, any such exception
+						will be propagated to the calling thread.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="event-types" type="xsd:string" use="optional">
 				<xsd:annotation>
 					<xsd:documentation>

spring-integration-event/src/test/java/org/springframework/integration/event/config/EventInboundChannelAdapterParserTests-context.xml

@@ -11,7 +11,8 @@
 
 	<int:message-history/>
 
-	<int-event:inbound-channel-adapter id="eventAdapterSimple" channel="input"/>
+	<int-event:inbound-channel-adapter id="eventAdapterSimple" channel="input"
+	                                   error-channel="errorChannel"/>
 
 	<int:channel id="input">
 		<int:queue/>

spring-integration-event/src/test/java/org/springframework/integration/event/config/EventInboundChannelAdapterParserTests.java

@@ -18,6 +18,7 @@ package org.springframework.integration.event.config;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Properties;
@@ -35,6 +36,7 @@ import org.springframework.context.ApplicationEvent;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.expression.Expression;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
 import org.springframework.integration.history.MessageHistory;
@@ -54,6 +56,8 @@ public class EventInboundChannelAdapterParserTests {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	MessageChannel errorChannel;
 
 	@Test
 	public void validateEventParser() {
@@ -62,6 +66,7 @@ public class EventInboundChannelAdapterParserTests {
 		Assert.assertTrue(adapter instanceof ApplicationEventListeningMessageProducer);
 		DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
 		Assert.assertEquals(context.getBean("input"), adapterAccessor.getPropertyValue("outputChannel"));
+		Assert.assertSame(errorChannel, adapterAccessor.getPropertyValue("errorChannel"));
 	}
 
 	@Test
@@ -77,6 +82,7 @@ public class EventInboundChannelAdapterParserTests {
 		assertTrue(eventTypes.size() == 2);	
 		assertTrue(eventTypes.contains(SampleEvent.class));
 		assertTrue(eventTypes.contains(AnotherSampleEvent.class));
+		assertNull(adapterAccessor.getPropertyValue("errorChannel"));
 	}
 
 	@Test

spring-integration-event/src/test/java/org/springframework/integration/event/inbound/ApplicationEventListeningMessageProducerTests.java

@@ -19,9 +19,9 @@ package org.springframework.integration.event.inbound;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
 
 import org.junit.Test;
-
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.context.event.ContextRefreshedEvent;
@@ -29,10 +29,14 @@ import org.springframework.context.event.ContextStartedEvent;
 import org.springframework.context.event.ContextStoppedEvent;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.Message;
+import org.springframework.integration.MessageHandlingException;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.event.core.MessagingEvent;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.test.annotation.ExpectedException;
 
 /**
  * @author Mark Fisher
@@ -145,6 +149,30 @@ public class ApplicationEventListeningMessageProducerTests {
 		assertEquals("test", message2.getPayload());
 	}
 
+	@Test @ExpectedException(value=MessageHandlingException.class)
+	public void anyApplicationEventCausesExceptionWithErrorHandling() {
+		DirectChannel channel = new DirectChannel();
+		channel.subscribe(new AbstractReplyProducingMessageHandler() {
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				throw new RuntimeException("Failed");
+			}
+		});
+		ApplicationEventListeningMessageProducer adapter = new ApplicationEventListeningMessageProducer();
+		adapter.setOutputChannel(channel);
+		QueueChannel errorChannel = new QueueChannel();
+		adapter.setErrorChannel(errorChannel);
+		adapter.start();
+		adapter.onApplicationEvent(new TestApplicationEvent1());
+		Message<?> message = errorChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("Failed", ((Exception) message.getPayload()).getCause().getMessage());
+		adapter.setErrorChannel(null);
+		try {
+			adapter.onApplicationEvent(new TestApplicationEvent1());
+			fail("Expected MessageHandlingException");
+		} catch (MessageHandlingException e) { }
+	}
+
 
 	@SuppressWarnings("serial")
 	private static class TestApplicationEvent1 extends ApplicationEvent {