2.x: fix flatMapIterable appearing to be empty when fused (#5256)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlattenIterable.java

@@ -411,7 +411,10 @@ public void clear() {
         @Override
         public boolean isEmpty() {
             Iterator<? extends R> it = current;
-            return (it != null && !it.hasNext()) || queue.isEmpty();
+            if (it == null) {
+                return queue.isEmpty();
+            }
+            return !it.hasNext();
         }
 
         @Nullable

src/test/java/io/reactivex/internal/operators/flowable/FlowableFlattenIterableTest.java

@@ -863,4 +863,55 @@ public void remove() {
 
         ts.assertResult(1);
     }
+
+    @Test
+    public void doubleShare() {
+        Iterable<Integer> it = Flowable.range(1, 300).blockingIterable();
+            Flowable.just(it, it)
+            .flatMapIterable(Functions.<Iterable<Integer>>identity())
+            .share()
+            .share()
+            .count()
+            .test()
+            .assertResult(600L);
+    }
+
+    @Test
+    public void multiShare() {
+        Iterable<Integer> it = Flowable.range(1, 300).blockingIterable();
+        for (int i = 0; i < 5; i++) {
+            Flowable<Integer> f = Flowable.just(it, it)
+            .flatMapIterable(Functions.<Iterable<Integer>>identity());
+
+            for (int j = 0; j < i; j++) {
+                f = f.share();
+            }
+
+            f
+            .count()
+            .test()
+            .withTag("Share: " + i)
+            .assertResult(600L);
+        }
+    }
+
+    @Test
+    public void multiShareHidden() {
+        Iterable<Integer> it = Flowable.range(1, 300).blockingIterable();
+        for (int i = 0; i < 5; i++) {
+            Flowable<Integer> f = Flowable.just(it, it)
+            .flatMapIterable(Functions.<Iterable<Integer>>identity())
+            .hide();
+
+            for (int j = 0; j < i; j++) {
+                f = f.share();
+            }
+
+            f
+            .count()
+            .test()
+            .withTag("Share: " + i)
+            .assertResult(600L);
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/flowable/FlowableToListTest.java

@@ -395,7 +395,7 @@ public void onNextCancelRace() {
         for (int i = 0; i < 1000; i++) {
             final PublishProcessor<Integer> pp = PublishProcessor.create();
             final TestObserver<List<Integer>> ts = pp.toList().test();
-            
+
             Runnable r1 = new Runnable() {
                 @Override
                 public void run() {
@@ -408,18 +408,17 @@ public void run() {
                     ts.cancel();
                 }
             };
-            
+
             TestHelper.race(r1, r2);
         }
-        
     }
 
     @Test
     public void onNextCancelRaceFlowable() {
         for (int i = 0; i < 1000; i++) {
             final PublishProcessor<Integer> pp = PublishProcessor.create();
             final TestSubscriber<List<Integer>> ts = pp.toList().toFlowable().test();
-            
+
             Runnable r1 = new Runnable() {
                 @Override
                 public void run() {
@@ -432,20 +431,20 @@ public void run() {
                     ts.cancel();
                 }
             };
-            
+
             TestHelper.race(r1, r2);
         }
-        
+
     }
 
     @Test
     public void onCompleteCancelRaceFlowable() {
         for (int i = 0; i < 1000; i++) {
             final PublishProcessor<Integer> pp = PublishProcessor.create();
             final TestSubscriber<List<Integer>> ts = pp.toList().toFlowable().test();
-            
+
             pp.onNext(1);
-            
+
             Runnable r1 = new Runnable() {
                 @Override
                 public void run() {
@@ -458,14 +457,13 @@ public void run() {
                     ts.cancel();
                 }
             };
-            
+
             TestHelper.race(r1, r2);
-            
+
             if (ts.valueCount() != 0) {
                 ts.assertValue(Arrays.asList(1))
                 .assertNoErrors();
             }
         }
-        
     }
 }