diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java index 98950158ad..ab4600d87b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java @@ -85,4 +85,11 @@ public @interface Poller { */ String cron() default ""; + /** + * @return The the bean name of default error channel + * for the underlying {@code MessagePublishingErrorHandler}. + * @since 4.3.3 + */ + String errorChannel() default ""; + } 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 81f5ce6bc7..9ba55c2cfa 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 @@ -39,6 +39,7 @@ import org.springframework.util.ErrorHandler; * @author Iwein Fuld * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryAware { @@ -48,6 +49,8 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA private volatile MessageChannel defaultErrorChannel; + private volatile String defaultErrorChannelName; + private volatile long sendTimeout = 1000; @@ -70,9 +73,25 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA * @since 4.3 */ public MessageChannel getDefaultErrorChannel() { + String defaultErrorChannelName = this.defaultErrorChannelName; + if (defaultErrorChannelName != null) { + if (this.channelResolver != null) { + this.defaultErrorChannel = this.channelResolver.resolveDestination(defaultErrorChannelName); + this.defaultErrorChannelName = null; + } + } return this.defaultErrorChannel; } + /** + * Specify the bean name of default error channel for this error handler. + * @param defaultErrorChannelName the bean name of the error channel + * @since 4.3.3 + */ + public void setDefaultErrorChannelName(String defaultErrorChannelName) { + this.defaultErrorChannelName = defaultErrorChannelName; + } + public void setSendTimeout(long sendTimeout) { this.sendTimeout = sendTimeout; } @@ -87,7 +106,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA @Override public final void handleError(Throwable t) { - MessageChannel errorChannel = this.resolveErrorChannel(t); + MessageChannel errorChannel = resolveErrorChannel(t); boolean sent = false; if (errorChannel != null) { try { @@ -123,7 +142,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA private MessageChannel resolveErrorChannel(Throwable t) { Message failedMessage = (t instanceof MessagingException) ? ((MessagingException) t).getFailedMessage() : null; - if (this.defaultErrorChannel == null && this.channelResolver != null) { + if (getDefaultErrorChannel() == null && this.channelResolver != null) { this.defaultErrorChannel = this.channelResolver.resolveDestination( IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); } @@ -137,7 +156,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA } Assert.isInstanceOf(String.class, errorChannelHeader, "Unsupported error channel header type. Expected MessageChannel or String, but actual type is [" + - errorChannelHeader.getClass() + "]"); + errorChannelHeader.getClass() + "]"); return this.channelResolver.resolveDestination((String) errorChannelHeader); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index dfc7496cae..568be2fa07 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -47,6 +47,7 @@ import org.springframework.core.task.TaskExecutor; import org.springframework.integration.annotation.IdempotentReceiver; import org.springframework.integration.annotation.Poller; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessagePublishingErrorHandler; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.context.Orderable; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -230,8 +231,8 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations, AbstractReplyProducingMessageHandler handler) { - String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel", String.class); + protected void setOutputChannelIfPresent(List annotations, + AbstractReplyProducingMessageHandler handler) { + String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel", + String.class); if (StringUtils.hasText(outputChannelName)) { handler.setOutputChannelName(outputChannelName); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java index 60adf2bdb9..6d6b572fe4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java @@ -24,6 +24,7 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import java.util.ArrayList; import java.util.List; @@ -45,6 +46,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; import org.springframework.expression.ExpressionParser; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.MessageRejectedException; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy; import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy; @@ -75,6 +77,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -119,6 +122,9 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Qualifier("skippedMessageSource") private MessageSource skippedMessageSource; + @Autowired + private PollableChannel counterErrorChannel; + @Test public void testMessagingAnnotationsFlow() { this.sourcePollingChannelAdapter.start(); @@ -126,6 +132,17 @@ public class MessagingAnnotationsWithBeanAnnotationTests { Message receive = this.discardChannel.receive(10000); assertNotNull(receive); assertTrue(((Integer) receive.getPayload()) % 2 == 0); + + receive = this.counterErrorChannel.receive(10000); + assertNotNull(receive); + assertThat(receive, instanceOf(ErrorMessage.class)); + assertThat(receive.getPayload(), instanceOf(MessageRejectedException.class)); + MessageRejectedException exception = (MessageRejectedException) receive.getPayload(); + assertThat(exception.getMessage(), + containsString("MessageFilter " + + "'messagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.filter.filter.handler'" + + " rejected Message")); + } for (Message message : collector) { assertFalse(((Integer) message.getPayload()) % 2 == 0); @@ -174,24 +191,30 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Bean @InboundChannelAdapter(value = "routerChannel", autoStartup = "false", - poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1")) + poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1", errorChannel = "counterErrorChannel")) public MessageSource counterMessageSource(final AtomicInteger counter) { return new MessageSource() { @Override public Message receive() { - return new GenericMessage(counter.incrementAndGet()); + return new GenericMessage<>(counter.incrementAndGet()); } + }; } + @Bean + public PollableChannel counterErrorChannel() { + return new QueueChannel(); + } + @Bean public MessageChannel routerChannel() { return new DirectChannel(); } @Bean - @Router(inputChannel = "routerChannel", channelMappings = {"true=odd", "false=filter"}, suffix = "Channel") + @Router(inputChannel = "routerChannel", channelMappings = { "true=odd", "false=filter" }, suffix = "Channel") public MessageSelector router() { return new ExpressionEvaluatingSelector("payload % 2 == 0"); } @@ -208,7 +231,10 @@ public class MessagingAnnotationsWithBeanAnnotationTests { } @Bean - @Filter(inputChannel = "filterChannel", outputChannel = "aggregatorChannel", discardChannel = "discardChannel") + @Filter(inputChannel = "filterChannel", + outputChannel = "aggregatorChannel", + discardChannel = "discardChannel", + throwExceptionOnRejection = "true") public MessageSelector filter() { return new ExpressionEvaluatingSelector("payload % 2 != 0"); } @@ -277,7 +303,7 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Filter(inputChannel = "skippedChannel5") @Profile("foo") public MessageHandler skippedMessageHandler() { - return m -> { }; + return mock(MessageHandler.class); } @Bean @@ -298,7 +324,14 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @InboundChannelAdapter("serviceChannel") @Profile("foo") public MessageSource skippedMessageSource() { - return () -> new GenericMessage<>("foo"); + return new MessageSource() { + + @Override + public GenericMessage receive() { + return new GenericMessage<>("foo"); + } + + }; } }