From fb35a736d6fc159f9bd34794ff9fdb97b05462d9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 21 Oct 2020 13:48:31 -0400 Subject: [PATCH] Get correct next execution date from the trigger With the `time-source` in Spring Cloud Stream we noticed that first two items have the same (or close) value when the `time-supplier` is used with the reactive output channel. Turns out the `contactMap()` with `prefetch == 1` asks two upstream elements immediately not waiting for a completion for the fist one. This way we were not able to update `triggerContext` for the proper `lastCompletionTime` * Change `AbstractPollingEndpoint` to `prefetch = 0` for the "fair" upstream request * Verify the behavior with the new `ReactiveInboundChannelAdapterTests.testTimeSupplierConsistency()` * Mark `ReactiveInboundChannelAdapterTests` as `@LongRunningTest` since it is now pretty long waiting for all the 3 dates to verify * Fix `InboundChannelAdapterAnnotationPostProcessor` to properly register `MethodInvokingMessageSource` when we have more than one `Supplier` with the `@InboundChannelAdapter` in the same configuration class --- ...ChannelAdapterAnnotationPostProcessor.java | 5 +- .../endpoint/AbstractPollingEndpoint.java | 2 +- .../ReactiveInboundChannelAdapterTests.java | 46 ++++++++++++++++++- 3 files changed, 49 insertions(+), 4 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java index ddabc5f874..831d59ee09 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java @@ -85,10 +85,11 @@ public class InboundChannelAdapterAnnotationPostProcessor extends return adapter; } - private MessageSource createMessageSource(Object beanArg, String beanName, Method methodArg) { + private MessageSource createMessageSource(Object beanArg, String beanNameArg, Method methodArg) { MessageSource messageSource = null; Object bean = beanArg; Method method = methodArg; + String beanName = beanNameArg; if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) { Object target = resolveTargetBeanFromMethodWithBeanAnnotation(method); Class targetClass = target.getClass(); @@ -106,10 +107,12 @@ public class InboundChannelAdapterAnnotationPostProcessor extends else if (target instanceof Supplier) { method = ClassUtils.SUPPLIER_GET_METHOD; bean = target; + beanName += '.' + methodArg.getName(); } else if (ClassUtils.KOTLIN_FUNCTION_0_INVOKE_METHOD != null) { method = ClassUtils.KOTLIN_FUNCTION_0_INVOKE_METHOD; bean = target; + beanName += '.' + methodArg.getName(); } } if (messageSource == null) { 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 6e89c8f1d8..7ec1b09a10 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 @@ -366,7 +366,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement .update(triggerContext.lastScheduledExecutionTime(), triggerContext.lastActualExecutionTime(), new Date()) - )), 1) + )), 0) .repeat(this::isRunning) .doOnSubscribe(subs -> this.subscription = subs); } 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 e2c42df751..fb6902e46b 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 @@ -16,7 +16,12 @@ package org.springframework.integration.endpoint; +import static org.assertj.core.api.Assertions.assertThat; + import java.time.Duration; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; @@ -31,6 +36,7 @@ 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; @@ -46,15 +52,16 @@ import reactor.test.StepVerifier; */ @SpringJUnitConfig @DirtiesContext +@LongRunningTest public class ReactiveInboundChannelAdapterTests { @Autowired - private FluxMessageChannel fluxMessageChannel; + private FluxMessageChannel fluxChannel; @Test public void testReactiveInboundChannelAdapter() { Flux testFlux = - Flux.from(this.fluxMessageChannel) + Flux.from(this.fluxChannel) .map(Message::getPayload) .cast(Integer.class); @@ -64,6 +71,30 @@ public class ReactiveInboundChannelAdapterTests { .verify(Duration.ofSeconds(10)); } + @Autowired + private FluxMessageChannel fluxChannel2; + + @Test + public void testTimeSupplierConsistency() { + Flux testFlux = + Flux.from(this.fluxChannel2) + .map(Message::getPayload) + .cast(Date.class) + .map(Date::getTime); + + List dates = new ArrayList<>(); + + StepVerifier.create(testFlux) + .consumeNextWith(dates::add) + .consumeNextWith(dates::add) + .consumeNextWith(dates::add) + .thenCancel() + .verify(Duration.ofSeconds(10)); + + assertThat(dates.get(1) - dates.get(0)).isGreaterThanOrEqualTo(1000); + assertThat(dates.get(2) - dates.get(1)).isGreaterThanOrEqualTo(1000); + } + @Configuration @EnableIntegration public static class Config { @@ -93,6 +124,17 @@ public class ReactiveInboundChannelAdapterTests { return new FluxMessageChannel(); } + @Bean + @InboundChannelAdapter(value = "fluxChannel2", poller = @Poller(fixedDelay = "1000")) + public Supplier timeSupplier() { + return Date::new; + } + + @Bean + public MessageChannel fluxChannel2() { + return new FluxMessageChannel(); + } + } }