Merge pull request #1721 from abersnaze/onBackpressure-request-max-0.20

Bug in the onBackpressure operators
ReactiveX · Oct 4, 2014 · ab85374 · ab85374
2 parents 0e80124 + b36d833
commit ab85374
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 2 deletions.
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorOnBackpressureBuffer.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorOnBackpressureBuffer.java
@@ -48,6 +48,10 @@ public void request(long n) {
         // don't pass through subscriber as we are async and doing queue draining
         // a parent being unsubscribed should not affect the children
         Subscriber<T> parent = new Subscriber<T>() {
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
 
             @Override
             public void onCompleted() {

diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java
@@ -36,6 +36,10 @@ public void request(long n) {
 
         });
         return new Subscriber<T>(child) {
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
 
             @Override
             public void onCompleted() {

diff --git a/rxjava-core/src/test/java/rx/internal/operators/OperatorOnBackpressureDropTest.java b/rxjava-core/src/test/java/rx/internal/operators/OperatorOnBackpressureDropTest.java
@@ -17,17 +17,18 @@
 
 import static org.junit.Assert.assertEquals;
 
-import java.util.concurrent.CountDownLatch;
-
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
 import rx.Subscriber;
+import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
 
+import java.util.concurrent.CountDownLatch;
+
 public class OperatorOnBackpressureDropTest {
 
     @Test
@@ -42,6 +43,13 @@ public void testNoBackpressureSupport() {
         ts.assertNoErrors();
     }
 
+    @Test(timeout = 500)
+    public void testWithObserveOn() throws InterruptedException {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        Observable.range(0, RxRingBuffer.SIZE * 10).onBackpressureDrop().onBackpressureDrop().observeOn(Schedulers.io()).subscribe(ts);
+        ts.awaitTerminalEvent();
+    }
+
     @Test(timeout = 500)
     public void testFixBackpressureWithBuffer() throws InterruptedException {
         final CountDownLatch l1 = new CountDownLatch(100);