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

Fix maxMessagesPerPoll for SourcePollingChAdapter (#8747)

Browse Source 
* Fix maxMessagesPerPoll for SourcePollingChAdapter

The `AbstractMethodAnnotationPostProcessor` does not check
for `PollerMetadata.MAX_MESSAGES_UNBOUNDED` before setting
`maxMessagesPerPoll` into a `SourcePollingChannelAdapter`
which in this case must be `1`

Also fix `SourcePollingChannelAdapterFactoryBean` to not mutate
the provided `PollerMetadata` (which might be global default)
with a new `maxMessagesPerPoll`

**Cherry-pick to `6.1.x` & `6.0.x`**

* * Fix `this.` prefix in `SourcePollingChannelAdapterFactoryBean`
This commit is contained in:
Artem Bilan
2023-10-04 11:49:28 -04:00
committed by Gary Russell
parent f0768cdf4a
commit 8c146d562d
3 changed files with 18 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-integration-core/src/main/java/org/springframework/integration/config/AbstractMethodAnnotationPostProcessor.java
View File

@@ -688,7 +688,14 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
pollingEndpoint.setTaskExecutor(pollerMetadata.getTaskExecutor());
pollingEndpoint.setTrigger(pollerMetadata.getTrigger());
pollingEndpoint.setAdviceChain(pollerMetadata.getAdviceChain());
pollingEndpoint.setMaxMessagesPerPoll(pollerMetadata.getMaxMessagesPerPoll());
long maxMessagesPerPoll = pollerMetadata.getMaxMessagesPerPoll();
if (maxMessagesPerPoll == PollerMetadata.MAX_MESSAGES_UNBOUNDED &&
pollingEndpoint instanceof SourcePollingChannelAdapter) {
// the default is 1 since a source might return
// a non-null and non-interruptible value every time it is invoked
maxMessagesPerPoll = 1;
}
pollingEndpoint.setMaxMessagesPerPoll(maxMessagesPerPoll);
pollingEndpoint.setErrorHandler(pollerMetadata.getErrorHandler());
if (pollingEndpoint instanceof PollingConsumer) {
((PollingConsumer) pollingEndpoint).setReceiveTimeout(pollerMetadata.getReceiveTimeout());

7
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -181,12 +181,13 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
Assert.notNull(this.pollerMetadata, () -> "No poller has been defined for channel-adapter '"
+ this.beanName + "', and no default poller is available within the context.");
}
if (this.pollerMetadata.getMaxMessagesPerPoll() == Integer.MIN_VALUE) {
long maxMessagesPerPoll = this.pollerMetadata.getMaxMessagesPerPoll();
if (maxMessagesPerPoll == PollerMetadata.MAX_MESSAGES_UNBOUNDED) {
// the default is 1 since a source might return
// a non-null and non-interruptible value every time it is invoked
this.pollerMetadata.setMaxMessagesPerPoll(1);
maxMessagesPerPoll = 1;
}
spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
spca.setMaxMessagesPerPoll(maxMessagesPerPoll);
if (this.sendTimeout != null) {
spca.setSendTimeout(this.sendTimeout);
}

8
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -104,6 +104,7 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.expression.SpelPropertyAccessorRegistrar;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -416,12 +417,15 @@ public class EnableIntegrationTests {
assertThat(this.counterChannel.receive(10)).isNull();
SmartLifecycle countSA = this.context.getBean("annotationTestService.count.inboundChannelAdapter",
SmartLifecycle.class);
SourcePollingChannelAdapter countSA =
this.context.getBean("annotationTestService.count.inboundChannelAdapter",
SourcePollingChannelAdapter.class);
assertThat(countSA.isAutoStartup()).isFalse();
assertThat(countSA.getPhase()).isEqualTo(23);
countSA.start();
assertThat(countSA.getMaxMessagesPerPoll()).isEqualTo(1);
for (int i = 0; i < 10; i++) {
Message<?> message = this.counterChannel.receive(10_000);
assertThat(message).isNotNull();