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

INT-4471: PubSubChannel: Add errorHandler warn (#2459)

Browse Source 
* INT-4471: PubSubChannel: Add errorHandler warn

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

* When an `Executor` is not provided, log warn that the provided
`ErrorHandler` is ignored.

**Cherry-pick to 5.0.x and 4.3.x**

* * Polish warn message
This commit is contained in:
Artem Bilan
2018-05-31 15:25:38 -04:00
committed by Gary Russell
parent 9a88140fb3
commit 17e794d2c8
4 changed files with 45 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -35,19 +35,18 @@ import org.springframework.util.ErrorHandler;
*/
public class PublishSubscribeChannel extends AbstractExecutorChannel {
private volatile ErrorHandler errorHandler;
private ErrorHandler errorHandler;
private volatile boolean ignoreFailures;
private boolean ignoreFailures;
private volatile boolean applySequence;
private boolean applySequence;
private volatile int minSubscribers;
private int minSubscribers;
/**
* Create a PublishSubscribeChannel that will use an {@link Executor}
* to invoke the handlers. If this is null, each invocation will occur in
* the message sender's thread.
*
* @param executor The executor.
*/
public PublishSubscribeChannel(Executor executor) {
@@ -79,9 +78,7 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel {
* a {@link MessagePublishingErrorHandler} that sends error messages to
* the failed request Message's error channel header if available or to
* the default 'errorChannel' otherwise.
*
* @param errorHandler The error handler.
*
* @see #PublishSubscribeChannel(Executor)
*/
public void setErrorHandler(ErrorHandler errorHandler) {
@@ -149,6 +146,14 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel {
getDispatcher().setApplySequence(this.applySequence);
getDispatcher().setMinSubscribers(this.minSubscribers);
}
else if (this.errorHandler != null) {
if (this.logger.isWarnEnabled()) {
this.logger.warn("The 'errorHandler' is ignored for the '" + getComponentName() +
"' (an 'executor' is not provided) and exceptions will be thrown " +
"directly within the sending Thread");
}
}
if (this.maxSubscribers == null) {
Integer maxSubscribers =
getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);

2
spring-integration-core/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml → spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests-context.xml
View File

@@ -19,7 +19,7 @@
<publish-subscribe-channel id="channelWithApplySequenceEnabledAndTaskExecutor" apply-sequence="true" task-executor="pool"/>
<publish-subscribe-channel id="channelWithErrorHandler" error-handler="testErrorHandler"/>
<publish-subscribe-channel id="channelWithErrorHandler" error-handler="testErrorHandler" task-executor="pool"/>
<task:executor id="pool" pool-size="1"/>

69
spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2018 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.
@@ -26,14 +26,17 @@ import static org.junit.Assert.assertTrue;
import java.util.concurrent.Executor;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.ErrorHandler;
/**
@@ -41,61 +44,53 @@ import org.springframework.util.ErrorHandler;
* @author Gary Russell
* @author Artem Bilan
*/
@RunWith(SpringRunner.class)
public class PublishSubscribeChannelParserTests {
@Autowired
private ApplicationContext context;
@Test
public void defaultChannel() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("defaultChannel");
PublishSubscribeChannel channel = this.context.getBean("defaultChannel", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
dispatcher.setApplySequence(true);
dispatcher.addHandler(message -> { });
dispatcher.dispatch(new GenericMessage<String>("foo"));
dispatcher.dispatch(new GenericMessage<>("foo"));
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertNull(dispatcherAccessor.getPropertyValue("executor"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("ignoreFailures"));
assertTrue((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
Object mbf = this.context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, dispatcherAccessor.getPropertyValue("messageBuilderFactory"));
context.close();
}
@Test
public void ignoreFailures() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithIgnoreFailures");
PublishSubscribeChannel channel =
this.context.getBean("channelWithIgnoreFailures", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("ignoreFailures"));
context.close();
}
@Test
public void applySequenceEnabled() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithApplySequenceEnabled");
PublishSubscribeChannel channel =
this.context.getBean("channelWithApplySequenceEnabled", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("applySequence"));
context.close();
}
@Test
public void channelWithTaskExecutor() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithTaskExecutor");
PublishSubscribeChannel channel =
this.context.getBean("channelWithTaskExecutor", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
@@ -106,15 +101,12 @@ public class PublishSubscribeChannelParserTests {
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
context.close();
}
@Test
public void ignoreFailuresWithTaskExecutor() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithIgnoreFailuresAndTaskExecutor");
PublishSubscribeChannel channel =
this.context.getBean("channelWithIgnoreFailuresAndTaskExecutor", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
@@ -125,16 +117,13 @@ public class PublishSubscribeChannelParserTests {
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
context.close();
assertEquals(this.context.getBean("pool"), innerExecutor);
}
@Test
public void applySequenceEnabledWithTaskExecutor() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithApplySequenceEnabledAndTaskExecutor");
PublishSubscribeChannel channel =
this.context.getBean("channelWithApplySequenceEnabledAndTaskExecutor", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
@@ -145,21 +134,17 @@ public class PublishSubscribeChannelParserTests {
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
context.close();
assertEquals(this.context.getBean("pool"), innerExecutor);
}
@Test
public void channelWithErrorHandler() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"publishSubscribeChannelParserTests.xml", this.getClass());
PublishSubscribeChannel channel = (PublishSubscribeChannel)
context.getBean("channelWithErrorHandler");
PublishSubscribeChannel channel =
this.context.getBean("channelWithErrorHandler", PublishSubscribeChannel.class);
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
ErrorHandler errorHandler = (ErrorHandler) accessor.getPropertyValue("errorHandler");
assertNotNull(errorHandler);
assertEquals(context.getBean("testErrorHandler"), errorHandler);
context.close();
assertEquals(this.context.getBean("testErrorHandler"), errorHandler);
}
}

4
src/reference/asciidoc/channel.adoc
View File

@@ -572,6 +572,10 @@ When using this element, you can also specify the `task-executor` used for publi
<int:publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>
----
Alongside with the `Executor`, an `ErrorHandler` can be configured as well.
By default the `PublishSubscribeChannel` uses a `MessagePublishingErrorHandler` implementation to send error to the `MessageChannel` from the `errorChannel` header or a global `errorChannel` instance.
If an `Executor` is not configured, the `ErrorHandler` is ignored and exceptions are thrown directly to the caller's Thread.
If you are providing a _Resequencer_ or _Aggregator_ downstream from a `PublishSubscribeChannel`, then you can set the 'apply-sequence' property on the channel to `true`.
That will indicate that the channel should set the sequence-size and sequence-number Message headers as well as the correlation id prior to passing the Messages along.
For example, if there are 5 subscribers, the sequence-size would be set to 5, and the Messages would have sequence-number header values ranging from 1 to 5.