Handle overflow

Author: zsxwing

rxjava-core/src/main/java/rx/internal/operators/OperatorBufferWithSize.java

@@ -19,7 +19,6 @@
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Observable;
 import rx.Observable.Operator;
@@ -73,9 +72,17 @@ public Subscriber<? super T> call(final Subscriber<? super List<T>> child) {
                 @Override
                 public void setProducer(final Producer producer) {
                     child.setProducer(new Producer() {
+
+                        private volatile boolean infinite = false;
+
                         @Override
                         public void request(long n) {
-                            if (n == Long.MAX_VALUE) {
+                            if (infinite) {
+                                return;
+                            }
+                            if (n >= Long.MAX_VALUE / count) {
+                                // n == Long.MAX_VALUE or n * count >= Long.MAX_VALUE
+                                infinite = true;
                                 producer.request(Long.MAX_VALUE);
                             } else {
                                 producer.request(n * count);
@@ -127,21 +134,42 @@ public void onCompleted() {
             public void setProducer(final Producer producer) {
                 child.setProducer(new Producer() {
 
-                    private final AtomicBoolean firstRequest = new AtomicBoolean(false);
+                    private volatile boolean firstRequest = true;
+                    private volatile boolean infinite = false;
+
+                    private void requestInfinite() {
+                        infinite = true;
+                        producer.request(Long.MAX_VALUE);
+                    }
 
                     @Override
                     public void request(long n) {
+                        if (infinite) {
+                            return;
+                        }
                         if (n == Long.MAX_VALUE) {
-                            producer.request(Long.MAX_VALUE);
+                            requestInfinite();
+                            return;
                         } else {
-                            if (firstRequest.compareAndSet(false, true)) {
+                            if (firstRequest) {
+                                firstRequest = false;
+                                if (n - 1 >= (Long.MAX_VALUE - count) / skip) {
+                                    // count + skip * (n - 1) >= Long.MAX_VALUE
+                                    requestInfinite();
+                                    return;
+                                }
                                 // count = 5, skip = 2, n = 3
                                 // * * * * *
                                 //     * * * * *
                                 //         * * * * *
                                 // request = 5 + 2 * ( 3 - 1)
                                 producer.request(count + skip * (n - 1));
                             } else {
+                                if (n >= Long.MAX_VALUE / skip) {
+                                    // skip * n >= Long.MAX_VALUE
+                                    requestInfinite();
+                                    return;
+                                }
                                 // count = 5, skip = 2, n = 3
                                 // (* * *) * *
                                 // (    *) * * * *

rxjava-core/src/test/java/rx/internal/operators/OperatorBufferTest.java

@@ -869,7 +869,6 @@ public void request(long n) {
         assertEquals(6, requested.get());
     }
 
-
     @Test
     public void testProducerRequestThroughBufferWithSize4() {
         TestSubscriber<List<Integer>> ts = new TestSubscriber<List<Integer>>();
@@ -891,4 +890,95 @@ public void request(long n) {
         }).buffer(5, 2).subscribe(ts);
         assertEquals(Long.MAX_VALUE, requested.get());
     }
+
+
+    @Test
+    public void testProducerRequestOverflowThroughBufferWithSize1() {
+        TestSubscriber<List<Integer>> ts = new TestSubscriber<List<Integer>>();
+        ts.requestMore(Long.MAX_VALUE / 2);
+        final AtomicLong requested = new AtomicLong();
+        Observable.create(new Observable.OnSubscribe<Integer>() {
+
+            @Override
+            public void call(Subscriber<? super Integer> s) {
+                s.setProducer(new Producer() {
+
+                    @Override
+                    public void request(long n) {
+                        requested.set(n);
+                    }
+
+                });
+            }
+
+        }).buffer(3, 3).subscribe(ts);
+        assertEquals(Long.MAX_VALUE, requested.get());
+    }
+
+    @Test
+    public void testProducerRequestOverflowThroughBufferWithSize2() {
+        TestSubscriber<List<Integer>> ts = new TestSubscriber<List<Integer>>();
+        ts.requestMore(Long.MAX_VALUE / 2);
+        final AtomicLong requested = new AtomicLong();
+        Observable.create(new Observable.OnSubscribe<Integer>() {
+
+            @Override
+            public void call(Subscriber<? super Integer> s) {
+                s.setProducer(new Producer() {
+
+                    @Override
+                    public void request(long n) {
+                        requested.set(n);
+                    }
+
+                });
+            }
+
+        }).buffer(3, 2).subscribe(ts);
+        assertEquals(Long.MAX_VALUE, requested.get());
+    }
+
+    @Test
+    public void testProducerRequestOverflowThroughBufferWithSize3() {
+        final AtomicLong requested = new AtomicLong();
+        Observable.create(new Observable.OnSubscribe<Integer>() {
+
+            @Override
+            public void call(final Subscriber<? super Integer> s) {
+                s.setProducer(new Producer() {
+
+                    @Override
+                    public void request(long n) {
+                        requested.set(n);
+                        s.onNext(1);
+                        s.onNext(2);
+                        s.onNext(3);
+                    }
+
+                });
+            }
+
+        }).buffer(3, 2).subscribe(new Subscriber<List<Integer>>() {
+
+            @Override
+            public void onStart() {
+                request(Long.MAX_VALUE / 2 - 4);
+            }
+
+            @Override
+            public void onCompleted() {
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(List<Integer> t) {
+                request(Long.MAX_VALUE / 2);
+            }
+
+        });
+        assertEquals(Long.MAX_VALUE, requested.get());
+    }
 }