Implemented multi-sequence mapcat

Author: daveray

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

@@ -597,16 +597,35 @@
   (Observable/zip ^Iterable observables
                   ^rx.functions.FuncN (iop/fnN* f)))
 
+(defn ^Observable mapcat*
+  "Same as multi-arg mapcat, but input is an Observable of Observables.
+
+  See:
+    mapcat
+    clojure.core/mapcat
+  "
+  [f ^Observable xs]
+  (->> xs
+       (map* f)
+       (concat*)))
+
 (defn ^Observable mapcat
   "Returns an observable which, for each value x in xs, calls (f x), which must
   return an Observable. The resulting observables are concatentated together
   into one observable.
 
+  If multiple Observables are given, the arguments to f are the first item from
+  each observable, then the second item, etc.
+
   See:
     clojure.core/mapcat
     rx.Observable/flatMap
   "
-  ([f ^Observable xs] (.flatMap xs (iop/fn* f))))
+  [f & xs]
+  (if (clojure.core/next xs)
+    (mapcat* f (seq->o xs))
+    ; use built-in flatMap for single-arg case
+    (.flatMap ^Observable (clojure.core/first xs) (iop/fn* f))))
 
 (defn map-indexed
   "Returns an observable that invokes (f index value) for each value of the input

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

@@ -423,11 +423,31 @@
       (catch java.io.FileNotFoundException e
         (is (= :b (-> e .getCause .getValue)))))))
 
+(deftest test-mapcat*
+  (let [f (fn [a b c d e]
+            [(+ a b) (+ c d) e])]
+    (is (= (->> (range 5)
+                (map (fn [_] (range 5)))
+                (apply mapcat f))
+           (->> (range 5)
+                (map (fn [_] (rx/seq->o (range 5))))
+                (rx/seq->o)
+                (rx/mapcat* (fn [& args] (rx/seq->o (apply f args))))
+                (b/into []))))))
+
 (deftest test-mapcat
   (let [f  (fn [v] [v (* v v)])
         xs (range 10)]
     (is (= (mapcat f xs)
-           (b/into [] (rx/mapcat (comp rx/seq->o f) (rx/seq->o xs)))))))
+           (b/into [] (rx/mapcat (comp rx/seq->o f) (rx/seq->o xs))))))
+
+  (let [f  (fn [a b] [a b (* a b)])
+        as (range 10)
+        bs (range 15)]
+    (is (= (mapcat f as bs)
+           (b/into [] (rx/mapcat (comp rx/seq->o f)
+                                 (rx/seq->o as)
+                                 (rx/seq->o bs)))))))
 
 (deftest test-next
   (let [in [:q :r :s :t :u]]