Update merge/merge-delay-error impls.

Split merge into merge/merge* for consistency with other functions that
can take one or more observables or observable of observables.

Author: daveray

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

@@ -288,39 +288,45 @@
   ([lock ^Observable xs]
   (.synchronize xs lock)))
 
-(defn ^Observable merge
-  "Observable.merge, renamed because merge means something else in Clojure
+(defn merge*
+  "Merge an Observable of Observables into a single Observable
+
+  If you want clojure.core/merge, it's just this:
 
-  os is one of:
+    (rx/reduce clojure.core/merge {} maps)
 
-    * An Iterable of Observables to merge
-    * An Observable<Observable<T>> to merge
+  See:
+    merge
+    merge-delay-error*
+    rx.Observable/merge
+  "
+  [^Observable xs]
+  (Observable/merge xs))
+
+(defn ^Observable merge
+  "Merge one or more Observables into a single observable.
 
   If you want clojure.core/merge, it's just this:
 
     (rx/reduce clojure.core/merge {} maps)
 
+  See:
+    merge*
+    merge-delay-error
+    rx.Observable/merge
   "
-  [os]
-  (cond
-    (instance? Iterable os)
-      (Observable/merge (Observable/from ^Iterable os))
-    (instance? Observable os)
-      (Observable/merge ^Observable os)
-    :else
-      (throw (IllegalArgumentException. (str "Don't know how to merge " (type os))))))
+  [& os]
+  (merge* (seq->o os)))
 
-(defn ^Observable merge-delay-error
-  "Observable.mergeDelayError"
-  [os]
-  (cond
-    (instance? java.util.List os)
-      (Observable/mergeDelayError ^java.util.List os)
-    (instance? Observable os)
-      (Observable/mergeDelayError ^Observable os)
-    :else
-      (throw (IllegalArgumentException. (str "Don't know how to merge " (type os))))))
+(defn ^Observable merge-delay-error*
+  "Same as merge*, but all values are emitted before errors are propagated"
+  [^Observable xs]
+  (Observable/mergeDelayError xs))
 
+(defn ^Observable merge-delay-error
+  "Same as merge, but all values are emitted before errors are propagated"
+  [& os]
+  (merge-delay-error* (seq->o os)))
 
 (defn cache
   "caches the observable value so that multiple subscribers don't re-evaluate it.

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

@@ -111,20 +111,40 @@
                 (rx/synchronize lock)
                 (b/into []))))))
 
-(deftest test-merge
-  (is (= [[1 3 5] [2 4 6]]
-         (let [r (b/into []
-                   (rx/merge [(f/future-generator f/default-runner [o]
-                                                   (doseq [x [1 3 5]]
-                                                     (Thread/sleep 10)
-                                                     (rx/on-next o x)))
-                              (f/future-generator f/default-runner [o]
-                                                   (doseq [x [2 4 6]]
-                                                     (Thread/sleep 10)
-                                                     (rx/on-next o x)))]))]
-           ; make sure each sequence maintained original order
-           [(keep #{1 3 5} r)
-            (keep #{2 4 6} r) ]))))
+(let [expected-result [[1 3 5] [2 4 6]]
+      sleepy-o        #(f/future-generator f/default-runner [o]
+                                           (doseq [x %]
+                                             (Thread/sleep 10)
+                                             (rx/on-next o x))) 
+      make-inputs (fn [] (mapv sleepy-o expected-result))
+      make-output (fn [r] [(keep #{1 3 5} r)
+                           (keep #{2 4 6} r)])]
+  (deftest test-merge*
+    (is (= expected-result
+           (->> (make-inputs)
+                (rx/seq->o)
+                (rx/merge*)
+                (b/into [])
+                (make-output)))))
+  (deftest test-merge
+    (is (= expected-result
+           (->> (make-inputs)
+                (apply rx/merge)
+                (b/into [])
+                (make-output)))))
+  (deftest test-merge-delay-error*
+    (is (= expected-result
+           (->> (make-inputs)
+                (rx/seq->o)
+                (rx/merge-delay-error*)
+                (b/into [])
+                (make-output)))))
+  (deftest test-merge-delay-error
+    (is (= expected-result
+           (->> (make-inputs)
+                (apply rx/merge-delay-error)
+                (b/into [])
+                (make-output))))))
 
 (deftest test-generator
   (testing "calls on-completed automatically"