INT-3993: Add Error Flow Logic to Object Map

JIRA: https://jira.spring.io/browse/INT-3993

Does not include "standard" `errorChannel` routing from pollers,
this is implied.

spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java

@@ -38,6 +38,7 @@ import org.springframework.util.ErrorHandler;
  * @author Mark Fisher
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  */
 public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryAware {
 
@@ -63,6 +64,15 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
 		this.defaultErrorChannel = defaultErrorChannel;
 	}
 
+	/**
+	 * Return the default error channel for this error handler.
+	 * @return the error channel.
+	 * @since 4.3
+	 */
+	public MessageChannel getDefaultErrorChannel() {
+		return this.defaultErrorChannel;
+	}
+
 	public void setSendTimeout(long sendTimeout) {
 		this.sendTimeout = sendTimeout;
 	}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java

@@ -36,6 +36,7 @@ import org.springframework.integration.transaction.IntegrationResourceHolderSync
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ErrorMessage;
@@ -60,6 +61,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 
 	private volatile ErrorHandler errorHandler;
 
+	private volatile boolean errorHandlerIsDefault;
+
 	private volatile Trigger trigger = new PeriodicTrigger(10);
 
 	private volatile List<Advice> adviceChain;
@@ -111,6 +114,21 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 		this.transactionSynchronizationFactory = transactionSynchronizationFactory;
 	}
 
+	/**
+	 * Return the default error channel if the error handler is explicitly provided and
+	 * it is a {@link MessagePublishingErrorHandler}.
+	 * @return the channel or null.
+	 * @since 4.3
+	 */
+	public MessageChannel getDefaultErrorChannel() {
+		if (!this.errorHandlerIsDefault && this.errorHandler instanceof MessagePublishingErrorHandler) {
+			return ((MessagePublishingErrorHandler) this.errorHandler).getDefaultErrorChannel();
+		}
+		else {
+			return null;
+		}
+	}
+
 	protected ClassLoader getBeanClassLoader() {
 		return this.beanClassLoader;
 	}
@@ -145,6 +163,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 						Assert.notNull(this.getBeanFactory(), "BeanFactory is required");
 						this.errorHandler = new MessagePublishingErrorHandler(
 								new BeanFactoryChannelResolver(getBeanFactory()));
+						this.errorHandlerIsDefault = true;
 					}
 					this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);
 				}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -47,6 +47,8 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 
 	private volatile MessageChannel errorChannel;
 
+	private volatile String errorChannelName;
+
 	private volatile boolean shouldTrack = false;
 
 	protected MessageProducerSupport() {
@@ -86,6 +88,36 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 		this.errorChannel = errorChannel;
 	}
 
+	/**
+	 * Set the error channel name. If no error channel is provided, this endpoint will
+	 * propagate Exceptions to the message-driven source. To completely suppress
+	 * Exceptions, provide a reference to the "nullChannel" here.
+	 * @param errorChannelName The error channel bean name.
+	 * @since 4.3
+	 */
+	public void setErrorChannelName(String errorChannelName) {
+		Assert.hasText(errorChannelName, "'errorChannelName' must not be empty");
+		this.errorChannelName = errorChannelName;
+	}
+
+	/**
+	 * Return the error channel (if provided) to which error messages will
+	 * be routed.
+	 * @return the channel or null.
+	 * @since 4.3
+	 */
+	public MessageChannel getErrorChannel() {
+		if (this.errorChannelName != null) {
+			synchronized (this) {
+				if (this.errorChannelName != null) {
+					this.errorChannel = getChannelResolver().resolveDestination(this.errorChannelName);
+					this.errorChannelName = null;
+				}
+			}
+		}
+		return this.errorChannel;
+	}
+
 	public void setSendTimeout(long sendTimeout) {
 		this.messagingTemplate.setSendTimeout(sendTimeout);
 	}
@@ -139,8 +171,9 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
 			this.messagingTemplate.send(getOutputChannel(), message);
 		}
 		catch (RuntimeException e) {
-			if (this.errorChannel != null) {
-				this.messagingTemplate.send(this.errorChannel, new ErrorMessage(e));
+			MessageChannel errorChannel = getErrorChannel();
+			if (errorChannel != null) {
+				this.messagingTemplate.send(errorChannel, new ErrorMessage(e));
 			}
 			else  {
 				throw e;

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -315,6 +315,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		}
 	}
 
+	/**
+	 * Return this gateway's request channel.
+	 * @return the channel.
+	 * @since 4.2
+	 */
 	public MessageChannel getRequestChannel() {
 		if (this.requestChannelName != null) {
 			synchronized (this) {
@@ -339,7 +344,13 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
 		return this.replyChannel;
 	}
 
-	protected MessageChannel getErrorChannel() {
+	/**
+	 * Return the error channel (if provided) to which error messages will
+	 * be routed.
+	 * @return the channel or null.
+	 * @since 4.3
+	 */
+	public MessageChannel getErrorChannel() {
 		if (this.errorChannelName != null) {
 			synchronized (this) {
 				if (this.errorChannelName != null) {

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java

@@ -27,10 +27,6 @@ public abstract class EndpointNode extends IntegrationNode {
 
 	private final String output;
 
-	protected EndpointNode(int nodeId, String name, Object nodeObject, Stats stats) {
-		this(nodeId, name, nodeObject, null, stats);
-	}
-
 	protected EndpointNode(int nodeId, String name, Object nodeObject, String output, Stats stats) {
 		super(nodeId, name, nodeObject, stats);
 		this.output = output;

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableEndpointNode.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016 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.support.management.graph;
+
+/**
+ * Represents nodes that can natively handle errors.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public class ErrorCapableEndpointNode extends EndpointNode implements ErrorCapableNode {
+
+	private final String errors;
+
+	protected ErrorCapableEndpointNode(int nodeId, String name, Object nodeObject, String output, String errors,
+			Stats stats) {
+		super(nodeId, name, nodeObject, output, stats);
+		this.errors = errors;
+	}
+
+	@Override
+	public String getErrors() {
+		return this.errors;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableMessageHandlerNode.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2016 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.support.management.graph;
+
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * Represents a message handler that can produce errors (pollable).
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public class ErrorCapableMessageHandlerNode extends MessageHandlerNode implements ErrorCapableNode {
+
+	private final String errors;
+
+	public ErrorCapableMessageHandlerNode(int nodeId, String name, MessageHandler handler, String input,
+			String output, String errors) {
+		super(nodeId, name, handler, input, output);
+		this.errors = errors;
+	}
+
+	@Override
+	public String getErrors() {
+		return this.errors;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableNode.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2016 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.support.management.graph;
+
+/**
+ * Nodes implementing this interface are capable of emitting errors.
+ *
+ * @author Gary Russell
+ * @since 4.3
+ *
+ */
+public interface ErrorCapableNode {
+
+	String getErrors();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java

@@ -30,6 +30,7 @@ import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.integration.endpoint.IntegrationConsumer;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.messaging.MessageChannel;
@@ -142,10 +143,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
 			SourcePollingChannelAdapter adapter = entry.getValue();
 			MessageSourceNode sourceNode = this.nodeFactory.sourceNode(entry.getKey(), adapter);
 			nodes.add(sourceNode);
-			MessageChannelNode channelNode = channelNodes.get(sourceNode.getOutput());
-			if (channelNode != null) {
-				links.add(new LinkNode(sourceNode.getNodeId(), channelNode.getNodeId()));
-			}
+			producerLink(links, channelNodes, sourceNode);
 		}
 	}
 
@@ -157,10 +155,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
 			MessagingGatewaySupport gateway = entry.getValue();
 			MessageGatewayNode gatewayNode = this.nodeFactory.gatewayNode(entry.getKey(), gateway);
 			nodes.add(gatewayNode);
-			MessageChannelNode channelInfo = channelNodes.get(gatewayNode.getOutput());
-			if (channelInfo != null) {
-				links.add(new LinkNode(gatewayNode.getNodeId(), channelInfo.getNodeId()));
-			}
+			producerLink(links, channelNodes, gatewayNode);
 		}
 	}
 
@@ -172,10 +167,7 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
 			MessageProducerSupport producer = entry.getValue();
 			MessageProducerNode producerNode = this.nodeFactory.producerNode(entry.getKey(), producer);
 			nodes.add(producerNode);
-			MessageChannelNode channelNode = channelNodes.get(producerNode.getOutput());
-			if (channelNode != null) {
-				links.add(new LinkNode(producerNode.getNodeId(), channelNode.getNodeId()));
-			}
+			producerLink(links, channelNodes, producerNode);
 		}
 	}
 
@@ -184,17 +176,31 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
 		Map<String, IntegrationConsumer> consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class);
 		for (Entry<String, IntegrationConsumer> entry : consumers.entrySet()) {
 			IntegrationConsumer consumer = entry.getValue();
-			MessageHandlerNode handlerNode = this.nodeFactory.handlerNode(entry.getKey(), consumer);
+			MessageHandlerNode handlerNode = consumer instanceof PollingConsumer
+					? this.nodeFactory.polledHandlerNode(entry.getKey(), (PollingConsumer) consumer)
+					: this.nodeFactory.handlerNode(entry.getKey(), consumer);
 			nodes.add(handlerNode);
 			MessageChannelNode channelNode = channelNodes.get(handlerNode.getInput());
 			if (channelNode != null) {
 				links.add(new LinkNode(channelNode.getNodeId(), handlerNode.getNodeId()));
 			}
-			if (handlerNode.getOutput() != null) {
-				channelNode = channelNodes.get(handlerNode.getOutput());
-				if (channelNode != null) {
-					links.add(new LinkNode(handlerNode.getNodeId(), channelNode.getNodeId()));
-				}
+			producerLink(links, channelNodes, handlerNode);
+		}
+	}
+
+	private void producerLink(Collection<LinkNode> links, Map<String, MessageChannelNode> channelNodes,
+			EndpointNode endpointNode) {
+		MessageChannelNode channelNode;
+		if (endpointNode.getOutput() != null) {
+			channelNode = channelNodes.get(endpointNode.getOutput());
+			if (channelNode != null) {
+				links.add(new LinkNode(endpointNode.getNodeId(), channelNode.getNodeId()));
+			}
+		}
+		if (endpointNode instanceof ErrorCapableNode) {
+			channelNode = channelNodes.get(((ErrorCapableNode) endpointNode).getErrors());
+			if (channelNode != null) {
+				links.add(new LinkNode(endpointNode.getNodeId(), channelNode.getNodeId()));
 			}
 		}
 	}
@@ -218,18 +224,22 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
 		}
 
 		private MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) {
+			String errorChannel = gateway.getErrorChannel() != null ? gateway.getErrorChannel().toString() : null;
 			return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway,
-					gateway.getRequestChannel().toString());
+					gateway.getRequestChannel().toString(), errorChannel);
 		}
 
 		private MessageProducerNode producerNode(String name, MessageProducerSupport producer) {
-			MessageChannel outputChannel = producer.getOutputChannel();
-			return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer, outputChannel.toString());
+			String errorChannel = producer.getErrorChannel() != null ? producer.getErrorChannel().toString() : null;
+			return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer,
+					producer.getOutputChannel().toString(), errorChannel);
 		}
 
 		private MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) {
+			String errorChannel = adapter.getDefaultErrorChannel() != null
+					? adapter.getDefaultErrorChannel().toString() : null;
 			return new MessageSourceNode(this.nodeId.incrementAndGet(), name, adapter.getMessageSource(),
-					adapter.getOutputChannel().toString());
+					adapter.getOutputChannel().toString(), errorChannel);
 		}
 
 		private MessageHandlerNode handlerNode(String name, IntegrationConsumer consumer) {
@@ -239,6 +249,15 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
 					consumer.getInputChannel().toString(), outputChannelName);
 		}
 
+		private MessageHandlerNode polledHandlerNode(String name, PollingConsumer consumer) {
+			MessageChannel outputChannel = consumer.getOutputChannel();
+			String outputChannelName = outputChannel == null ? null : outputChannel.toString();
+			String errorChannel = consumer.getDefaultErrorChannel() != null
+					? consumer.getDefaultErrorChannel().toString() : null;
+			return new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, consumer.getHandler(),
+					consumer.getInputChannel().toString(), outputChannelName, errorChannel);
+		}
+
 		private void reset() {
 			this.nodeId.set(0);
 		}

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java

@@ -25,10 +25,10 @@ import org.springframework.integration.gateway.MessagingGatewaySupport;
  * @since 4.3
  *
  */
-public class MessageGatewayNode extends EndpointNode {
+public class MessageGatewayNode extends ErrorCapableEndpointNode {
 
-	public MessageGatewayNode(int nodeId, String name, MessagingGatewaySupport gateway, String output) {
-		super(nodeId, name, gateway, output, new Stats(gateway));
+	public MessageGatewayNode(int nodeId, String name, MessagingGatewaySupport gateway, String output, String errors) {
+		super(nodeId, name, gateway, output, errors, new Stats(gateway));
 	}
 
 

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java

@@ -25,10 +25,10 @@ import org.springframework.integration.endpoint.MessageProducerSupport;
  * @since 4.3
  *
  */
-public class MessageProducerNode extends EndpointNode {
+public class MessageProducerNode extends ErrorCapableEndpointNode {
 
-	public MessageProducerNode(int nodeId, String name, MessageProducerSupport producer, String output) {
-		super(nodeId, name, producer, output, new IntegrationNode.Stats());
+	public MessageProducerNode(int nodeId, String name, MessageProducerSupport producer, String output, String errors) {
+		super(nodeId, name, producer, output, errors, new IntegrationNode.Stats());
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java

@@ -26,10 +26,10 @@ import org.springframework.integration.support.management.MessageSourceMetrics;
  * @since 4.3
  *
  */
-public class MessageSourceNode extends EndpointNode {
+public class MessageSourceNode extends ErrorCapableEndpointNode {
 
-	public MessageSourceNode(int nodeId, String name, MessageSource<?> messageSource, String output) {
-		super(nodeId, name, messageSource, output, messageSource instanceof MessageSourceMetrics
+	public MessageSourceNode(int nodeId, String name, MessageSource<?> messageSource, String output, String errors) {
+		super(nodeId, name, messageSource, output, errors, messageSource instanceof MessageSourceMetrics
 				? new Stats((MessageSourceMetrics) messageSource) : new IntegrationNode.Stats());
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java

@@ -33,6 +33,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableIntegrationManagement;
@@ -40,11 +41,13 @@ import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.endpoint.PollingConsumer;
+import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -74,11 +77,11 @@ public class IntegrationGraphServerTests {
 		@SuppressWarnings("unchecked")
 		List<Map<?, ?>> nodes = (List<Map<?, ?>>) map.get("nodes");
 		assertThat(nodes, is(notNullValue()));
-		assertThat(nodes.size(), is(equalTo(12)));
+		assertThat(nodes.size(), is(equalTo(13)));
 		@SuppressWarnings("unchecked")
 		List<Map<?, ?>> links = (List<Map<?, ?>>) map.get("links");
 		assertThat(links, is(notNullValue()));
-		assertThat(links.size(), is(equalTo(7)));
+		assertThat(links.size(), is(equalTo(9)));
 	}
 
 	@Configuration
@@ -104,6 +107,7 @@ public class IntegrationGraphServerTests {
 
 			};
 			producer.setOutputChannelName("one");
+			producer.setErrorChannelName("myErrors");
 			return producer;
 		}
 
@@ -124,6 +128,11 @@ public class IntegrationGraphServerTests {
 			return pollingConsumer;
 		}
 
+		@Bean
+		public PollableChannel two() {
+			return new QueueChannel();
+		}
+
 		@Bean
 		public SubscribableChannel three() {
 			return new DirectChannel();
@@ -134,6 +143,21 @@ public class IntegrationGraphServerTests {
 			return new QueueChannel();
 		}
 
+		@Bean
+		public PollableChannel myErrors() {
+			return new QueueChannel();
+		}
+
+		@Bean(name = PollerMetadata.DEFAULT_POLLER)
+		public PollerMetadata defaultPoller() {
+			PollerMetadata poller = new PollerMetadata();
+			poller.setTrigger(new PeriodicTrigger(60000));
+			MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
+			errorHandler.setDefaultErrorChannel(myErrors());
+			poller.setErrorHandler(errorHandler);
+			return poller;
+		}
+
 	}
 
 	public static class Services {