Merge pull request #1443 from benjchristensen/request-long-max-value

Infinite request with Long.MAX_VALUE instead of -1

Author: benjchristensen

rxjava-core/src/main/java/rx/Subscriber.java

@@ -126,9 +126,9 @@ public final void setProducer(Producer producer) {
         if (setProducer) {
             op.setProducer(p);
         } else {
-            // we execute the request with whatever has been requested (or -1)
+            // we execute the request with whatever has been requested (or Long.MAX_VALUE)
             if (toRequest == Long.MIN_VALUE) {
-                p.request(-1);
+                p.request(Long.MAX_VALUE);
             } else {
                 p.request(toRequest);
             }

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

@@ -62,7 +62,7 @@ private IterableProducer(Subscriber<? super T> o, Iterator<? extends T> it) {
 
         @Override
         public void request(long n) {
-            if (n < 0) {
+            if (n == Long.MAX_VALUE) {
                 // fast-path without backpressure
                 while (it.hasNext()) {
                     if (o.isUnsubscribed()) {
@@ -71,7 +71,7 @@ public void request(long n) {
                     o.onNext(it.next());
                 }
                 o.onCompleted();
-            } else {
+            } else if(n > 0) {
                 // backpressure is requested
                 long _c = REQUESTED_UPDATER.getAndAdd(this, n);
                 if (_c == 0) {

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

@@ -56,7 +56,7 @@ private RangeProducer(Subscriber<? super Integer> o, int start, int end) {
 
         @Override
         public void request(long n) {
-            if (n < 0) {
+            if (n == Long.MAX_VALUE) {
                 // fast-path without backpressure
                 for (long i = index; i <= end; i++) {
                     if (o.isUnsubscribed()) {

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

@@ -391,8 +391,8 @@ public MergeProducer(MergeSubscriber<T> ms) {
 
         @Override
         public void request(long n) {
-            if (n < 0) {
-                requested = -1;
+            if (n == Long.MAX_VALUE) {
+                requested = Long.MAX_VALUE;
             } else {
                 REQUESTED.getAndAdd(this, n);
                 ms.drainQueuesIfNeeded();

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

@@ -67,17 +67,17 @@ protected Producer onSetProducer(final Producer producer) {
 
                     @Override
                     public void request(long n) {
-                        // add the skip num to the requested amount, since we'll skip everything and then emit to the buffer downstream
-                        if (n > 0) {
-                            producer.request(n + (toSkip - skipped));
-                        } else {
+                        if (n == Long.MAX_VALUE) {
                             // infinite so leave it alone
                             producer.request(n);
+                        } else if (n > 0) {
+                            // add the skip num to the requested amount, since we'll skip everything and then emit to the buffer downstream
+                            producer.request(n + (toSkip - skipped));
                         }
                     }
                 };
             }
-            
+
         };
     }
 }

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

@@ -52,7 +52,7 @@ public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
             @Override
             public void onStart() {
                 // we do this to break the chain of the child subscriber being passed through
-                request(-1);
+                request(Long.MAX_VALUE);
             }
 
             @Override
@@ -109,8 +109,8 @@ void startEmitting() {
         @Override
         public void request(long n) {
             long _c = 0;
-            if (n < 0) {
-                requested = -1;
+            if (n == Long.MAX_VALUE) {
+                requested = Long.MAX_VALUE;
             } else {
                 _c = REQUESTED_UPDATER.getAndAdd(this, n);
             }

rxjava-core/src/test/java/rx/SubscriberTest.java

@@ -64,7 +64,7 @@ public void request(long n) {
             }
 
         });
-        assertEquals(-1, r.get());
+        assertEquals(Long.MAX_VALUE, r.get());
     }
 
     @Test
@@ -154,8 +154,8 @@ public void request(long n) {
             }
 
         });
-        // this will be -1 because it is decoupled and nothing requsted on the Operator subscriber
-        assertEquals(-1, r.get());
+        // this will be Long.MAX_VALUE because it is decoupled and nothing requsted on the Operator subscriber
+        assertEquals(Long.MAX_VALUE, r.get());
     }
 
     @Test

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

@@ -143,7 +143,7 @@ public void testNoBackpressure() {
         OnSubscribeFromIterable<Integer> o = new OnSubscribeFromIterable<Integer>(Arrays.asList(1, 2, 3, 4, 5));
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
         ts.assertReceivedOnNext(Collections.<Integer> emptyList());
-        ts.request(-1); // infinite
+        ts.request(Long.MAX_VALUE); // infinite
         o.call(ts);
         ts.assertReceivedOnNext(Arrays.asList(1, 2, 3, 4, 5));
         ts.assertTerminalEvent();

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

@@ -126,7 +126,7 @@ public void testNoBackpressure() {
         OnSubscribeRange o = new OnSubscribeRange(1, list.size());
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
         ts.assertReceivedOnNext(Collections.<Integer> emptyList());
-        ts.request(-1); // infinite
+        ts.request(Long.MAX_VALUE); // infinite
         o.call(ts);
         ts.assertReceivedOnNext(list);
         ts.assertTerminalEvent();