From 585258373eb3ccc16705036671b8eb9259b79dbb Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 17 Sep 2018 11:31:26 -0400 Subject: [PATCH] Add `IntegrationFlowDefinition.nullChannel()` (#2555) * Add `IntegrationFlowDefinition.nullChannel()` When a `NullChannel` is used in the middle of the flow, it may be not so obvious why our flow is stopped after accidentally added the next endpoint * For convenient add a terminal `nullChannel()` operator into the `IntegrationFlowDefinition` * * Add WARN about `NullChannel` subscription from the endpoints --- .../dsl/IntegrationFlowDefinition.java | 11 ++++++++ .../integration/endpoint/PollingConsumer.java | 5 ++++ .../endpoint/ReactiveStreamsConsumer.java | 9 +++++- .../dsl/flows/IntegrationFlowTests.java | 28 +++++++++++++++++-- 4 files changed, 50 insertions(+), 3 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java index 0f67cf6580..e3b05a7a14 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java @@ -2961,6 +2961,17 @@ public abstract class IntegrationFlowDefinition(this.integrationComponents, publisher); } + /** + * Add a {@value IntegrationContextUtils#NULL_CHANNEL_BEAN_NAME} bean into this flow + * definition as a terminal operator. + * @return The {@link IntegrationFlow} instance based on this definition. + * @since 5.1 + */ + public IntegrationFlow nullChannel() { + return channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME) + .get(); + } + @SuppressWarnings("unchecked") private > B register(S endpointSpec, Consumer endpointConfigurer) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index dfb1d462d2..4b8315dfba 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -25,6 +25,7 @@ import org.reactivestreams.Subscriber; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ExecutorChannelInterceptorAware; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.router.MessageRouter; @@ -62,6 +63,10 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) { Assert.notNull(inputChannel, "inputChannel must not be null"); Assert.notNull(handler, "handler must not be null"); + if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) { + logger.warn("The polling from the NullChannel does not have any effects: " + + "it doesn't forward messages sent to it. A NullChannel is the end of the flow."); + } this.inputChannel = inputChannel; this.handler = handler; if (this.inputChannel instanceof ExecutorChannelInterceptorAware) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index 4e29b0ce87..cd8b90e6ea 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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. @@ -25,6 +25,7 @@ import org.reactivestreams.Subscription; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.router.MessageRouter; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; @@ -41,6 +42,7 @@ import reactor.core.publisher.BaseSubscriber; /** * @author Artem Bilan + * * @since 5.0 */ public class ReactiveStreamsConsumer extends AbstractEndpoint implements IntegrationConsumer { @@ -72,6 +74,11 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra Assert.notNull(inputChannel, "'inputChannel' must not be null"); Assert.notNull(subscriber, "'subscriber' must not be null"); + if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) { + logger.warn("The consuming from the NullChannel does not have any effects: " + + "it doesn't forward messages sent to it. A NullChannel is the end of the flow."); + } + this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel); this.subscriber = subscriber; this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index 8805d6be93..cfcc852148 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -56,6 +56,7 @@ import org.springframework.integration.annotation.MessagingGateway; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.context.IntegrationContextUtils; @@ -481,6 +482,23 @@ public class IntegrationFlowTests { assertEquals("dedicatedTaskScheduler-1", threadNameReference.get()); } + @Autowired + private MessageChannel flowWithNullChannelInput; + + @Autowired + private NullChannel nullChannel; + + @Test + public void testNullChannelInTheEndOfFlow() { + this.nullChannel.setCountsEnabled(true); + + this.flowWithNullChannelInput.send(new GenericMessage<>("foo")); + + assertEquals(1, this.nullChannel.getSendCount()); + + this.nullChannel.setCountsEnabled(false); + } + @MessagingGateway public interface ControlBusGateway { @@ -495,7 +513,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow supplierFlow() { return IntegrationFlows.from(() -> "foo") - .transform(p -> p.toUpperCase()) + .transform(String::toUpperCase) .channel("suppliedChannel") .get(); } @@ -594,7 +612,7 @@ public class IntegrationFlowTests { .channel("foo") .fixedSubscriberChannel() .transform(Integer::parseInt) - .transform(i -> new Foo(i)) + .transform(Foo::new) .transform(new PayloadSerializingTransformer(), c -> c.autoStartup(false).id("payloadSerializingTransformer")) .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue")) @@ -841,6 +859,12 @@ public class IntegrationFlowTests { return new ThreadPoolTaskScheduler(); } + @Bean + public IntegrationFlow flowWithNullChannel() { + return IntegrationFlows.from("flowWithNullChannelInput") + .nullChannel(); + } + } @Service