From 98fbf62db24039205cd33aaeb38d0a7239eee01a Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 1 Feb 2021 12:36:48 -0500 Subject: [PATCH] Fix maxMessagesPerPoll for reactive poll endpoint The current `Flux.take()` doesn't allow an arg `< 0` treating it as an unbound request. * Change `take()` to `limitRequest()` according strict `MessageSource.receive()` producing expectations * Treat `maxMessagesPerPoll < 0` as a `Long.MAX_VALUE` for unbound requests; `0` is treated in the `limitRequest()` as "no more requests - cancel" * Revise `AbstractPollingEndpoint` for `LogAccessor` usage * Add `AbstractPollingEndpoint` class JavaDocs * Fix tests according `AbstractPollingEndpoint` changes **Cherry-pick to `5.4.x`** --- .../endpoint/AbstractPollingEndpoint.java | 52 ++++++++++--------- ...PollingChannelAdapterFactoryBeanTests.java | 9 ++-- .../ReactiveInboundChannelAdapterTests.java | 6 +-- 3 files changed, 36 insertions(+), 31 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 d05ecab98f..d516d8c8f0 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 @@ -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. @@ -24,6 +24,7 @@ import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledFuture; +import java.util.function.Consumer; import java.util.stream.Collectors; import org.aopalliance.aop.Advice; @@ -48,6 +49,7 @@ import org.springframework.integration.util.ErrorHandlingTaskExecutor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.scheduling.support.SimpleTriggerContext; @@ -65,6 +67,13 @@ import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; /** + * An {@link AbstractEndpoint} extension for Polling Consumer pattern basics. + * The standard polling logic is based on a periodic task scheduling according the provided + * {@link Trigger}. + * When this endpoint is treated as {@link #isReactive()}, a polling logic is turned into a + * {@link Flux#generate(Consumer)} and {@link Mono#delay(Duration)} combination based on the + * {@link SimpleTriggerContext} state. + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell @@ -90,8 +99,6 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private Trigger trigger = new PeriodicTrigger(DEFAULT_POLLING_PERIOD); - private long maxMessagesPerPoll = -1; - private ErrorHandler errorHandler; private boolean errorHandlerIsDefault; @@ -100,6 +107,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private TransactionSynchronizationFactory transactionSynchronizationFactory; + private volatile long maxMessagesPerPoll = -1; + private volatile Callable> pollingTask; private volatile Flux> pollingFlux; @@ -204,8 +213,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement } this.appliedAdvices.clear(); this.appliedAdvices.addAll(chain); - if (!(isSyncExecutor()) && logger.isWarnEnabled()) { - logger.warn(getComponentName() + ": A task executor is supplied and " + chain.size() + if (!(isSyncExecutor())) { + logger.warn(() -> getComponentName() + ": A task executor is supplied and " + chain.size() + "ReceiveMessageAdvice(s) is/are provided. If an advice mutates the source, such " + "mutations are not thread safe and could cause unexpected results, especially with " + "high frequency pollers. Consider using a downstream ExecutorChannel instead of " @@ -261,8 +270,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement try { super.onInit(); } - catch (Exception e) { - throw new BeanInitializationException("Cannot initialize: " + this, e); + catch (Exception ex) { + throw new BeanInitializationException("Cannot initialize: " + this, ex); } } @@ -280,11 +289,9 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.pollingFlux = createFluxGenerator(); } else { - Assert.state(getTaskScheduler() != null, "unable to start polling, no taskScheduler available"); - - this.runningTask = - getTaskScheduler() - .schedule(createPoller(), this.trigger); + TaskScheduler taskScheduler = getTaskScheduler(); + Assert.state(taskScheduler != null, "unable to start polling, no taskScheduler available"); + this.runningTask = taskScheduler.schedule(createPoller(), this.trigger); } } @@ -360,7 +367,10 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement fluxSink.complete(); } }) - .take(this.maxMessagesPerPoll) + .limitRequest( + this.maxMessagesPerPoll < 0 + ? Long.MAX_VALUE + : this.maxMessagesPerPoll) .subscribeOn(Schedulers.fromExecutor(this.taskExecutor)) .doOnComplete(() -> triggerContext @@ -407,22 +417,18 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement try { message = receiveMessage(); } - catch (Exception e) { + catch (Exception ex) { if (Thread.interrupted()) { - if (logger.isDebugEnabled()) { - logger.debug("Poll interrupted - during stop()? : " + e.getMessage()); - } + logger.debug(() -> "Poll interrupted - during stop()? : " + ex.getMessage()); return null; } else { - ReflectionUtils.rethrowRuntimeException(e); + ReflectionUtils.rethrowRuntimeException(ex); } } if (message == null) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Received no Message during the poll, returning 'false'"); - } + this.logger.debug("Received no Message during the poll, returning 'false'"); return null; } else { @@ -433,9 +439,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement } private void messageReceived(IntegrationResourceHolder holder, Message message) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Poll resulted in Message: " + message); - } + this.logger.debug(() -> "Poll resulted in Message: " + message); if (holder != null) { holder.setMessage(message); } 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 52ce5afd71..654f52ec6a 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 @@ -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. @@ -18,7 +18,6 @@ package org.springframework.integration.config; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.contains; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; @@ -32,12 +31,14 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentMatchers; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; @@ -188,7 +189,9 @@ public class SourcePollingChannelAdapterFactoryBeanTests { taskScheduler.shutdown(); verifyNoInteractions(errorHandlerLogger); - verify(adapterLogger).debug(contains("Poll interrupted - during stop()?")); + verify(adapterLogger) + .debug(ArgumentMatchers.>argThat(logMessage -> + logMessage.get().contains("Poll interrupted - during stop()?"))); } @Test 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 fb6902e46b..571289bc0f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-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. @@ -36,7 +36,6 @@ import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Poller; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.test.condition.LongRunningTest; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; @@ -52,7 +51,6 @@ import reactor.test.StepVerifier; */ @SpringJUnitConfig @DirtiesContext -@LongRunningTest public class ReactiveInboundChannelAdapterTests { @Autowired @@ -111,7 +109,7 @@ public class ReactiveInboundChannelAdapterTests { @Bean @InboundChannelAdapter(value = "fluxChannel", - poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "3", taskExecutor = "taskExecutor")) + poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "-1", taskExecutor = "taskExecutor")) public Supplier counterMessageSupplier() { return () -> { int i = counter().incrementAndGet();