OnSubscribeRange

Author: benjchristensen

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

@@ -33,6 +33,7 @@
 import rx.observables.ConnectableObservable;
 import rx.observables.GroupedObservable;
 import rx.observers.SafeSubscriber;
+import rx.operators.OnSubscribeRange;
 import rx.operators.OperationAll;
 import rx.operators.OperationAmb;
 import rx.operators.OperationAny;
@@ -120,7 +121,6 @@
 import rx.subscriptions.Subscriptions;
 import rx.util.Exceptions;
 import rx.util.OnErrorNotImplementedException;
-import rx.util.Range;
 import rx.util.TimeInterval;
 import rx.util.Timestamped;
 import rx.util.functions.Action0;
@@ -2439,7 +2439,10 @@ 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) {
-        return from(Range.createWithCount(start, count));
+        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));
     }
 
     /**
@@ -2459,7 +2462,10 @@ 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) {
-        return from(Range.createWithCount(start, count), 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);
     }
 
     /**

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

@@ -0,0 +1,44 @@
+/**
+ * Copyright 2014 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.operators;
+
+import rx.Observable.OnSubscribe;
+import rx.Subscriber;
+
+/**
+ */
+public final class OnSubscribeRange implements OnSubscribe<Integer> {
+
+    private final int start;
+    private final int end;
+
+    public OnSubscribeRange(int start, int end) {
+        this.start = start;
+        this.end = end;
+    }
+
+    @Override
+    public void call(Subscriber<? super Integer> o) {
+        for (int i = start; i < end; i++) {
+            if (o.isUnsubscribed()) {
+                return;
+            }
+            o.onNext(i);
+        }
+        o.onCompleted();
+    }
+
+}

rxjava-core/src/main/java/rx/util/Range.java

@@ -0,0 +1,59 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.IntegerSumObserver;
+import rx.util.functions.Action0;
+
+public class OperatorRangePerformance extends AbstractPerformanceTester {
+
+    static int reps = Integer.MAX_VALUE / 8;
+
+    OperatorRangePerformance() {
+        super(reps);
+    }
+
+    public static void main(String args[]) {
+
+        final OperatorRangePerformance spt = new OperatorRangePerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    spt.timeRange();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * 
+     * -- 0.17
+     * 
+     * Run: 10 - 271,147,198 ops/sec
+     * Run: 11 - 274,821,481 ops/sec
+     * Run: 12 - 271,632,295 ops/sec
+     * Run: 13 - 277,876,014 ops/sec
+     * Run: 14 - 274,821,763 ops/sec
+     * 
+     * -- 0.16.1
+     * 
+     * Run: 10 - 222,104,280 ops/sec
+     * Run: 11 - 224,311,761 ops/sec
+     * Run: 12 - 222,999,339 ops/sec
+     * Run: 13 - 222,344,174 ops/sec
+     * Run: 14 - 225,247,983 ops/sec
+     * 
+     * @return
+     */
+    public long timeRange() {
+        IntegerSumObserver o = new IntegerSumObserver();
+        Observable.range(1, reps).subscribe(o);
+        return o.sum;
+    }
+
+}

rxjava-core/src/perf/java/rx/operators/OperatorRangePerformance.java

@@ -0,0 +1,40 @@
+/**
+ * Copyright 2014 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.operators;
+
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observer;
+
+public class OnSubscribeRangeTest {
+
+    @Test
+    public void testRangeStartAt2Count3() {
+        @SuppressWarnings("unchecked")
+        Observer<Integer> observer = mock(Observer.class);
+        Observable.range(2, 3).subscribe(observer);
+
+        verify(observer, times(1)).onNext(2);
+        verify(observer, times(1)).onNext(3);
+        verify(observer, times(1)).onNext(4);
+        verify(observer, never()).onNext(5);
+        verify(observer, never()).onError(org.mockito.Matchers.any(Throwable.class));
+        verify(observer, times(1)).onCompleted();
+    }
+}