Merge pull request #2854 from akarnokd/AbstractOnSubscribeRequestFix

Fixes wrong request accounting in AbstractOnSubscribe

Author: akarnokd

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

@@ -22,8 +22,11 @@
  * Utility functions for use with backpressure.
  *
  */
-final class BackpressureUtils {
-
+public final class BackpressureUtils {
+    /** Utility class, no instances. */
+    private BackpressureUtils() {
+        throw new IllegalStateException("No instances!");
+    }
     /**
      * Adds {@code n} to {@code requested} field and returns the value prior to
      * addition once the addition is successful (uses CAS semantics). If
@@ -37,16 +40,18 @@ final class BackpressureUtils {
      *            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) {
+    public 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)
+            if (next < 0) {
                 next = Long.MAX_VALUE;
-            if (requested.compareAndSet(object, current, next))
+            }
+            if (requested.compareAndSet(object, current, next)) {
                 return current;
+            }
         }
     }
 
@@ -63,16 +68,18 @@ static <T> long getAndAddRequest(AtomicLongFieldUpdater<T> requested, T object,
      *            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) {
+    public static 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)
+            if (next < 0) {
                 next = Long.MAX_VALUE;
-            if (requested.compareAndSet(current, next))
+            }
+            if (requested.compareAndSet(current, next)) {
                 return current;
+            }
         }
     }
 }

src/main/java/rx/observables/AbstractOnSubscribe.java

@@ -24,6 +24,7 @@
 import rx.annotations.Experimental;
 import rx.exceptions.CompositeException;
 import rx.functions.*;
+import rx.internal.operators.BackpressureUtils;
 
 /**
  * Abstract base class for the {@link OnSubscribe} interface that helps you build Observable sources one
@@ -332,14 +333,15 @@ private SubscriptionProducer(SubscriptionState<T, S> state) {
         }
         @Override
         public void request(long n) {
-            if (n == Long.MAX_VALUE) {
-                for (; !state.subscriber.isUnsubscribed(); ) {
-                    if (!doNext()) {
-                        break;
+            if (n > 0 && BackpressureUtils.getAndAddRequest(state.requestCount, n) == 0) {
+                if (n == Long.MAX_VALUE) {
+                    // fast-path
+                    for (; !state.subscriber.isUnsubscribed(); ) {
+                        if (!doNext()) {
+                            break;
+                        }
                     }
-                }
-            } else 
-            if (n > 0 && state.requestCount.getAndAdd(n) == 0) {
+                } else 
                 if (!state.subscriber.isUnsubscribed()) {
                     do {
                         if (!doNext()) {

src/test/java/rx/observables/AbstractOnSubscribeTest.java

@@ -16,12 +16,13 @@
 
 package rx.observables;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -503,4 +504,37 @@ public void testMissingEmission() {
         verify(o, never()).onNext(any(Object.class));
         verify(o).onError(any(IllegalStateException.class));
     }
+    
+    @Test
+    public void testCanRequestInOnNext() {
+        AbstractOnSubscribe<Integer, Void> aos = new AbstractOnSubscribe<Integer, Void>() {
+            @Override
+            protected void next(SubscriptionState<Integer, Void> state) {
+                state.onNext(1);
+                state.onCompleted();
+            }
+        };
+        final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
+        aos.toObservable().subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                exception.set(e);
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                request(1);
+            }
+        });
+        if (exception.get()!=null) {
+            exception.get().printStackTrace();
+        }
+        assertNull(exception.get());
+    }
 }