add request overflow check for OnSubscribeFromIterable

Author: davidmoten

src/main/java/rx/internal/operators/OnSubscribeFromIterable.java

@@ -80,7 +80,7 @@ public void request(long n) {
                 }
             } else if(n > 0) {
                 // backpressure is requested
-                long _c = REQUESTED_UPDATER.getAndAdd(this, n);
+                long _c = Util.getAndAddRequest(REQUESTED_UPDATER, this, n);
                 if (_c == 0) {
                     while (true) {
                         /*

src/main/java/rx/internal/operators/OperatorMerge.java

@@ -545,16 +545,7 @@ public void request(long n) {
             if (n == Long.MAX_VALUE) {
                 requested = Long.MAX_VALUE;
             } else {
-                // add n to requested but check for overflow
-                while (true) {
-                    long current = REQUESTED.get(this);
-                    long next = current + n;
-                    //check for overflow
-                    if (next < 0)
-                        next = Long.MAX_VALUE;
-                    if (REQUESTED.compareAndSet(this, current, next))
-                        break;
-                }
+                Util.getAndAddRequest(REQUESTED, this, n);
                 if (ms.drainQueuesIfNeeded()) {
                     boolean sendComplete = false;
                     synchronized (ms) {

src/main/java/rx/internal/operators/Util.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package rx.internal.operators;
+
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+/**
+ * Utility functions for use with Operators.
+ *
+ */
+final class Util {
+
+    /**
+     * Adds n to requested field and returns the value prior to addition once the addition 
+     * is successful (uses CAS semantics). If overflows then sets requested field to Long.MAX_VALUE.
+     * 
+     * @param requested atomic field updater for a request count
+     * @param object contains the field updated by the updater
+     * @param n the number of requests to add to the requested count
+     * @return requested value just prior to successful addition
+     */
+    static <T> long getAndAddRequest(AtomicLongFieldUpdater<T> requested, T object, long n) {
+        // add n to field but check for overflow
+        while (true) {
+            long current = requested.get(object);
+            long next = current + n;
+            //check for overflow
+            if (next < 0)
+                next = Long.MAX_VALUE;
+            if (requested.compareAndSet(object, current, next))
+                return current;
+        }
+    }
+}

src/test/java/rx/internal/operators/OnSubscribeFromIterableTest.java

@@ -15,6 +15,7 @@
  */
 package rx.internal.operators;
 
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -24,14 +25,18 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import org.mockito.Mockito;
 
 import rx.Observable;
 import rx.Observer;
+import rx.Subscriber;
 import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
+import rx.schedulers.Schedulers;
 
 public class OnSubscribeFromIterableTest {
 
@@ -157,5 +162,36 @@ public void testSubscribeMultipleTimes() {
         o.call(ts);
         ts.assertReceivedOnNext(Arrays.asList(1, 2, 3));
     }
+    
+    @Test
+    public void testFromIterableRequestOverflow() throws InterruptedException {
+        Observable<Integer> o = Observable.from(Arrays.asList(1,2,3,4));
+        final int expectedCount = 4;
+        final CountDownLatch latch = new CountDownLatch(expectedCount);
+        o.subscribeOn(Schedulers.computation()).subscribe(new Subscriber<Integer>() {
+            
+            @Override
+            public void onStart() {
+                request(2);
+            }
+
+            @Override
+            public void onCompleted() {
+                //ignore
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                throw new RuntimeException(e);
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                latch.countDown();
+                request(Long.MAX_VALUE-1);
+            }});
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+    }
+
 
 }