Fix issue ReactiveX#1522

zsxwing · Jul 28, 2014 · 212af0a · 212af0a
1 parent 4f1b09c
commit 212af0a
Show file tree

Hide file tree

Showing 2 changed files with 141 additions and 12 deletions.
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorTakeLast.java b/rxjava-core/src/main/java/rx/internal/operators/OperatorTakeLast.java
@@ -108,11 +108,21 @@ void startEmitting() {
 
         @Override
         public void request(long n) {
-            long _c = 0;
+            long _c;
             if (n == Long.MAX_VALUE) {
-                requested = Long.MAX_VALUE;
+                _c = REQUESTED_UPDATER.getAndSet(this, Long.MAX_VALUE);
             } else {
-                _c = REQUESTED_UPDATER.getAndAdd(this, n);
+                for (;;) {
+                    _c = requested;
+                    if (_c == Long.MAX_VALUE) {
+                        // If `requested` is Long.MAX_VALUE, `c+n` will be overflow.
+                        // Therefore, always check before setting to `c+n`
+                        return;
+                    }
+                    if (REQUESTED_UPDATER.compareAndSet(this, _c, _c + n)) {
+                        break;
+                    }
+                }
             }
             if (!emittingStarted) {
                 // we haven't started yet, so record what was requested and return
@@ -122,16 +132,20 @@ public void request(long n) {
         }
 
         void emit(long previousRequested) {
-            if (requested < 0) {
+            if (requested == Long.MAX_VALUE) {
                 // fast-path without backpressure
-                try {
-                    for (Object value : deque) {
-                        notification.accept(subscriber, value);
+                if (previousRequested == 0) {
+                    try {
+                        for (Object value : deque) {
+                            notification.accept(subscriber, value);
+                        }
+                    } catch (Throwable e) {
+                        subscriber.onError(e);
+                    } finally {
+                        deque.clear();
                     }
-                } catch (Throwable e) {
-                    subscriber.onError(e);
-                } finally {
-                    deque.clear();
+                } else {
+                    // backpressure path will handle Long.MAX_VALUE and emit the rest events.
                 }
             } else {
                 // backpressure is requested
@@ -160,7 +174,6 @@ void emit(long previousRequested) {
                             // we're done emitting the number requested so return
                             return;
                         }
-
                     }
                 }
             }

diff --git a/rxjava-core/src/test/java/rx/internal/operators/OperatorTakeLastTest.java b/rxjava-core/src/test/java/rx/internal/operators/OperatorTakeLastTest.java
@@ -30,7 +30,9 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.Subscriber;
 import rx.functions.Func1;
+import rx.functions.Functions;
 import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
@@ -148,4 +150,118 @@ public Integer call(Integer i) {
         };
     }
 
+    @Test
+    public void testIssue1522() {
+        // https://github.com/Netflix/RxJava/issues/1522
+        assertEquals(0, Observable
+                .empty()
+                .count()
+                .filter(Functions.alwaysFalse())
+                .toList()
+                .toBlocking().single().size());
+    }
+
+    @Test
+    public void testIgnoreRequest1() {
+        // If `takeLast` does not ignore `request` properly, StackOverflowError will be thrown.
+        Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(Integer integer) {
+                request(Long.MAX_VALUE);
+            }
+        });
+    }
+
+    @Test
+    public void testIgnoreRequest2() {
+        // If `takeLast` does not ignore `request` properly, StackOverflowError will be thrown.
+        Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(1);
+            }
+
+            @Override
+            public void onCompleted() {
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(Integer integer) {
+                request(1);
+            }
+        });
+    }
+
+    @Test(timeout = 30000)
+    public void testIgnoreRequest3() {
+        // If `takeLast` does not ignore `request` properly, it will enter an infinite loop.
+        Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(1);
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(Integer integer) {
+                request(Long.MAX_VALUE);
+            }
+        });
+    }
+
+
+    @Test
+    public void testIgnoreRequest4() {
+        // If `takeLast` does not ignore `request` properly, StackOverflowError will be thrown.
+        Observable.range(0, 100000).takeLast(100000).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE);
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(Integer integer) {
+                request(1);
+            }
+        });
+    }
 }