add request overflow check for OnSubscribeFromIterable

Author: davidmoten

src/main/java/rx/Producer.java

@@ -23,6 +23,16 @@ public interface Producer {
     /**
      * Request a certain maximum number of items from this Producer. This is a way of requesting backpressure.
      * To disable backpressure, pass {@code Long.MAX_VALUE} to this method.
+     * <p>
+     * Requests are additive but if a sequence of requests totals more than {@code Long.MAX_VALUE} then 
+     * {@code Long.MAX_VALUE} requests will be actioned and the extras <i>may</i> be ignored. Arriving at 
+     * {@code Long.MAX_VALUE} by addition of requests cannot be assumed to disable backpressure. For example, 
+     * the code below may result in {@code Long.MAX_VALUE} requests being actioned only.
+     * 
+     * <pre>
+     * request(100);
+     * request(Long.MAX_VALUE-1);
+     * </pre> 
      *
      * @param n the maximum number of items you want this Producer to produce, or {@code Long.MAX_VALUE} if you
      *          want the Producer to produce items at its own pace

src/main/java/rx/Subscriber.java

@@ -89,7 +89,17 @@ public void onStart() {
      * Request a certain maximum number of emitted items from the Observable this Subscriber is subscribed to.
      * This is a way of requesting backpressure. To disable backpressure, pass {@code Long.MAX_VALUE} to this
      * method.
-     *
+     * <p>
+     * Requests are additive but if a sequence of requests totals more than {@code Long.MAX_VALUE} then 
+     * {@code Long.MAX_VALUE} requests will be actioned and the extras <i>may</i> be ignored. Arriving at 
+     * {@code Long.MAX_VALUE} by addition of requests cannot be assumed to disable backpressure. For example, 
+     * the code below may result in {@code Long.MAX_VALUE} requests being actioned only.
+     * 
+     * <pre>
+     * request(100);
+     * request(Long.MAX_VALUE-1);
+     * </pre>
+     * 
      * @param n the maximum number of items you want the Observable to emit to the Subscriber at this time, or
      *           {@code Long.MAX_VALUE} if you want the Observable to emit items at its own pace
      * @throws IllegalArgumentException

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

@@ -0,0 +1,78 @@
+/**
+ * Copyright 2015 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.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+/**
+ * Utility functions for use with backpressure.
+ *
+ */
+final class BackpressureUtils {
+
+    /**
+     * Adds {@code n} to {@code requested} field and returns the value prior to
+     * addition once the addition is successful (uses CAS semantics). If
+     * overflows then sets {@code requested} field to {@code 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;
+        }
+    }
+
+    /**
+     * Adds {@code n} to {@code requested} and returns the value prior to addition once the
+     * addition is successful (uses CAS semantics). If overflows then sets
+     * {@code requested} field to {@code 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(AtomicLong requested, long n) {
+        // add n to field but check for overflow
+        while (true) {
+            long current = requested.get();
+            long next = current + n;
+            // check for overflow
+            if (next < 0)
+                next = Long.MAX_VALUE;
+            if (requested.compareAndSet(current, next))
+                return current;
+        }
+    }
+}

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 = BackpressureUtils.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;
-                }
+                BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
                 if (ms.drainQueuesIfNeeded()) {
                     boolean sendComplete = false;
                     synchronized (ms) {

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));
+    }
+
 
 }