From ee736255eca72c819d8b4d4f9790012d4f94cb7c Mon Sep 17 00:00:00 2001 From: Stephane Maldini Date: Mon, 9 Jan 2017 12:53:08 +0000 Subject: [PATCH] polish test for #342 --- .../publisher/FluxSubscribeOnValueTest.java | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/test/java/reactor/core/publisher/FluxSubscribeOnValueTest.java b/src/test/java/reactor/core/publisher/FluxSubscribeOnValueTest.java index a350fe4735..13213b127d 100644 --- a/src/test/java/reactor/core/publisher/FluxSubscribeOnValueTest.java +++ b/src/test/java/reactor/core/publisher/FluxSubscribeOnValueTest.java @@ -15,13 +15,21 @@ */ package reactor.core.publisher; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + import org.junit.Test; import reactor.core.Exceptions; +import reactor.core.Fuseable; import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; +import static org.junit.Assert.assertTrue; + public class FluxSubscribeOnValueTest { + ConcurrentMap execs = new ConcurrentHashMap<>(); + @Test public void testSubscribeOnValueFusion() { @@ -30,13 +38,25 @@ public void testSubscribeOnValueFusion() { .subscribeOn(Schedulers.parallel()) .log() .map(this::slow))) + .expectFusion(Fuseable.ASYNC, Fuseable.NONE) .expectNextCount(100) .verifyComplete(); + int minExec = 2; + + for (Integer counted : execs.values()) { + assertTrue("Thread used less than " + minExec + " " + "times", + counted >= minExec); + } + } int slow(int slow){ try { + execs.computeIfAbsent(Thread.currentThread() + .hashCode(), i -> 0); + execs.compute(Thread.currentThread() + .hashCode(), (k, v) -> v + 1); Thread.sleep(10); return slow; }