2.x: Fix blockingIterable hang when force-disposed (#6627)

ReactiveX · Aug 21, 2019 · a9df239 · a9df239
1 parent 8db3569
commit a9df239
Show file tree

Hide file tree

Showing 10 changed files with 85 additions and 14 deletions.
diff --git a/src/main/java/io/reactivex/Flowable.java b/src/main/java/io/reactivex/Flowable.java
@@ -89,7 +89,7 @@
  * </code></pre>
  * <p>
  * The Reactive Streams specification is relatively strict when defining interactions between {@code Publisher}s and {@code Subscriber}s, so much so
- * that there is a significant performance penalty due certain timing requirements and the need to prepare for invalid 
+ * that there is a significant performance penalty due certain timing requirements and the need to prepare for invalid
  * request amounts via {@link Subscription#request(long)}.
  * Therefore, RxJava has introduced the {@link FlowableSubscriber} interface that indicates the consumer can be driven with relaxed rules.
  * All RxJava operators are implemented with these relaxed rules in mind.
@@ -112,7 +112,7 @@
  *
  *         // could be some blocking operation
  *         Thread.sleep(1000);
- *         
+ *
  *         // the consumer might have cancelled the flow
  *         if (emitter.isCancelled() {
  *             return;
@@ -138,7 +138,7 @@
  * RxJava reactive sources, such as {@code Flowable}, are generally synchronous and sequential in nature. In the ReactiveX design, the location (thread)
  * where operators run is <i>orthogonal</i> to when the operators can work with data. This means that asynchrony and parallelism
  * has to be explicitly expressed via operators such as {@link #subscribeOn(Scheduler)}, {@link #observeOn(Scheduler)} and {@link #parallel()}. In general,
- * operators featuring a {@link Scheduler} parameter are introducing this type of asynchrony into the flow. 
+ * operators featuring a {@link Scheduler} parameter are introducing this type of asynchrony into the flow.
  * <p>
  * For more information see the <a href="http://reactivex.io/documentation/Publisher.html">ReactiveX
  * documentation</a>.

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java b/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java
@@ -62,7 +62,7 @@ static final class BlockingFlowableIterator<T>
         long produced;
 
         volatile boolean done;
-        Throwable error;
+        volatile Throwable error;
 
         BlockingFlowableIterator(int batchSize) {
             this.queue = new SpscArrayQueue<T>(batchSize);
@@ -75,6 +75,13 @@ static final class BlockingFlowableIterator<T>
         @Override
         public boolean hasNext() {
             for (;;) {
+                if (isDisposed()) {
+                    Throwable e = error;
+                    if (e != null) {
+                        throw ExceptionHelper.wrapOrThrow(e);
+                    }
+                    return false;
+                }
                 boolean d = done;
                 boolean empty = queue.isEmpty();
                 if (d) {
@@ -90,7 +97,7 @@ public boolean hasNext() {
                     BlockingHelper.verifyNonBlocking();
                     lock.lock();
                     try {
-                        while (!done && queue.isEmpty()) {
+                        while (!done && queue.isEmpty() && !isDisposed()) {
                             condition.await();
                         }
                     } catch (InterruptedException ex) {
@@ -175,6 +182,7 @@ public void remove() {
         @Override
         public void dispose() {
             SubscriptionHelper.cancel(this);
+            signalConsumer();
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java
@@ -419,7 +419,7 @@ public void onComplete() {
         public void cancel() {
             SubscriptionHelper.cancel(this);
         }
-        
+
         public void request(long n) {
             if (fusionMode != QueueSubscription.SYNC) {
                 get().request(n);

diff --git a/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableIterable.java b/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableIterable.java
@@ -53,7 +53,7 @@ static final class BlockingObservableIterator<T>
         final Condition condition;
 
         volatile boolean done;
-        Throwable error;
+        volatile Throwable error;
 
         BlockingObservableIterator(int batchSize) {
             this.queue = new SpscLinkedArrayQueue<T>(batchSize);
@@ -64,6 +64,13 @@ static final class BlockingObservableIterator<T>
         @Override
         public boolean hasNext() {
             for (;;) {
+                if (isDisposed()) {
+                    Throwable e = error;
+                    if (e != null) {
+                        throw ExceptionHelper.wrapOrThrow(e);
+                    }
+                    return false;
+                }
                 boolean d = done;
                 boolean empty = queue.isEmpty();
                 if (d) {
@@ -80,7 +87,7 @@ public boolean hasNext() {
                         BlockingHelper.verifyNonBlocking();
                         lock.lock();
                         try {
-                            while (!done && queue.isEmpty()) {
+                            while (!done && queue.isEmpty() && !isDisposed()) {
                                 condition.await();
                             }
                         } finally {
@@ -146,6 +153,7 @@ public void remove() {
         @Override
         public void dispose() {
             DisposableHelper.dispose(this);
+            signalConsumer();
         }
 
         @Override

diff --git a/src/test/java/io/reactivex/internal/operators/flowable/BlockingFlowableToIteratorTest.java b/src/test/java/io/reactivex/internal/operators/flowable/BlockingFlowableToIteratorTest.java
@@ -16,14 +16,18 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.*;
 import org.reactivestreams.*;
 
 import io.reactivex.Flowable;
+import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.*;
 import io.reactivex.internal.operators.flowable.BlockingFlowableIterable.BlockingFlowableIterator;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
+import io.reactivex.processors.PublishProcessor;
+import io.reactivex.schedulers.Schedulers;
 
 public class BlockingFlowableToIteratorTest {
 
@@ -185,4 +189,28 @@ protected void subscribeActual(Subscriber<? super Integer> s) {
 
         it.next();
     }
+
+    @Test(expected = NoSuchElementException.class)
+    public void disposedIteratorHasNextReturns() {
+        Iterator<Integer> it = PublishProcessor.<Integer>create()
+                .blockingIterable().iterator();
+        ((Disposable)it).dispose();
+        assertFalse(it.hasNext());
+        it.next();
+    }
+
+    @Test
+    public void asyncDisposeUnblocks() {
+        final Iterator<Integer> it = PublishProcessor.<Integer>create()
+                .blockingIterable().iterator();
+
+        Schedulers.single().scheduleDirect(new Runnable() {
+            @Override
+            public void run() {
+                ((Disposable)it).dispose();
+            }
+        }, 1, TimeUnit.SECONDS);
+
+        assertFalse(it.hasNext());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableBufferTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableBufferTest.java
@@ -2770,6 +2770,7 @@ public void timedSizeBufferAlreadyCleared() {
     }
 
     @Test
+    @SuppressWarnings("unchecked")
     public void bufferExactFailingSupplier() {
         Flowable.empty()
                 .buffer(1, TimeUnit.SECONDS, Schedulers.computation(), 10, new Callable<List<Object>>() {

diff --git a/src/test/java/io/reactivex/internal/operators/observable/BlockingObservableNextTest.java b/src/test/java/io/reactivex/internal/operators/observable/BlockingObservableNextTest.java
@@ -28,7 +28,6 @@
 import io.reactivex.exceptions.TestException;
 import io.reactivex.internal.operators.observable.BlockingObservableNext.NextObserver;
 import io.reactivex.plugins.RxJavaPlugins;
-import io.reactivex.processors.BehaviorProcessor;
 import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subjects.*;
 
@@ -332,9 +331,9 @@ public void testSingleSourceManyIterators() throws InterruptedException {
 
     @Test
     public void testSynchronousNext() {
-        assertEquals(1, BehaviorProcessor.createDefault(1).take(1).blockingSingle().intValue());
-        assertEquals(2, BehaviorProcessor.createDefault(2).blockingIterable().iterator().next().intValue());
-        assertEquals(3, BehaviorProcessor.createDefault(3).blockingNext().iterator().next().intValue());
+        assertEquals(1, BehaviorSubject.createDefault(1).take(1).blockingSingle().intValue());
+        assertEquals(2, BehaviorSubject.createDefault(2).blockingIterable().iterator().next().intValue());
+        assertEquals(3, BehaviorSubject.createDefault(3).blockingNext().iterator().next().intValue());
     }
 
     @Test

diff --git a/...est/java/io/reactivex/internal/operators/observable/BlockingObservableToIteratorTest.java b/...est/java/io/reactivex/internal/operators/observable/BlockingObservableToIteratorTest.java
@@ -16,15 +16,18 @@
 import static org.junit.Assert.*;
 
 import java.util.*;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.*;
 
 import io.reactivex.Observable;
 import io.reactivex.ObservableSource;
 import io.reactivex.Observer;
-import io.reactivex.disposables.Disposables;
+import io.reactivex.disposables.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.internal.operators.observable.BlockingObservableIterable.BlockingObservableIterator;
+import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subjects.PublishSubject;
 
 public class BlockingObservableToIteratorTest {
 
@@ -119,4 +122,28 @@ public void remove() {
         BlockingObservableIterator<Integer> it = new BlockingObservableIterator<Integer>(128);
         it.remove();
     }
+
+    @Test(expected = NoSuchElementException.class)
+    public void disposedIteratorHasNextReturns() {
+        Iterator<Integer> it = PublishSubject.<Integer>create()
+                .blockingIterable().iterator();
+        ((Disposable)it).dispose();
+        assertFalse(it.hasNext());
+        it.next();
+    }
+
+    @Test
+    public void asyncDisposeUnblocks() {
+        final Iterator<Integer> it = PublishSubject.<Integer>create()
+                .blockingIterable().iterator();
+
+        Schedulers.single().scheduleDirect(new Runnable() {
+            @Override
+            public void run() {
+                ((Disposable)it).dispose();
+            }
+        }, 1, TimeUnit.SECONDS);
+
+        assertFalse(it.hasNext());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableBufferTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableBufferTest.java
@@ -2137,6 +2137,7 @@ public ObservableSource<List<Object>> apply(Observable<Object> o)
     }
 
     @Test
+    @SuppressWarnings("unchecked")
     public void bufferExactFailingSupplier() {
         Observable.empty()
                 .buffer(1, TimeUnit.SECONDS, Schedulers.computation(), 10, new Callable<List<Object>>() {

diff --git a/src/test/java/io/reactivex/schedulers/TrampolineSchedulerTest.java b/src/test/java/io/reactivex/schedulers/TrampolineSchedulerTest.java
@@ -31,7 +31,6 @@
 import org.reactivestreams.Subscriber;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 
 public class TrampolineSchedulerTest extends AbstractSchedulerTests {