Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4113: Add @Poller.errorChannel() Attribute

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4113

* To simplify `MessagePublishingErrorHandler` and avoid extra `PollerMetadata` beans, added the `errorChannel()` attribute to the `@Poller` annotation
* The `MessagePublishingErrorHandler` now supports the late-binding via new `defaultErrorChannelName`
* Docs about new `errorChannel()` attribute
* Some other docs polishing

Polishing

**Cherry-pick to 4.3.x**

Polishing docs according PR comments

Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

Resolved and reworked Java8 features in tests.
This commit is contained in:
Artem Bilan
2016-09-27 14:52:05 -04:00
committed by Gary Russell
parent 166a732447
commit 5187cbf1a8
4 changed files with 87 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java
View File

@@ -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 "";
}

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

@@ -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);
}

24
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -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<T extends Annotation
boolean createEndpoint = StringUtils.hasText(inputChannel);
if (!createEndpoint && beanAnnotationAware()) {
boolean isBean = AnnotatedElementUtils.isAnnotated(method, Bean.class.getName());
Assert.isTrue(!isBean, "A channel name in '" + getInputChannelAttribute() + "' is required when " + this.annotationType +
" is used on '@Bean' methods.");
Assert.isTrue(!isBean, "A channel name in '" + getInputChannelAttribute() + "' is required when " +
this.annotationType + " is used on '@Bean' methods.");
}
return createEndpoint;
}
@@ -320,7 +321,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
PollerMetadata pollerMetadata = null;
Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class);
if (!ObjectUtils.isEmpty(pollers)) {
Assert.state(pollers.length == 1, "The 'poller' for an Annotation-based endpoint can have only one '@Poller'.");
Assert.state(pollers.length == 1,
"The 'poller' for an Annotation-based endpoint can have only one '@Poller'.");
Poller poller = pollers[0];
String ref = poller.value();
@@ -330,6 +332,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
String fixedRateValue = this.beanFactory.resolveEmbeddedValue(poller.fixedRate());
String maxMessagesPerPollValue = this.beanFactory.resolveEmbeddedValue(poller.maxMessagesPerPoll());
String cron = this.beanFactory.resolveEmbeddedValue(poller.cron());
String errorChannel = this.beanFactory.resolveEmbeddedValue(poller.errorChannel());
if (StringUtils.hasText(ref)) {
Assert.state(!StringUtils.hasText(triggerRef) && !StringUtils.hasText(executorRef) &&
@@ -347,9 +350,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
// SPCAs default to 1 message per poll
pollerMetadata.setMaxMessagesPerPoll(1);
}
if (StringUtils.hasText(executorRef)) {
pollerMetadata.setTaskExecutor(this.beanFactory.getBean(executorRef, TaskExecutor.class));
}
Trigger trigger = null;
if (StringUtils.hasText(triggerRef)) {
Assert.state(!StringUtils.hasText(cron) && !StringUtils.hasText(fixedDelayValue)
@@ -373,6 +378,13 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
//'Trigger' can be null. 'PollingConsumer' does fallback to the 'new PeriodicTrigger(10)'.
pollerMetadata.setTrigger(trigger);
if (StringUtils.hasText(errorChannel)) {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setDefaultErrorChannelName(errorChannel);
errorHandler.setBeanFactory(this.beanFactory);
pollerMetadata.setErrorHandler(errorHandler);
}
}
}
else {
@@ -402,8 +414,10 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
return name + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX;
}
protected void setOutputChannelIfPresent(List<Annotation> annotations, AbstractReplyProducingMessageHandler handler) {
String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel", String.class);
protected void setOutputChannelIfPresent(List<Annotation> annotations,
AbstractReplyProducingMessageHandler handler) {
String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel",
String.class);
if (StringUtils.hasText(outputChannelName)) {
handler.setOutputChannelName(outputChannelName);
}

45
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java
View File

@@ -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<Integer> counterMessageSource(final AtomicInteger counter) {
return new MessageSource<Integer>() {
@Override
public Message<Integer> receive() {
return new GenericMessage<Integer>(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<String>() {
@Override
public GenericMessage<String> receive() {
return new GenericMessage<>("foo");
}
};
}
}