diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 86583239bd..0fe3aaff86 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -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. @@ -36,19 +36,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) { @@ -80,9 +79,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) { @@ -147,6 +144,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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests-context.xml similarity index 96% rename from spring-integration-core/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml rename to spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests-context.xml index a9fc6d0001..d36f6fb7e0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests-context.xml @@ -19,7 +19,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java index a8cb2c0126..855e7769ed 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java @@ -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,9 +26,11 @@ 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; @@ -37,6 +39,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.ErrorHandler; /** @@ -44,14 +47,15 @@ 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"); @@ -69,43 +73,34 @@ public class PublishSubscribeChannelParserTests { 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"); @@ -116,15 +111,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"); @@ -135,16 +127,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"); @@ -155,21 +144,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); } } diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index 7646f871e4..7bd59a1a12 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/src/reference/asciidoc/channel.adoc @@ -565,6 +565,10 @@ When using this element, you can also specify the `task-executor` used for publi ---- +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.