Subscribe to Publisher in the NullChannel (#3448)

* Subscribe to Publisher in the NullChannel

If a payload of the message in the flow is a `org.reactivestreams.Publisher`,
it must be subscribed somewhere downstream to initiate reactive processing.
The `NullChannel` just ignores the message altogether and therefore `Publisher`
is lost

* Check the payload of the message in the `NullChannel` for `Publisher` type
and subscribe to it
* Verify in the test and mention this logic in the docs

**Cherry-pick to 5.4.x**

* Apply suggestions from code review

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java

@@ -20,6 +20,9 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.integration.IntegrationPattern;
@@ -35,13 +38,16 @@ import org.springframework.messaging.PollableChannel;
 
 /**
  * A channel implementation that essentially behaves like "/dev/null".
- * All receive() calls will return <em>null</em>, and all send() calls
- * will return <em>true</em> although no action is performed.
+ * All {@link #receive()} calls will return {@code null},
+ * and all {@link #send} calls will return {@code true} although no action is performed.
+ * Unless the payload of a sent message is a {@link Publisher} implementation, in
+ * which case the {@link Publisher#subscribe(Subscriber)} is called to initiate
+ * the reactive stream, although the data is discarded by this channel.
  * Note however that the invocations are logged at debug-level.
  *
  * @author Mark Fisher
  * @author Gary Russell
- * @author Artyem Bilan
+ * @author Artem Bilan
  */
 @IntegrationManagedResource
 public class NullChannel implements PollableChannel,
@@ -119,6 +125,31 @@ public class NullChannel implements PollableChannel,
 		if (this.loggingEnabled && this.logger.isDebugEnabled()) {
 			this.logger.debug("message sent to null channel: " + message);
 		}
+
+		Object payload = message.getPayload();
+		if (payload instanceof Publisher<?>) {
+			((Publisher<?>) payload).subscribe(
+					new Subscriber<Object>() {
+
+						@Override public void onSubscribe(Subscription subscription) {
+							subscription.request(Long.MAX_VALUE);
+						}
+
+						@Override public void onNext(Object o) {
+
+						}
+
+						@Override public void onError(Throwable t) {
+
+						}
+
+						@Override public void onComplete() {
+
+						}
+
+					});
+		}
+
 		if (this.metricsCaptor != null) {
 			sendTimer().record(0, TimeUnit.MILLISECONDS);
 		}

spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java

@@ -19,6 +19,8 @@ package org.springframework.integration.channel;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.Test;
@@ -28,6 +30,7 @@ import org.springframework.messaging.support.GenericMessage;
 
 import reactor.core.Disposable;
 import reactor.core.Disposables;
+import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.util.concurrent.Queues;
@@ -104,4 +107,13 @@ class MessageChannelReactiveUtilsTests {
 				.isLessThanOrEqualTo(Queues.SMALL_BUFFER_SIZE);
 	}
 
+	@Test
+	void testPublisherPayloadWithNullChannel() throws InterruptedException {
+		NullChannel nullChannel = new NullChannel();
+		CountDownLatch publisherSubscribed = new CountDownLatch(1);
+		Mono<Object> mono = Mono.empty().doOnSubscribe((s) -> publisherSubscribed.countDown());
+		nullChannel.send(new GenericMessage<>(mono));
+		assertThat(publisherSubscribed.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
 }

src/reference/asciidoc/channel.adoc

@@ -1142,9 +1142,11 @@ For example, you can use this technique to configure a test case to verify messa
 [[channel-special-channels]]
 ==== Special Channels
 
-If namespace support is enabled, two special channels are defined within the application context by default: `errorChannel` and `nullChannel`.
-The 'nullChannel' acts like `/dev/null`, logging any message sent to it at the `DEBUG` level and returning immediately.
+Two special channels are defined within the application context by default: `errorChannel` and `nullChannel`.
+The 'nullChannel' (an instance of `NullChannel`) acts like `/dev/null`, logging any message sent to it at the `DEBUG` level and returning immediately.
+The special treatment is applied for an `org.reactivestreams.Publisher` payload of a sent message: it is subscribed to in this channel immediately, to initiate reactive stream processing, although the data is discarded.
 Any time you face channel resolution errors for a reply that you do not care about, you can set the affected component's `output-channel` attribute to 'nullChannel' (the name, 'nullChannel', is reserved within the application context).
+
 The 'errorChannel' is used internally for sending error messages and may be overridden with a custom configuration.
 This is discussed in greater detail in <<./error-handling.adoc#error-handling,Error Handling>>.