Merge pull request #2548 from davidmoten/request-check

Subscriber.request should throw exception if negative request made

Author: akarnokd

src/main/java/rx/Subscriber.java

@@ -92,8 +92,13 @@ public void onStart() {
      *
      * @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
+     *             if {@code n} is negative
      */
     protected final void request(long n) {
+        if (n < 0) {
+            throw new IllegalArgumentException("number requested cannot be negative: " + n);
+        } 
         Producer shouldRequest = null;
         synchronized (this) {
             if (p != null) {

src/test/java/rx/SubscriberTest.java

@@ -16,9 +16,13 @@
 package rx;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 
@@ -419,4 +423,35 @@ public void onNext(Integer t) {
 
         assertEquals(1, c.get());
     }
+    
+    @Test
+    public void testNegativeRequestThrowsIllegalArgumentException() throws InterruptedException {
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
+        Observable.just(1,2,3,4).subscribe(new Subscriber<Integer>() {
+
+            @Override
+            public void onStart() {
+                request(1);
+            }
+            
+            @Override
+            public void onCompleted() {
+                
+            }
+
+            @Override
+            public void onError(Throwable e) {
+               exception.set(e);
+               latch.countDown();
+            }
+
+            @Override
+            public void onNext(Integer t) {
+                request(-1);
+                request(1);
+            }});
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        assertTrue(exception.get() instanceof IllegalArgumentException);
+    }
 }