Implement interleave and interleave*

Author: daveray

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

@@ -4,7 +4,7 @@
                             empty every?
                             filter first future
                             group-by
-                            interpose into
+                            interleave interpose into
                             keep keep-indexed
                             map mapcat map-indexed
                             merge next nth partition reduce reductions
@@ -25,7 +25,7 @@
 
 (set! *warn-on-reflection* true)
 
-(declare concat map map-indexed reduce take take-while)
+(declare concat* concat map* map map-indexed reduce take take-while)
 
 (defn ^Func1 fn->predicate
   "Turn f into a predicate that returns true/false like Rx predicates should"
@@ -484,6 +484,33 @@
         (map (fn [^GroupedObservable go]
                (clojure.lang.MapEntry. (.getKey go) go))))))
 
+(defn interleave*
+  "Returns an Observable of the first item in each Observable emitted by observables, then
+  the second etc.
+
+  observables is an Observable of Observables
+
+  See:
+    interleave
+    clojure.core/interleave
+  "
+  [observables]
+  (->> (map* #(seq->o %&) observables)
+       (concat*)))
+
+(defn interleave
+  "Returns an Observable of the first item in each Observable, then the second etc.
+
+  Each argument is an individual Observable
+
+  See:
+    observable*
+    clojure.core/interleave
+  "
+  [o1 & observables]
+  (->> (apply map #(seq->o %&) o1 observables)
+       (concat*)))
+
 (defn interpose
   [sep xs]
   (let [op (operator* (fn [o]

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

@@ -284,8 +284,35 @@
                   (rx/mapcat (fn [[k vo :as me]]
                                (is (instance? clojure.lang.MapEntry me))
                                (rx/map #(vector k %) vo)))
-                  (b/into [])))))
-    ))
+                  (b/into [])))))))
+
+(deftest test-interleave
+  (are [inputs] (= (apply interleave inputs)
+                   (->> (apply rx/interleave (map rx/seq->o inputs))
+                        (b/into [])))
+       [[] []]
+       [[] [1]]
+       [(range 5) (range 10) (range 10) (range 3)]
+       [(range 50) (range 10)]
+       [(range 5) (range 10 60) (range 10) (range 50)])
+
+  ; one-arg case, not supported by clojure.core/interleave
+  (is (= (range 10)
+         (->> (rx/interleave (rx/seq->o (range 10)))
+              (b/into [])))))
+
+(deftest test-interleave*
+  (are [inputs] (= (apply interleave inputs)
+                   (->> (rx/interleave* (->> inputs
+                                             (map rx/seq->o)
+                                             (rx/seq->o)))
+                        (b/into [])))
+       [[] []]
+       [[] [1]]
+       [(range 5) (range 10) (range 10) (range 3)]
+       [(range 50) (range 10)]
+       [(range 5) (range 10 60) (range 10) (range 50)]))
+
 (deftest test-interpose
   (is (= (interpose \, [1 2 3])
          (b/into [] (rx/interpose \, (rx/seq->o [1 2 3]))))))