2.x: make parallel() a fusion-async-boundary

ReactiveX · Oct 17, 2017 · 9549ed5 · 9549ed5
1 parent 1ad6647
commit 9549ed5
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 1 deletion.
diff --git a/src/main/java/io/reactivex/internal/operators/parallel/ParallelFromPublisher.java b/src/main/java/io/reactivex/internal/operators/parallel/ParallelFromPublisher.java
@@ -116,7 +116,7 @@ public void onSubscribe(Subscription s) {
                     @SuppressWarnings("unchecked")
                     QueueSubscription<T> qs = (QueueSubscription<T>) s;
 
-                    int m = qs.requestFusion(QueueSubscription.ANY);
+                    int m = qs.requestFusion(QueueSubscription.ANY | QueueSubscription.BOUNDARY);
 
                     if (m == QueueSubscription.SYNC) {
                         sourceMode = m;

diff --git a/src/test/java/io/reactivex/parallel/ParallelFlowableTest.java b/src/test/java/io/reactivex/parallel/ParallelFlowableTest.java
@@ -1356,4 +1356,45 @@ public void flatMapSubscriberCount() {
     public void fromArraySubscriberCount() {
         ParallelFlowableTest.checkSubscriberCount(ParallelFlowable.fromArray(new Publisher[] { Flowable.just(1) }));
     }
+
+    @Test
+    public void boundaryConfinement() {
+        final Set<String> between = new HashSet<String>();
+        final ConcurrentHashMap<String, String> processing = new ConcurrentHashMap<String, String>();
+
+        Flowable.range(1, 10)
+        .observeOn(Schedulers.single(), false, 1)
+        .doOnNext(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer v) throws Exception {
+                between.add(Thread.currentThread().getName());
+            }
+        })
+        .parallel(2, 1)
+        .runOn(Schedulers.computation(), 1)
+        .map(new Function<Integer, Object>() {
+            @Override
+            public Object apply(Integer v) throws Exception {
+                processing.putIfAbsent(Thread.currentThread().getName(), "");
+                return v;
+            }
+        })
+        .sequential()
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertSubscribed()
+        .assertValueSet(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
+        .assertComplete()
+        .assertNoErrors()
+        ;
+
+        assertEquals(between.toString(), 1, between.size());
+        assertTrue(between.toString(), between.iterator().next().contains("RxSingleScheduler"));
+
+        Map<String, String> map = processing; // AnimalSniffer: CHM.keySet() in Java 8 returns KeySetView
+
+        for (String e : map.keySet()) {
+            assertTrue(map.toString(), e.contains("RxComputationThreadPool"));
+        }
+    }
 }