Merge pull request #2770 from davidmoten/onBackpressureDrop-request-o…

…verflow OperatorOnBackpressureDrop request overflow check
ReactiveX · Feb 24, 2015 · 2605ede · 2605ede
2 parents 3a6ce5a + 9edfdac
commit 2605ede
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 3 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java b/src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java
@@ -43,7 +43,7 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
 
             @Override
             public void request(long n) {
-                requested.getAndAdd(n);
+                BackpressureUtils.getAndAddRequest(requested, n);
             }
 
         });

diff --git a/src/test/java/rx/internal/operators/OperatorOnBackpressureDropTest.java b/src/test/java/rx/internal/operators/OperatorOnBackpressureDropTest.java
@@ -17,6 +17,9 @@
 
 import static org.junit.Assert.assertEquals;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.junit.Test;
 
 import rx.Observable;
@@ -27,8 +30,6 @@
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
 
-import java.util.concurrent.CountDownLatch;
-
 public class OperatorOnBackpressureDropTest {
 
     @Test
@@ -87,6 +88,35 @@ public void onNext(Long t) {
         ts.assertNoErrors();
         assertEquals(0, ts.getOnNextEvents().get(0).intValue());
     }
+
+    @Test
+    public void testRequestOverflow() throws InterruptedException {
+        final AtomicInteger count = new AtomicInteger();
+        int n = 10;
+        range(n).onBackpressureDrop().subscribe(new Subscriber<Long>() {
+
+            @Override
+            public void onStart() {
+                request(10);
+            }
+
+            @Override
+            public void onCompleted() {
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                throw new RuntimeException(e);
+            }
+
+            @Override
+            public void onNext(Long t) {
+                count.incrementAndGet();
+                //cause overflow of requested if not handled properly in onBackpressureDrop operator
+                request(Long.MAX_VALUE-1);
+            }});
+        assertEquals(n, count.get());
+    }
 
     static final Observable<Long> infinite = Observable.create(new OnSubscribe<Long>() {
 
@@ -99,4 +129,22 @@ public void call(Subscriber<? super Long> s) {
         }
 
     });
+
+    private static final Observable<Long> range(final long n) {
+        return Observable.create(new OnSubscribe<Long>() {
+
+            @Override
+            public void call(Subscriber<? super Long> s) {
+                for (long i=0;i < n;i++) {
+                    if (s.isUnsubscribed()) {
+                        break;
+                    }
+                    s.onNext(i);
+                }
+                s.onCompleted();
+            }
+
+        });
+    }
+
 }