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

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) {

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);
 	}

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<Integer> 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<Long> testFlux =
+				Flux.from(this.fluxChannel2)
+						.map(Message::getPayload)
+						.cast(Date.class)
+				.map(Date::getTime);
+
+		List<Long> 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<Date> timeSupplier() {
+			return Date::new;
+		}
+
+		@Bean
+		public MessageChannel fluxChannel2() {
+			return new FluxMessageChannel();
+		}
+
 	}
 
 }