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 a182ea3d3f..c8a360798a 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-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -39,6 +39,8 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements private ErrorHandler errorHandler; + private final boolean requireSubscribers; + private boolean ignoreFailures; private boolean applySequence; @@ -50,7 +52,18 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements * message sender's thread. */ public PublishSubscribeChannel() { - this(null); + this(false); + } + + /** + * Create a PublishSubscribeChannel that will invoke the handlers in the + * message sender's thread considering the provided {@code requireSubscribers} flag. + * @param requireSubscribers if set to true, the sent message is considered as non-dispatched + * and rejected to the caller with the {@code "Dispatcher has no subscribers"}. + * @since 5.4.3 + */ + public PublishSubscribeChannel(boolean requireSubscribers) { + this(null, requireSubscribers); } /** @@ -60,8 +73,22 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements * @param executor The executor. */ public PublishSubscribeChannel(@Nullable Executor executor) { + this(executor, false); + } + + /** + * 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. + * @param requireSubscribers if set to true, the sent message is considered as non-dispatched + * and rejected to the caller with the {@code "Dispatcher has no subscribers"}. + * @since 5.4.3 + */ + public PublishSubscribeChannel(@Nullable Executor executor, boolean requireSubscribers) { super(executor); - this.dispatcher = new BroadcastingDispatcher(executor); + this.requireSubscribers = requireSubscribers; + this.dispatcher = new BroadcastingDispatcher(executor, requireSubscribers); } @@ -77,7 +104,7 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements /** * Provide an {@link ErrorHandler} strategy for handling Exceptions that - * occur downstream from this channel. This will only be applied if + * occur downstream from this channel. This will only be applied if * an Executor has been configured to dispatch the Messages for this * channel. Otherwise, Exceptions will be thrown directly within the * sending Thread. If no ErrorHandler is provided, and this channel does @@ -94,9 +121,9 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements /** * Specify whether failures for one or more of the handlers should be - * ignored. By default this is false meaning that an Exception + * ignored. By default this is false meaning that an Exception * will be thrown whenever a handler fails. To override this and suppress - * Exceptions, set the value to true. + * Exceptions, set the value to true. * @param ignoreFailures true if failures should be ignored. */ public void setIgnoreFailures(boolean ignoreFailures) { @@ -107,10 +134,10 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements /** * Specify whether to apply the sequence number and size headers to the * messages prior to invoking the subscribed handlers. By default, this - * value is false meaning that sequence headers will - * not be applied. If planning to use an Aggregator downstream + * value is false meaning that sequence headers will + * not be applied. If planning to use an Aggregator downstream * with the default correlation and completion strategies, you should set - * this flag to true. + * this flag to true. * @param applySequence true if the sequence information should be applied. */ public void setApplySequence(boolean applySequence) { @@ -147,7 +174,7 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel implements } this.executor = new ErrorHandlingTaskExecutor(this.executor, this.errorHandler); } - dispatcherToUse = new BroadcastingDispatcher(this.executor); + dispatcherToUse = new BroadcastingDispatcher(this.executor, this.requireSubscribers); dispatcherToUse.setIgnoreFailures(this.ignoreFailures); dispatcherToUse.setApplySequence(this.applySequence); dispatcherToUse.setMinSubscribers(this.minSubscribers); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java index 8147979193..5a43b5a211 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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. @@ -215,7 +215,10 @@ class DefaultConfiguringBeanFactoryPostProcessor "Therefore, a default PublishSubscribeChannel will be created."); } this.registry.registerBeanDefinition(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, - new RootBeanDefinition(PublishSubscribeChannel.class)); + BeanDefinitionBuilder.rootBeanDefinition(PublishSubscribeChannel.class) + .addConstructorArgValue(IntegrationProperties.getExpressionFor( + IntegrationProperties.ERROR_CHANNEL_REQUIRE_SUBSCRIBERS)) + .getBeanDefinition()); BeanDefinition loggingHandler = BeanDefinitionBuilder.genericBeanDefinition(LoggingHandler.class) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java index bc584b2c67..668cb56270 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PublishSubscribeChannelParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -40,6 +40,7 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser { if (StringUtils.hasText(taskExecutorRef)) { builder.addConstructorArgReference(taskExecutorRef); } + builder.addConstructorArgValue(element.getAttribute("require-subscribers")); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-failures"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java index 235bc67a4f..8020974915 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2021 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. @@ -47,14 +47,24 @@ public final class IntegrationProperties { * in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.ExecutorChannel}), * if the attribute {@code max-subscribers} isn't configured on the channel component. */ - public static final String CHANNELS_MAX_UNICAST_SUBSCRIBERS = INTEGRATION_PROPERTIES_PREFIX + "channels.maxUnicastSubscribers"; + public static final String CHANNELS_MAX_UNICAST_SUBSCRIBERS = + INTEGRATION_PROPERTIES_PREFIX + "channels.maxUnicastSubscribers"; /** * Specifies the value for {@link org.springframework.integration.dispatcher.BroadcastingDispatcher#maxSubscribers} * in case of point-to-point channels (e.g. {@link org.springframework.integration.channel.PublishSubscribeChannel}), * if the attribute {@code max-subscribers} isn't configured on the channel component. */ - public static final String CHANNELS_MAX_BROADCAST_SUBSCRIBERS = INTEGRATION_PROPERTIES_PREFIX + "channels.maxBroadcastSubscribers"; + public static final String CHANNELS_MAX_BROADCAST_SUBSCRIBERS = + INTEGRATION_PROPERTIES_PREFIX + "channels.maxBroadcastSubscribers"; + + + /** + * Specifies the value for {@link org.springframework.integration.channel.PublishSubscribeChannel#requireSubscribers} + * on a global default {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}. + */ + public static final String ERROR_CHANNEL_REQUIRE_SUBSCRIBERS = + INTEGRATION_PROPERTIES_PREFIX + "channels.error.requireSubscribers"; /** * Specifies the value of {@link org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler#poolSize} @@ -65,7 +75,8 @@ public final class IntegrationProperties { /** * Specifies the value of {@link org.springframework.messaging.core.GenericMessagingTemplate#throwExceptionOnLateReply}. */ - public static final String THROW_EXCEPTION_ON_LATE_REPLY = INTEGRATION_PROPERTIES_PREFIX + "messagingTemplate.throwExceptionOnLateReply"; + public static final String THROW_EXCEPTION_ON_LATE_REPLY = + INTEGRATION_PROPERTIES_PREFIX + "messagingTemplate.throwExceptionOnLateReply"; /** * Specifies the value of {@link org.springframework.integration.support.DefaultMessageBuilderFactory#readOnlyHeaders}. @@ -77,18 +88,19 @@ public final class IntegrationProperties { */ public static final String ENDPOINTS_NO_AUTO_STARTUP = INTEGRATION_PROPERTIES_PREFIX + "endpoints.noAutoStartup"; - private static Properties defaults; + private static final Properties DEFAULTS; static { String resourcePattern = "classpath*:META-INF/spring.integration.default.properties"; try { - ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver(IntegrationProperties.class.getClassLoader()); + ResourcePatternResolver resourceResolver = + new PathMatchingResourcePatternResolver(IntegrationProperties.class.getClassLoader()); Resource[] defaultResources = resourceResolver.getResources(resourcePattern); PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean(); propertiesFactoryBean.setLocations(defaultResources); propertiesFactoryBean.afterPropertiesSet(); - defaults = propertiesFactoryBean.getObject(); + DEFAULTS = propertiesFactoryBean.getObject(); } catch (IOException e) { throw new IllegalStateException("Can't load '" + resourcePattern + "' resources.", e); @@ -100,7 +112,7 @@ public final class IntegrationProperties { * from resources 'META-INF/spring.integration.default.properties'. */ public static Properties defaults() { - return defaults; + return DEFAULTS; } /** @@ -112,11 +124,13 @@ public final class IntegrationProperties { * @throws IllegalArgumentException if provided {@code key} isn't an Integration property. */ public static String getExpressionFor(String key) { - if (defaults.containsKey(key)) { - return "#{T(org.springframework.integration.context.IntegrationContextUtils).getIntegrationProperties(beanFactory).getProperty('" + key + "')}"; + if (DEFAULTS.containsKey(key)) { + return "#{T(org.springframework.integration.context.IntegrationContextUtils)" + + ".getIntegrationProperties(beanFactory).getProperty('" + key + "')}"; } else { - throw new IllegalArgumentException("The provided key [" + key + "] isn't the one of Integration properties: " + defaults.keySet()); + throw new IllegalArgumentException("The provided key [" + key + + "] isn't the one of Integration properties: " + DEFAULTS.keySet()); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java index 07f97ac96e..b3c9770ba0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2021 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. @@ -98,22 +98,38 @@ public final class Channels { return MessageChannels.rendezvous(id); } - public PublishSubscribeChannelSpec publishSubscribe() { + public PublishSubscribeChannelSpec publishSubscribe() { return MessageChannels.publishSubscribe(); } + public PublishSubscribeChannelSpec publishSubscribe(boolean requireSubscribers) { + return MessageChannels.publishSubscribe(requireSubscribers); + } + public PublishSubscribeChannelSpec publishSubscribe(Executor executor) { return MessageChannels.publishSubscribe(executor); } + public PublishSubscribeChannelSpec publishSubscribe(Executor executor, boolean requireSubscribers) { + return MessageChannels.publishSubscribe(executor, requireSubscribers); + } + public PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { return MessageChannels.publishSubscribe(id, executor); } + public PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor, boolean requireSubscribers) { + return MessageChannels.publishSubscribe(id, executor, requireSubscribers); + } + public PublishSubscribeChannelSpec publishSubscribe(String id) { return MessageChannels.publishSubscribe(id); } + public PublishSubscribeChannelSpec publishSubscribe(String id, boolean requireSubscribers) { + return MessageChannels.publishSubscribe(id, requireSubscribers); + } + public ExecutorChannelSpec executor(Executor executor) { return MessageChannels.executor(executor); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageChannels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageChannels.java index 72ca43eb42..790a502f9c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageChannels.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageChannels.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2021 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. @@ -108,19 +108,37 @@ public final class MessageChannels { } public static PublishSubscribeChannelSpec publishSubscribe() { - return new PublishSubscribeChannelSpec<>(); + return publishSubscribe(false); + } + + public static PublishSubscribeChannelSpec publishSubscribe(boolean requireSubscribers) { + return new PublishSubscribeChannelSpec<>(requireSubscribers); } public static PublishSubscribeChannelSpec publishSubscribe(String id) { - return MessageChannels.publishSubscribe().id(id); + return publishSubscribe(id, false); + } + + public static PublishSubscribeChannelSpec publishSubscribe(String id, boolean requireSubscribers) { + return publishSubscribe(requireSubscribers).id(id); } public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) { - return new PublishSubscribeChannelSpec<>(executor); + return publishSubscribe(executor, false); + } + + public static PublishSubscribeChannelSpec publishSubscribe(Executor executor, boolean requireSubscribers) { + return new PublishSubscribeChannelSpec<>(executor, requireSubscribers); } public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { - return MessageChannels.publishSubscribe(executor).id(id); + return publishSubscribe(id, executor, false); + } + + public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor, + boolean requireSubscribers) { + + return publishSubscribe(executor, requireSubscribers).id(id); } public static FluxMessageChannelSpec flux() { @@ -128,8 +146,7 @@ public final class MessageChannels { } public static FluxMessageChannelSpec flux(String id) { - return flux() - .id(id); + return flux().id(id); } private MessageChannels() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeChannelSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeChannelSpec.java index d1f98d1849..61a7875153 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeChannelSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeChannelSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2021 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. @@ -34,11 +34,19 @@ public class PublishSubscribeChannelSpec { protected PublishSubscribeChannelSpec() { - this(null); + this(false); + } + + protected PublishSubscribeChannelSpec(boolean requireSubscribers) { + this(null, requireSubscribers); } protected PublishSubscribeChannelSpec(@Nullable Executor executor) { - this.channel = new PublishSubscribeChannel(executor); + this(executor, false); + } + + protected PublishSubscribeChannelSpec(@Nullable Executor executor, boolean requireSubscribers) { + this.channel = new PublishSubscribeChannel(executor, requireSubscribers); } public S errorHandler(ErrorHandler errorHandler) { diff --git a/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties b/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties index afdae85c7f..d75ad359de 100644 --- a/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties +++ b/spring-integration-core/src/main/resources/META-INF/spring.integration.default.properties @@ -1,6 +1,7 @@ spring.integration.channels.autoCreate=true spring.integration.channels.maxUnicastSubscribers=0x7fffffff spring.integration.channels.maxBroadcastSubscribers=0x7fffffff +spring.integration.channels.error.requireSubscribers=true spring.integration.taskScheduler.poolSize=10 spring.integration.messagingTemplate.throwExceptionOnLateReply=false # Defaults to MessageHeaders.ID and MessageHeaders.TIMESTAMP diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd index aa572b4b25..97d8d22932 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd @@ -537,13 +537,23 @@ - Specifies the minimum subscribers required to be subscribed to this channel; if the minimum number of subscribers receive the message, the send is deemed to be successful (returns true). Defaults to 0. - ]]> + + + + + Indicates if this channel may ignore or not a message when there are no subscribers. + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/PublishSubscribeChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/PublishSubscribeChannelTests.java index e2f4794d48..0b973bfdce 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/PublishSubscribeChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/PublishSubscribeChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2021 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. @@ -16,18 +16,23 @@ package org.springframework.integration.channel; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import static org.mockito.Mockito.mock; import java.util.concurrent.Executor; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.MessageDispatchingException; +import org.springframework.messaging.MessageDeliveryException; +import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell + * @author Artem Bilan + * * @since 5.0 * */ @@ -36,17 +41,21 @@ public class PublishSubscribeChannelTests { @Test public void testEarlySubscribe() { PublishSubscribeChannel channel = new PublishSubscribeChannel(mock(Executor.class)); - try { - channel.subscribe(m -> { }); - channel.setBeanFactory(mock(BeanFactory.class)); - channel.afterPropertiesSet(); - fail("expected Exception"); - } - catch (IllegalStateException e) { - assertThat(e.getMessage()).isEqualTo("When providing an Executor, you cannot subscribe() until the " + - "channel " - + "bean is fully initialized by the framework. Do not subscribe in a @Bean definition"); - } + channel.subscribe(m -> { }); + channel.setBeanFactory(mock(BeanFactory.class)); + assertThatIllegalStateException() + .isThrownBy(channel::afterPropertiesSet) + .withMessage("When providing an Executor, you cannot subscribe() until the channel " + + "bean is fully initialized by the framework. Do not subscribe in a @Bean definition"); + } + + @Test + public void testRequireSubscribers() { + PublishSubscribeChannel channel = new PublishSubscribeChannel(true); + assertThatExceptionOfType(MessageDeliveryException.class) + .isThrownBy(() -> channel.send(new GenericMessage<>("test"))) + .withCauseInstanceOf(MessageDispatchingException.class) + .withMessageContaining("Dispatcher has no subscribers"); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests-context.xml index 7acb9d40b9..233d9daffd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests-context.xml @@ -3,16 +3,10 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration" xmlns:p="http://www.springframework.org/schema/p" - xmlns:aop="http://www.springframework.org/schema/aop" xmlns:util="http://www.springframework.org/schema/util" - xmlns:beans="http://www.springframework.org/schema/c" + xmlns:aop="http://www.springframework.org/schema/aop" xsi:schemaLocation="http://www.springframework.org/schema/aop https://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd - http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd - http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd"> - - - true - + http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd"> diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ErrorChannelAutoCreationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ErrorChannelAutoCreationTests.java index 705b8c69f6..032e0464af 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ErrorChannelAutoCreationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ErrorChannelAutoCreationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -18,30 +18,29 @@ package org.springframework.integration.config.xml; import static org.assertj.core.api.Assertions.assertThat; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Oleg Zhurakousky * @author Artem Bilan */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig public class ErrorChannelAutoCreationTests { @Autowired private MessageChannel errorChannel; - // see INT-1899 @Test public void testErrorChannelIsPubSub() { - assertThat(errorChannel.getClass()).isEqualTo(PublishSubscribeChannel.class); + assertThat(this.errorChannel).isInstanceOf(PublishSubscribeChannel.class); + assertThat(TestUtils.getPropertyValue(this.errorChannel, "dispatcher.requireSubscribers", Boolean.class)) + .isTrue(); } } diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index 33ad4118e1..d5d6ff5d46 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/src/reference/asciidoc/channel.adoc @@ -794,6 +794,9 @@ public MessageChannel pubsubChannel() { NOTE: The `apply-sequence` value is `false` by default so that a publish-subscribe channel can send the exact same message instances to multiple outbound channels. Since Spring Integration enforces immutability of the payload and header references, when the flag is set to `true`, the channel creates new `Message` instances with the same payload reference but different header values. +Starting with version 5.4.3, the `PublishSubscribeChannel` can also be configured with the `requireSubscribers` option of its `BroadcastingDispatcher` to indicate that this channel will not ignore a message silently when it has no subscribers. +A `MessageDispatchingException` with a `Dispatcher has no subscribers` message is thrown when there are no subscribers and this option is set to `true`. + [[channel-configuration-executorchannel]] ===== `ExecutorChannel` @@ -1017,9 +1020,6 @@ To inject a global interceptor before the existing interceptors, use a negative NOTE: Note that both the `order` and `pattern` attributes are optional. The default value for `order` will be 0 and for `pattern`, the default is '*' (to match all channels). -Starting with version 4.3.15, you can configure the `spring.integration.postProcessDynamicBeans = true` property to apply any global interceptors to dynamically created `MessageChannel` beans. -See <<./configuration.adoc#global-properties,Global Properties>> for more information. - [[channel-wiretap]] ===== Wire Tap diff --git a/src/reference/asciidoc/configuration.adoc b/src/reference/asciidoc/configuration.adoc index cf22b4c8ad..5bd6ba2b2d 100644 --- a/src/reference/asciidoc/configuration.adoc +++ b/src/reference/asciidoc/configuration.adoc @@ -168,7 +168,7 @@ spring.integration.taskScheduler.poolSize=10 <4> spring.integration.messagingTemplate.throwExceptionOnLateReply=false <5> spring.integration.readOnly.headers= <6> spring.integration.endpoints.noAutoStartup= <7> -spring.integration.postProcessDynamicBeans=false <8> +spring.integration.channels.error.requireSubscribers=true <8> ---- <1> When true, `input-channel` instances are automatically declared as `DirectChannel` instances when not explicitly found in the application context. @@ -196,8 +196,9 @@ You can manually start these endpoints later by their bean name through a `Contr You can explicitly override the effect of this global property by specifying `auto-startup` XML annotation or the `autoStartup` annotation attribute or by calling `AbstractEndpoint.setAutoStartup()` in the bean definition. Since version 4.3.12. -<8> A boolean flag to indicate that `BeanPostProcessor` instances should post-process beans registered at runtime (for example, message channels created by `IntegrationFlowContext` can be supplied with global channel interceptors). -Since version 4.3.15. +<8> A boolean flag to indicate that default global `errorChannel` must be configured with the `requireSubscribers` option. +Since version 5.4.3. +See <<./error-handling.adoc#error-handling,Error Handling>> for more information. ==== These properties can be overridden by adding a `/META-INF/spring.integration.properties` file to the classpath. diff --git a/src/reference/asciidoc/error-handling.adoc b/src/reference/asciidoc/error-handling.adoc index 9200afde29..3efdad71cf 100644 --- a/src/reference/asciidoc/error-handling.adoc +++ b/src/reference/asciidoc/error-handling.adoc @@ -70,3 +70,8 @@ With Java & Annotation configuration, a resource is a `@Configuration` class and In most case the target integration flow solution is based on the out-of-the-box components and their configuration options. When an exception happens at runtime, there is no any end-user code involved in stack trace because an execution is against beans, not their configuration. Including a resource and source of the bean definition helps to determine possible configuration mistakes and provides better developer experience. + +Starting with version 5.4.3, the default error channel is configured with the property `requireSubscribers = true` to not silently ignore messages when there are no subscribers on this channel (e.g. when application context is stopped). +In this case a `MessageDispatchingException` is thrown which may lend on the client callback of the inbound channel adapter to negatively acknowledge (or roll back) an original message in the source system for redelivery or other future consideration. +To restore the previous behavior (ignore non dispatched error messages), the global integration property `spring.integration.channels.error.requireSubscribers` must be set to `false`. +See <<./configuration.adoc#global-properties,Global Properties>> and <<./channel.adoc#channel-configuration-pubsubchannel,`PublishSubscribeChannel` Configuration>> (if you configure a global `errorChannel` manually) for more information.