rename Util to BackpressureUtils, add AtomicLong overload to BackpressureUtil, add javadoc to Producer and Subscriber

Author: davidmoten

src/main/java/rx/Producer.java

@@ -23,6 +23,10 @@ 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 the total requests overflows {@code Long.MAX_VALUE} then the total requests 
+     * is set to {@code Long.MAX_VALUE}. Arriving at {@code Long.MAX_VALUE} by addition of requests cannot be 
+     * assumed to disable backpressure. 
      *
      * @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,11 @@ 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 the total requests overflows {@code Long.MAX_VALUE} then the total requests 
+     * is set to {@code Long.MAX_VALUE}. Arriving at {@code Long.MAX_VALUE} by addition of requests cannot be 
+     * assumed to disable backpressure.
+     * 
      * @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 = Util.getAndAddRequest(REQUESTED_UPDATER, this, n);
+                long _c = BackpressureUtils.getAndAddRequest(REQUESTED_UPDATER, this, n);
                 if (_c == 0) {
                     while (true) {
                         /*

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

@@ -545,7 +545,7 @@ public void request(long n) {
             if (n == Long.MAX_VALUE) {
                 requested = Long.MAX_VALUE;
             } else {
-                Util.getAndAddRequest(REQUESTED, this, n);
+                BackpressureUtils.getAndAddRequest(REQUESTED, this, n);
                 if (ms.drainQueuesIfNeeded()) {
                     boolean sendComplete = false;
                     synchronized (ms) {