GH-3450: Expose PubSub requireSubscribers option (#3459)

* GH-3450: Expose PubSub requireSubscribers option

Fixes https://github.com/spring-projects/spring-integration/issues/3450

* Add `PublishSubscribeChannel` ctors with the `requireSubscribers` option
with a direct delegation to the same option of underlying `BroadcastingDispatcher`
* Add factory methods to the `Channels` & `MessageChannels` to expose
this new `requireSubscribers` option
* Expose an XML `require-subscribers` attribute for the `<publish-subscribe-channel`
* Document this new `requireSubscribers` option
* Introduce a global `spring.integration.channels.error.requireSubscribers` property
for a default `errorChannel`
* Document this property and explain its default `true` in the `error-handling.adoc`
* Remove docs for `spring.integration.postProcessDynamicBeans` since it purpose was
removed since version `5.1` in favor of "always post-process behavior" as it was
always with all the `BeanPostProcessor`s

* * Fix language in docs according review

* * Add more docs about `requireSubscribers` and cross-links between chapters

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 <i>only</i> 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 <code>false</code> 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 <code>true</code>.
+	 * 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 <code>false</code> meaning that sequence headers will
-	 * <em>not</em> 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 <code>true</code>.
+	 * 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);

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)

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");

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

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

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() {

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<S extends PublishSubscribeChannelSpec<S
 		extends MessageChannelSpec<S, PublishSubscribeChannel> {
 
 	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) {

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

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration.xsd

@@ -537,13 +537,23 @@
 					</xsd:attribute>
 					<xsd:attribute name="min-subscribers" type="xsd:string">
 						<xsd:annotation>
-							<xsd:documentation><![CDATA[
+							<xsd:documentation>
 								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.
-							]]></xsd:documentation>
+							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="require-subscribers" default="false">
+						<xsd:annotation>
+							<xsd:documentation>
+								Indicates if this channel may ignore or not a message when there are no subscribers.
+							</xsd:documentation>
+						</xsd:annotation>
+						<xsd:simpleType>
+							<xsd:union memberTypes="xsd:boolean xsd:string"/>
+						</xsd:simpleType>
+					</xsd:attribute>
 					<xsd:attributeGroup ref="subscribersAttributeGroup"/>
 				</xsd:extension>
 			</xsd:complexContent>

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

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">
-
-	<util:properties id="integrationGlobalProperties">
-		<prop key="spring.integration.postProcessDynamicBeans">true</prop>
-	</util:properties>
+		http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
 
 	<int:channel id="inputA">
 		<int:interceptors>

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

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
 

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.

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.