From 5df0ef641698b502a6bda15885d1bb675f303fe0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sat, 19 Nov 2016 12:46:07 -0500 Subject: [PATCH] Fix Race in ReactiveStreamsTests Messages sent/published before the async task has fully set up, thus it does not receive the messages. --- .../dsl/reactivestreams/ReactiveStreamsTests.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java index 46b009c368..21a32be707 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java @@ -115,6 +115,8 @@ public class ReactiveStreamsTests { .take(6) .subscribe(); + final CountDownLatch asyncSetupLatch = new CountDownLatch(1); + Future> future = Executors.newSingleThreadExecutor().submit(() -> Flux.just("11,12,13") @@ -126,7 +128,9 @@ public class ReactiveStreamsTests { .map(Message::getPayload) .take(7) .collectList() - .block(Duration.ofSeconds(10))); + .block(getDuration(asyncSetupLatch))); + + assertTrue(asyncSetupLatch.await(10, TimeUnit.SECONDS)); this.inputChannel.send(new GenericMessage<>("6,7,8,9,10")); @@ -136,6 +140,11 @@ public class ReactiveStreamsTests { assertEquals(7, integers.size()); } + private Duration getDuration(CountDownLatch asyncSetupLatch) { + asyncSetupLatch.countDown(); + return Duration.ofSeconds(10); + } + @Configuration @EnableIntegration public static class ContextConfiguration {