Implement range

Author: daveray

language-adaptors/rxjava-clojure/src/main/clojure/rx/lang/clojure/core.clj

@@ -7,7 +7,8 @@
                             interleave interpose into iterate
                             keep keep-indexed
                             map mapcat map-indexed
-                            merge next nth partition-all reduce reductions
+                            merge next nth partition-all
+                            range reduce reductions
                             rest seq some sort sort-by split-with
                             take take-while throw])
   (:require [rx.lang.clojure.interop :as iop]
@@ -678,6 +679,30 @@
   ([n ^Observable xs] (.window xs (int n)))
   ([n step ^Observable xs] (.window xs (int n) (int step))))
 
+(defn range
+  "Returns an Observable nums from start (inclusive) to end
+  (exclusive), by step, where start defaults to 0, step to 1, and end
+  to infinity.
+
+  Note: this is not implemented on rx.Observable/range
+
+  See:
+    clojure.core/range
+  "
+  ([] (range 0 Double/POSITIVE_INFINITY 1))
+  ([end] (range 0 end 1))
+  ([start end] (range start end 1))
+  ([start end step]
+   (observable* (fn [s]
+                  (let [comp (if (pos? step) < >)]
+                    (loop [i start]
+                      (if-not (unsubscribed? s)
+                        (if (comp i end)
+                          (do
+                            (on-next s i)
+                            (recur (+ i step)))
+                          (on-completed s)))))))))
+
 (defn ^Observable reduce
   ([f ^Observable xs] (.reduce xs (iop/fn* f)))
   ([f val ^Observable xs] (.reduce xs val (iop/fn* f))))

language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/core_test.clj

@@ -498,6 +498,28 @@
        10 3
        15 30))
 
+(deftest test-range
+  (are [start end step] (= (range start end step)
+                           (->> (rx/range start end step) (b/into [])))
+       0   10  2
+       0 -100 -1
+       5 100   9)
+
+  (are [start end] (= (range start end)
+                      (->> (rx/range start end) (b/into [])))
+       0   10
+       0 -100
+       5 100)
+
+  (are [start] (= (->> (range start) (take 100))
+                  (->> (rx/range start) (rx/take 100) (b/into [])))
+       50
+       0
+       5
+       -20)
+  (is (= (->> (range) (take 500))
+         (->> (rx/range) (rx/take 500) (b/into [])))))
+
 (deftest test-reduce
   (is (= (reduce + 0 (range 4))
          (b/first (rx/reduce + 0 (rx/seq->o (range 4)))))))