From c38da2f4ee511ababc302773b95a1bb1d42a1990 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 22 Dec 2020 12:20:12 -0500 Subject: [PATCH] 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 Co-authored-by: Gary Russell --- .../integration/channel/NullChannel.java | 37 +++++++++++++++++-- .../MessageChannelReactiveUtilsTests.java | 12 ++++++ src/reference/asciidoc/channel.adoc | 6 ++- 3 files changed, 50 insertions(+), 5 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java index 09d28aec2e..b8594187b2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java +++ b/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 null, and all send() calls - * will return true 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() { + + @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); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java index bc0216db05..45436f3a33 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/MessageChannelReactiveUtilsTests.java +++ b/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 mono = Mono.empty().doOnSubscribe((s) -> publisherSubscribed.countDown()); + nullChannel.send(new GenericMessage<>(mono)); + assertThat(publisherSubscribed.await(10, TimeUnit.SECONDS)).isTrue(); + } + } diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index 9373e24ef3..bf87dfa0b9 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/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>>.