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

Fix race condition in the ReactiveInboundChannelAdapterTests

Browse Source 
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
This commit is contained in:
Artem Bilan
2024-01-30 17:03:24 -05:00
parent d3924db5a1
commit 7b2d8a4ba7
1 changed files with 13 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -61,6 +61,10 @@ public class ReactiveInboundChannelAdapterTests {
@Qualifier("counterEndpoint")
private AbstractPollingEndpoint abstractPollingEndpoint;
@Autowired
@Qualifier("timeEndpoint")
private AbstractPollingEndpoint timeEndpoint;
@Test
public void testReactiveInboundChannelAdapter() {
Flux<Integer> 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<Long> 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<Date> timeSupplier() {
return Date::new;
}