From d070eb5783264706b30330abc4945f077179cff0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 5 Feb 2021 10:50:17 -0500 Subject: [PATCH] GH-3464: Treat 0 as special for polling endpoint (#3487) * GH-3464: Treat 0 as special for polling endpoint Fixes https://github.com/spring-projects/spring-integration/issues/3464 The `maxMessagePerPoll <= 0` is considered as an unbound `receive()` call. End-users would like to have a special treatment for `0` value - skip the `receive()` call altogether for the current polling cycle. * Change the logic for a scheduled poller to not call `pollForMessage()` and just log an INFO when `maxMessagePerPoll == 0` * Fix reactive poller to deal with `maxMessagePerPoll == 0` properly * Expose `maxMessagesPerPoll` as a `@ManagedAttribute` to let it to be modified via Control Bus and JMX * Test and document a new behavior * * Fix unused imports in the test class * Fix language in docs accoridng PR review Co-authored-by: Gary Russell Co-authored-by: Gary Russell --- .../endpoint/AbstractPollingEndpoint.java | 45 +++++++++++++++---- ...PollingChannelAdapterFactoryBeanTests.java | 43 ++++++++++++++++++ .../ReactiveInboundChannelAdapterTests.java | 12 ++++- src/reference/asciidoc/channel-adapter.adoc | 4 +- src/reference/asciidoc/endpoint.adoc | 4 +- src/reference/asciidoc/reactive-streams.adoc | 1 + src/reference/asciidoc/whats-new.adoc | 4 ++ 7 files changed, 101 insertions(+), 12 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 635df75d2b..ccd6dc3787 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -45,6 +45,7 @@ import org.springframework.integration.transaction.IntegrationResourceHolderSync import org.springframework.integration.transaction.PassThroughTransactionSynchronizationFactory; import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.integration.util.ErrorHandlingTaskExecutor; +import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; @@ -145,10 +146,23 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.adviceChain = adviceChain; } + /** + * Configure a cap for messages to poll from the source per scheduling cycle. +* A negative number means retrieve unlimited messages until the {@code MessageSource} returns {@code null}. + * Zero means do not poll for any records - it + * can be considered as pausing if 'maxMessagesPerPoll' is later changed to a non-zero value. + * The polling cycle may exit earlier if the source returns null for the current receive call. + * @param maxMessagesPerPoll the number of message to poll per schedule. + */ + @ManagedAttribute public void setMaxMessagesPerPoll(long maxMessagesPerPoll) { this.maxMessagesPerPoll = maxMessagesPerPoll; } + public long getMaxMessagesPerPoll() { + return this.maxMessagesPerPoll; + } + public void setErrorHandler(ErrorHandler errorHandler) { this.errorHandler = errorHandler; } @@ -327,6 +341,10 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.taskExecutor.execute(() -> { int count = 0; while (this.initialized && (this.maxMessagesPerPoll <= 0 || count < this.maxMessagesPerPoll)) { + if (this.maxMessagesPerPoll == 0) { + logger.info("Polling disabled while 'maxMessagesPerPoll == 0'"); + break; + } if (pollForMessage() == null) { break; } @@ -357,19 +375,28 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement new Date(), null)) .flatMapMany(l -> Flux - .>generate(fluxSink -> { - Message message = pollForMessage(); - if (message != null) { - fluxSink.next(message); + .defer(() -> { + if (this.maxMessagesPerPoll == 0) { + logger.info("Polling disabled while 'maxMessagesPerPoll == 0'"); + return Mono.empty(); } else { - fluxSink.complete(); + return Flux + .>generate(fluxSink -> { + Message message = pollForMessage(); + if (message != null) { + fluxSink.next(message); + } + else { + fluxSink.complete(); + } + }) + .limitRequest( + this.maxMessagesPerPoll < 0 + ? Long.MAX_VALUE + : this.maxMessagesPerPoll); } }) - .limitRequest( - this.maxMessagesPerPoll < 0 - ? Long.MAX_VALUE - : this.maxMessagesPerPoll) .subscribeOn(Schedulers.fromExecutor(this.taskExecutor)) .doOnComplete(() -> triggerContext diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java index 654f52ec6a..97a81010f1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java @@ -18,7 +18,9 @@ package org.springframework.integration.config; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; @@ -222,6 +224,47 @@ public class SourcePollingChannelAdapterFactoryBeanTests { pollingChannelAdapter.stop(); } + @Test + public void testZeroForMaxMessagesPerPoll() throws InterruptedException { + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.afterPropertiesSet(); + + SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter(); + pollingChannelAdapter.setTaskScheduler(taskScheduler); + pollingChannelAdapter.setSource(() -> new GenericMessage<>("test")); + pollingChannelAdapter.setTrigger(new PeriodicTrigger(1)); + pollingChannelAdapter.setMaxMessagesPerPoll(0); + QueueChannel outputChannel = new QueueChannel(); + pollingChannelAdapter.setOutputChannel(outputChannel); + pollingChannelAdapter.setBeanFactory(mock(BeanFactory.class)); + + LogAccessor logger = spy(TestUtils.getPropertyValue(pollingChannelAdapter, "logger", LogAccessor.class)); + new DirectFieldAccessor(pollingChannelAdapter).setPropertyValue("logger", logger); + + CountDownLatch logCalledLatch = new CountDownLatch(1); + + willAnswer(invocation -> { + logCalledLatch.countDown(); + return invocation.callRealMethod(); + }) + .given(logger) + .info(anyString()); + + pollingChannelAdapter.afterPropertiesSet(); + pollingChannelAdapter.start(); + + assertThat(logCalledLatch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(logger, atLeastOnce()).info("Polling disabled while 'maxMessagesPerPoll == 0'"); + + pollingChannelAdapter.setMaxMessagesPerPoll(1); + + Message receive = outputChannel.receive(10_000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("test"); + pollingChannelAdapter.stop(); + } + + private static class LifecycleMessageSource implements MessageSource, Lifecycle { private volatile boolean running; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java index 571289bc0f..3f1d14ebaa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java @@ -28,10 +28,12 @@ import java.util.function.Supplier; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.annotation.EndpointId; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Poller; import org.springframework.integration.channel.FluxMessageChannel; @@ -56,6 +58,10 @@ public class ReactiveInboundChannelAdapterTests { @Autowired private FluxMessageChannel fluxChannel; + @Autowired + @Qualifier("counterEndpoint") + private AbstractPollingEndpoint abstractPollingEndpoint; + @Test public void testReactiveInboundChannelAdapter() { Flux testFlux = @@ -64,6 +70,9 @@ public class ReactiveInboundChannelAdapterTests { .cast(Integer.class); StepVerifier.create(testFlux) + .expectSubscription() + .expectNoEvent(Duration.ofSeconds(1)) + .then(() -> abstractPollingEndpoint.setMaxMessagesPerPoll(-1)) .expectNext(2, 4, 6, 8, 10, 12, 14, 16) .thenCancel() .verify(Duration.ofSeconds(10)); @@ -109,7 +118,8 @@ public class ReactiveInboundChannelAdapterTests { @Bean @InboundChannelAdapter(value = "fluxChannel", - poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "-1", taskExecutor = "taskExecutor")) + poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "0", taskExecutor = "taskExecutor")) + @EndpointId("counterEndpoint") public Supplier counterMessageSupplier() { return () -> { int i = counter().incrementAndGet(); diff --git a/src/reference/asciidoc/channel-adapter.adoc b/src/reference/asciidoc/channel-adapter.adoc index 7fc51af882..2f7bee13c1 100644 --- a/src/reference/asciidoc/channel-adapter.adoc +++ b/src/reference/asciidoc/channel-adapter.adoc @@ -62,7 +62,7 @@ However, what happens if the configuration looks like the following example: ==== Note that there is no `max-messages-per-poll` specified. -As we cover later, the identical poller configuration in the `PollingConsumer` (for example, service-activator, filter, router, and others) would have a default value of -1 for `max-messages-per-poll`, which means "`execute the polling task non-stop unless the polling method returns null (perhaps because there are no more messages in the `QueueChannel`)`" and then sleep for one second. +As we cover later, the identical poller configuration in the `PollingConsumer` (for example, `service-activator`, `filter`, `router`, and others) would have a default value of `-1` for `max-messages-per-poll`, which means "`execute the polling task non-stop unless the polling method returns null (perhaps because there are no more messages in the `QueueChannel`)`" and then sleep for one second. However, in the `SourcePollingChannelAdapter`, it is a bit different. The default value for `max-messages-per-poll` is `1`, unless you explicitly set it to a negative value (such as `-1`). @@ -76,6 +76,8 @@ However, if you are sure that your method can return null and you need to poll f ---- ==== + +Starting with version 5.5, a `0` value for `max-messages-per-poll` has a special meaning - skip the `MessageSource.receive()` call altogether, which may be considered as pausing for this inbound channel adapter until the `maxMessagesPerPoll` is changed to a non-zero value at a later time, e.g. via a Control Bus. ===== [[channel-adapter-namespace-outbound]] diff --git a/src/reference/asciidoc/endpoint.adoc b/src/reference/asciidoc/endpoint.adoc index f6eeab8f19..249c5ae6d6 100644 --- a/src/reference/asciidoc/endpoint.adoc +++ b/src/reference/asciidoc/endpoint.adoc @@ -141,9 +141,11 @@ consumer.setReceiveTimeout(5000); ==== The `maxMessagesPerPoll` property specifies the maximum number of messages to receive within a given poll operation. -This means that the poller continues calling receive() without waiting, until either `null` is returned or the maximum value is reached. +This means that the poller continues calling `receive()` without waiting, until either `null` is returned or the maximum value is reached. For example, if a poller has a ten-second interval trigger and a `maxMessagesPerPoll` setting of `25`, and it is polling a channel that has 100 messages in its queue, all 100 messages can be retrieved within 40 seconds. It grabs 25, waits ten seconds, grabs the next 25, and so on. +If `maxMessagesPerPoll` is configured with a negative value, then `MessageSource.receive()` is called within a single polling cycle until it returns `null`. +Starting with version 5.5, a `0` value has a special meaning - skip the `MessageSource.receive()` call altogether, which may be considered as pausing for this polling endpoint until the `maxMessagesPerPoll` is changed to a n non-zero value at a later time, e.g. via a Control Bus. The `receiveTimeout` property specifies the amount of time the poller should wait if no messages are available when it invokes the receive operation. For example, consider two options that seem similar on the surface but are actually quite different: The first has an interval trigger of 5 seconds and a receive timeout of 50 milliseconds, while the second has an interval trigger of 50 milliseconds and a receive timeout of 5 seconds. diff --git a/src/reference/asciidoc/reactive-streams.adoc b/src/reference/asciidoc/reactive-streams.adoc index 0055a86b82..7d92601260 100644 --- a/src/reference/asciidoc/reactive-streams.adoc +++ b/src/reference/asciidoc/reactive-streams.adoc @@ -73,6 +73,7 @@ A polling trigger is built from the provided options and used for periodic sched When an `outputChannel` is a `ReactiveStreamsSubscribableChannel`, the same `Trigger` is used to determine the next time for execution, but instead of scheduling tasks, the `SourcePollingChannelAdapter` creates a `Flux>` based on the `Flux.generate()` for the `nextExecutionTime` values and `Mono.delay()` for a duration from the previous step. A `Flux.flatMapMany()` is used then to poll `maxMessagesPerPoll` and sink them into an output `Flux`. This generator `Flux` is subscribed by the provided `ReactiveStreamsSubscribableChannel` honoring a back-pressure downstream. +Starting with version 5.5, when `maxMessagesPerPoll == 0`, the source is not called at all, and `flatMapMany()` is completed immediately via a `Mono.empty()` result until the `maxMessagesPerPoll` is changed to non-zero value at a later time, e.g. via a Control Bus. This way, any `MessageSource` implementation can be turned into a reactive hot source. See <<./polling-consumer.adoc#polling-consumer,Polling Consumer>> for more information. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 55ce4f9328..80174c3dd6 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -25,6 +25,10 @@ The `spring.integration.channels.error.requireSubscribers=true` global property The `spring.integration.channels.error.ignoreFailures=true` global property is added to indicate that the global default `errorChannel` must ignore (or not) dispatching errors and pass the message to the next handler. See <<./configuration.adoc#global-properties,Global Properties>> for more information. +An `AbstractPollingEndpoint` (source polling channel adapter and polling consumer) treats `maxMessagesPerPoll == 0` as to skip calling the source. +It can be changed to different value later on, e.g. via a Control Bus. +See <<./endpoint.adoc#endpoint-pollingconsumer,Polling Consumer>> for more information. + [[x5.5-amqp]] ==== AMQP Changes