Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix maxMessagesPerPoll for reactive poll endpoint

Browse Source 
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`**
This commit is contained in:
Artem Bilan
2021-02-01 12:36:48 -05:00
committed by Gary Russell
parent e752c70fec
commit 98fbf62db2
3 changed files with 36 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -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<Message<?>> pollingTask;
private volatile Flux<Message<?>> 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);
}

9
spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java
View File

@@ -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.<Supplier<String>>argThat(logMessage ->
logMessage.get().contains("Poll interrupted - during stop()?")));
}
@Test

6
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java
View File

@@ -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<Integer> counterMessageSupplier() {
return () -> {
int i = counter().incrementAndGet();