From 564eb165d1121ed0335ffa1cfa01d8b8276c9daf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sat, 23 Apr 2016 10:42:38 -0400 Subject: [PATCH] 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. --- .../MessagePublishingErrorHandler.java | 10 +++ .../endpoint/AbstractPollingEndpoint.java | 19 ++++++ .../endpoint/MessageProducerSupport.java | 37 ++++++++++- .../gateway/MessagingGatewaySupport.java | 13 +++- .../management/graph/EndpointNode.java | 4 -- .../graph/ErrorCapableEndpointNode.java | 41 ++++++++++++ .../graph/ErrorCapableMessageHandlerNode.java | 43 +++++++++++++ .../management/graph/ErrorCapableNode.java | 30 +++++++++ .../graph/IntegrationGraphServer.java | 63 ++++++++++++------- .../management/graph/MessageGatewayNode.java | 6 +- .../management/graph/MessageProducerNode.java | 6 +- .../management/graph/MessageSourceNode.java | 6 +- .../graph/IntegrationGraphServerTests.java | 28 ++++++++- 13 files changed, 266 insertions(+), 40 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableEndpointNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableMessageHandlerNode.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableNode.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java index 6b9f29adde..81f5ce6bc7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java +++ b/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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 70513204b1..2a3866a2da 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/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 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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index e7aee71eb2..74e393db9a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 021a826ae7..bdc2e8d212 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java index a02346546d..fc9dab4234 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/EndpointNode.java +++ b/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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableEndpointNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableEndpointNode.java new file mode 100644 index 0000000000..8bd4a9b965 --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableMessageHandlerNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableMessageHandlerNode.java new file mode 100644 index 0000000000..f179c2687f --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/ErrorCapableNode.java new file mode 100644 index 0000000000..977d947e56 --- /dev/null +++ b/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(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java index b224dffa23..45d007c2f5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/IntegrationGraphServer.java +++ b/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 consumers = this.applicationContext.getBeansOfType(IntegrationConsumer.class); for (Entry 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 links, Map 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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java index 3623b0fe41..4440bde846 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageGatewayNode.java +++ b/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)); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java index acf5db3890..b88a473265 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageProducerNode.java +++ b/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()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java index 9156bd7501..c2ab49b447 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/graph/MessageSourceNode.java +++ b/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()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java index 58bd486083..fd7071bd88 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java +++ b/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> nodes = (List>) map.get("nodes"); assertThat(nodes, is(notNullValue())); - assertThat(nodes.size(), is(equalTo(12))); + assertThat(nodes.size(), is(equalTo(13))); @SuppressWarnings("unchecked") List> links = (List>) 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 {