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 f8c40a911b..587b45cd1c 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 @@ -49,6 +49,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; @@ -236,8 +237,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..376d1efbe5 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 @@ -45,6 +45,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 +76,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 +121,9 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Qualifier("skippedMessageSource") private MessageSource skippedMessageSource; + @Autowired + private PollableChannel counterErrorChannel; + @Test public void testMessagingAnnotationsFlow() { this.sourcePollingChannelAdapter.start(); @@ -126,6 +131,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,15 +190,14 @@ 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() { + return () -> new GenericMessage<>(counter.incrementAndGet()); + } - @Override - public Message receive() { - return new GenericMessage(counter.incrementAndGet()); - } - }; + @Bean + public PollableChannel counterErrorChannel() { + return new QueueChannel(); } @Bean @@ -191,7 +206,7 @@ public class MessagingAnnotationsWithBeanAnnotationTests { } @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 +223,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 +295,8 @@ public class MessagingAnnotationsWithBeanAnnotationTests { @Filter(inputChannel = "skippedChannel5") @Profile("foo") public MessageHandler skippedMessageHandler() { - return m -> { }; + return m -> { + }; } @Bean diff --git a/src/reference/asciidoc/configuration.adoc b/src/reference/asciidoc/configuration.adoc index bd5ac04a5a..5371ddc363 100644 --- a/src/reference/asciidoc/configuration.adoc +++ b/src/reference/asciidoc/configuration.adoc @@ -374,7 +374,7 @@ public class AnnotationService { This annotation provides only simple `PollerMetadata` options. The `@Poller`'s attributes `maxMessagesPerPoll`, `fixedDelay`, `fixedRate` and `cron` can be configured with _property-placeholders_. -If it is necessary to provide more polling options (e.g. transaction, advice-chain, error-handler), the `PollerMetadata` should be configured as a generic bean with its bean name used for `@Poller`'s `value` attribute. +If it is necessary to provide more polling options (e.g. transaction, advice-chain, error-handler etc.), the `PollerMetadata` should be configured as a generic bean with its bean name used for `@Poller`'s `value` attribute. In this case, no other attributes are allowed (they would be specified on the `PollerMetadata` bean). Note, if `inputChannel` is `PollableChannel` and no `@Poller` is configured, the default `PollerMetadata` will be used, if it is present in the application context. To declare the default poller using `@Configuration`, use: @@ -424,6 +424,10 @@ public class AnnotationService { } ---- +Starting with _version 4.3.3_, the `@Poller` annotation now has the `errorChannel` attribute for easier configuration of the underlying `MessagePublishingErrorHandler`. +This attribute play the same role as `error-channel` in the `` xml component. +See <> for more information. + *@InboundChannelAdapter* Starting with _version 4.0_, the `@InboundChannelAdapter` method annotation is available. diff --git a/src/reference/asciidoc/endpoint.adoc b/src/reference/asciidoc/endpoint.adoc index 43dfd83b4f..4eef9ed6d7 100644 --- a/src/reference/asciidoc/endpoint.adoc +++ b/src/reference/asciidoc/endpoint.adoc @@ -152,14 +152,14 @@ For example, an _Aggregator_ waits for a number of Messages to arrive and is oft When using the namespace configuration, you do not strictly need to know all of the details, but it still might be worth knowing that several of these components share a common base class, the `AbstractReplyProducingMessageHandler`, and it provides a `setOutputChannel(..)` method. [[endpoint-namespace]] -==== Namespace Support +==== Endpoint Namespace Support Throughout the reference manual, you will see specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on. Most of these will support an _input-channel_ attribute and many will support an _output-channel_ attribute. After being parsed, these endpoint elements produce an instance of either the `PollingConsumer` or the `EventDrivenConsumer` depending on the type of the _input-channel_ that is referenced: `PollableChannel` or `SubscribableChannel` respectively. When the channel is pollable, then the polling behavior is determined based on the endpoint element's _poller_ sub-element and its attributes. -_Configuration_Below you find a _poller_ with all available configuration options: +In the configuration below you find a _poller_ with all available configuration options: [source,xml] ---- diff --git a/src/reference/asciidoc/mail.adoc b/src/reference/asciidoc/mail.adoc index f1b5c0a196..b5bf42bb49 100644 --- a/src/reference/asciidoc/mail.adoc +++ b/src/reference/asciidoc/mail.adoc @@ -188,7 +188,7 @@ Alternatively, provide the host, username, and password: ---- NOTE: Keep in mind, as with any outbound Channel Adapter, if the referenced channel is a `PollableChannel`, -a `` sub-element should be provided (see ). +a `` sub-element should be provided (see <>). When using the namespace support, a _header-enricher_ Message Transformer is also available. This simplifies the application of the headers mentioned above to any Message prior to sending to the Mail Outbound Channel Adapter. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c1d044c36c..70c601c2ea 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -15,6 +15,9 @@ development process. ==== Core Changes +The `@Poller` annotation now has the `errorChannel` attribute for easier configuration of the underlying `MessagePublishingErrorHandler`. +See <> for more information. + ==== JMS Changes Previously, Spring Integration JMS XML configuration used a default bean name `connectionFactory` for the JMS Connection Factory, allowing the property to be omitted from component definitions.