From f7a6d69c0acd6d8652222a6846ac9d4c31df7ece Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 16 Dec 2016 12:48:44 -0500 Subject: [PATCH] Wait for latch from the `doOnRequest()` https://build.spring.io/browse/INT-MASTER-473/ There is a race condition when we perform `onSubscribeCall()`, but the actual subscription isn't happened yet, therefore a new data for the `EmitterProcessor` is dropped because there is no subscriber yet to drain the buffer --- .../dsl/reactivestreams/ReactiveStreamsTests.java | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) 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 02187b1899..4a7f37e583 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 @@ -37,6 +37,7 @@ import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; import org.reactivestreams.Publisher; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.Lifecycle; @@ -110,15 +111,15 @@ public class ReactiveStreamsTests { Flux.from(this.pollablePublisher) .take(6) .filter(m -> m.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)) + .log("org.springframework.integration.flux2") .doOnNext(p -> latch.countDown()) .subscribe(); CountDownLatch secondSubscriberLatch = new CountDownLatch(1); Future> future = - Executors.newSingleThreadExecutor().submit(() -> { - Thread.sleep(100); - return Flux.just("11,12,13") + Executors.newSingleThreadExecutor().submit(() -> + Flux.just("11,12,13") .map(v -> v.split(",")) .flatMapIterable(Arrays::asList) .map(Integer::parseInt) @@ -128,9 +129,8 @@ public class ReactiveStreamsTests { .map(Message::getPayload) .log("org.springframework.integration.flux") .collectList() - .doOnSubscribe(s -> secondSubscriberLatch.countDown()) - .block(Duration.ofSeconds(10)); - } + .doOnRequest(s -> secondSubscriberLatch.countDown()) + .block(Duration.ofSeconds(10)) ); assertTrue(secondSubscriberLatch.await(10, TimeUnit.SECONDS));