Test Unsubscribe

- also cleaned up some stuff I remembered after merging last commits

Author: benjchristensen

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

@@ -2439,10 +2439,13 @@ public final static <T> Observable<Observable<T>> parallelMerge(Observable<Obser
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229460.aspx">MSDN: Observable.Range</a>
      */
     public final static Observable<Integer> range(int start, int count) {
+        if (count < 1) {
+            throw new IllegalArgumentException("Count must be positive");
+        }
         if ((start + count) > Integer.MAX_VALUE) {
             throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
         }
-        return Observable.create(new OnSubscribeRange(start, start + count));
+        return Observable.create(new OnSubscribeRange(start, start + (count - 1)));
     }
 
     /**
@@ -2462,10 +2465,7 @@ public final static Observable<Integer> range(int start, int count) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211896.aspx">MSDN: Observable.Range</a>
      */
     public final static Observable<Integer> range(int start, int count, Scheduler scheduler) {
-        if ((start + count) > Integer.MAX_VALUE) {
-            throw new IllegalArgumentException("start + count can not exceed Integer.MAX_VALUE");
-        }
-        return Observable.create(new OnSubscribeRange(start, start + count)).subscribeOn(scheduler);
+        return range(start, count).subscribeOn(scheduler);
     }
 
     /**

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

@@ -19,6 +19,7 @@
 import rx.Subscriber;
 
 /**
+ * Emit ints from start to end inclusive.
  */
 public final class OnSubscribeRange implements OnSubscribe<Integer> {
 
@@ -32,7 +33,7 @@ public OnSubscribeRange(int start, int end) {
 
     @Override
     public void call(Subscriber<? super Integer> o) {
-        for (int i = start; i < end; i++) {
+        for (int i = start; i <= end; i++) {
             if (o.isUnsubscribed()) {
                 return;
             }

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

@@ -15,12 +15,16 @@
  */
 package rx.operators;
 
+import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Observer;
+import rx.util.functions.Action1;
 
 public class OnSubscribeRangeTest {
 
@@ -37,4 +41,27 @@ public void testRangeStartAt2Count3() {
         verify(observer, never()).onError(org.mockito.Matchers.any(Throwable.class));
         verify(observer, times(1)).onCompleted();
     }
+
+    @Test
+    public void testRangeUnsubscribe() {
+        @SuppressWarnings("unchecked")
+        Observer<Integer> observer = mock(Observer.class);
+        final AtomicInteger count = new AtomicInteger();
+        Observable.range(1, 1000).doOnNext(new Action1<Integer>() {
+
+            @Override
+            public void call(Integer t1) {
+                count.incrementAndGet();
+            }
+
+        }).take(3).subscribe(observer);
+
+        verify(observer, times(1)).onNext(1);
+        verify(observer, times(1)).onNext(2);
+        verify(observer, times(1)).onNext(3);
+        verify(observer, never()).onNext(4);
+        verify(observer, never()).onError(org.mockito.Matchers.any(Throwable.class));
+        verify(observer, times(1)).onCompleted();
+        assertEquals(3, count.get());
+    }
 }