From 7b2d8a4ba73fe6d3b587f93622213ffa64d96709 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 30 Jan 2024 17:03:24 -0500 Subject: [PATCH] Fix race condition in the ReactiveInboundChannelAdapterTests The `ReactiveInboundChannelAdapterTests.testTimeSupplierConsistency()` subscribes a bit later than `@InboundChannelAdapter` might have started to emit messages. * Add `autoStartup = "false"` to the `timeEndpoint` and start in the test only when `StepVerifier` is subscribed * Add `stop()` for both endpoints in the test configuration to avoid memory pollution with not processed messages in between tests --- .../ReactiveInboundChannelAdapterTests.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) 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 e7cedee628..a961dfb8c7 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 @@ -61,6 +61,10 @@ public class ReactiveInboundChannelAdapterTests { @Qualifier("counterEndpoint") private AbstractPollingEndpoint abstractPollingEndpoint; + @Autowired + @Qualifier("timeEndpoint") + private AbstractPollingEndpoint timeEndpoint; + @Test public void testReactiveInboundChannelAdapter() { Flux testFlux = @@ -71,10 +75,12 @@ public class ReactiveInboundChannelAdapterTests { StepVerifier.create(testFlux) .expectSubscription() .expectNoEvent(Duration.ofSeconds(1)) - .then(() -> abstractPollingEndpoint.setMaxMessagesPerPoll(-1)) + .then(() -> this.abstractPollingEndpoint.setMaxMessagesPerPoll(-1)) .expectNext(2, 4, 6, 8, 10, 12, 14, 16) .thenCancel() .verify(Duration.ofSeconds(10)); + + this.abstractPollingEndpoint.stop(); } @Autowired @@ -91,6 +97,8 @@ public class ReactiveInboundChannelAdapterTests { List dates = new ArrayList<>(); StepVerifier.create(testFlux) + .expectSubscription() + .then(() -> this.timeEndpoint.start()) .consumeNextWith(dates::add) .consumeNextWith(dates::add) .consumeNextWith(dates::add) @@ -99,6 +107,8 @@ public class ReactiveInboundChannelAdapterTests { assertThat(dates.get(1) - dates.get(0)).isGreaterThanOrEqualTo(1000); assertThat(dates.get(2) - dates.get(1)).isGreaterThanOrEqualTo(1000); + + this.timeEndpoint.stop(); } @Configuration @@ -132,7 +142,8 @@ public class ReactiveInboundChannelAdapterTests { } @Bean - @InboundChannelAdapter(value = "fluxChannel2", poller = @Poller(fixedDelay = "1000")) + @InboundChannelAdapter(value = "fluxChannel2", autoStartup = "false", poller = @Poller(fixedDelay = "1000")) + @EndpointId("timeEndpoint") public Supplier timeSupplier() { return Date::new; }