Got rid of base. It was weird.

Author: daveray

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

@@ -1,7 +1,6 @@
 (ns rx.lang.clojure.chunk
   (:refer-clojure :exclude [chunk])
-  (:require [rx.lang.clojure.core :as rx]
-            [rx.lang.clojure.base :as rx-base]))
+  (:require [rx.lang.clojure.core :as rx]))
 
 (def ^:private -ns- *ns*)
 (set! *warn-on-reflection* true)

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

@@ -7,7 +7,6 @@
                             rest seq some sort sort-by split-with
                             take take-while throw])
   (:require [rx.lang.clojure.interop :as iop]
-            [rx.lang.clojure.base :as base]
             [rx.lang.clojure.graph :as graph]
             [rx.lang.clojure.realized :as realized])
   (:import [rx Observable Observer Subscriber Subscription Observable$Operator Observable$OnSubscribe]
@@ -156,6 +155,81 @@
 
 ;################################################################################
 
+(defn wrap-on-completed
+  "Wrap handler with code that automaticaly calls rx.Observable.onCompleted."
+  [handler]
+  (fn [^Observer observer]
+    (handler observer)
+    (.onCompleted observer)))
+
+(defn wrap-on-error
+  "Wrap handler with code that automaticaly calls (on-error) if an exception is thrown"
+  [handler]
+  (fn [^Observer observer]
+    (try
+      (handler observer)
+      (catch Throwable e
+        (.onError observer e)))))
+
+(defn ^Observable merge
+  "Observable.merge, renamed because merge means something else in Clojure
+
+  os is one of:
+
+    * An Iterable of Observables to merge
+    * An Observable<Observable<T>> to merge
+
+  If you want clojure.core/merge, it's just this:
+
+    (rx/reduce clojure.core/merge {} maps)
+
+  "
+  [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))))))
+
+(defn ^Observable merge-delay-error
+  "Observable.mergeDelayError, renamed because merge means something else in Clojure"
+  [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 zip
+  "Observable.zip. You want map."
+  ([f ^Observable a ^Observable b] (Observable/zip a b (iop/fn* f)))
+  ([f ^Observable a ^Observable b ^Observable c] (Observable/zip a b c (iop/fn* f)))
+  ([f ^Observable a ^Observable b ^Observable c ^Observable d] (Observable/zip a b c d (iop/fn* f)))
+  ([f a b c d & more]
+    ; recurse on more and then pull everything together with 4 parameter version
+   (zip (fn [a b c more-value]
+          (apply f a b c more-value))
+        a
+        b
+        c
+        (apply zip vector d more))))
+
+(defmacro zip-let
+  [bindings & body]
+  (let [pairs  (clojure.core/partition 2 bindings)
+        names  (clojure.core/mapv clojure.core/first pairs)
+        values (clojure.core/map second pairs)]
+    `(zip (fn ~names ~@body) ~@values)))
+;################################################################################
+
+
+
+
+
 (defn ^Observable never [] (Observable/never))
 (defn ^Observable empty [] (Observable/empty))
 
@@ -288,7 +362,7 @@
   "Map a function over an observable sequence. Unlike clojure.core/map, only supports up
   to 4 simultaneous source sequences at the moment."
   ([f ^Observable xs] (.map xs (iop/fn* f)))
-  ([f xs & observables] (apply base/zip f xs observables)))
+  ([f xs & observables] (apply zip f xs observables)))
 
 (defn ^Observable mapcat
   "Returns an observable which, for each value x in xs, calls (f x), which must
@@ -317,23 +391,6 @@
                                   (fn [o v] (on-next o (f (swap! n inc) v)))))))]
     (lift op xs)))
 
-; TODO which merge goes here?
-(defn merge
-  "
-  Returns an observable that emits a single map that consists of the rest of the
-  maps emitted by the input observable conj-ed onto the first.  If a key occurs
-  in more than one map, the mapping from the latter (left-to-right) will be the
-  mapping in the result.
-
-  NOTE: This is very different from rx.Observable/merge. See rx.base/merge for that
-  one.
-
-  See:
-    clojure.core/merge
-  "
-  [maps]
-  (reduce clojure.core/merge {} maps))
-
 (def next
   "Returns an observable that emits all but the first element of the input observable.
 
@@ -545,8 +602,8 @@
   "
   [f & args]
   (fn->o (-> #(apply f % args)
-             base/wrap-on-completed
-             base/wrap-on-error)))
+             wrap-on-completed
+             wrap-on-error)))
 
 (defmacro generator
   "Create an observable that executes body which should emit a sequence. bindings

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

@@ -1,8 +1,7 @@
 (ns rx.lang.clojure.future
   (:refer-clojure :exclude [future])
   (:require [rx.lang.clojure.interop :as iop]
-            [rx.lang.clojure.core :as rx :refer [fn->o fn->subscription]]
-            [rx.lang.clojure.base :as base]))
+            [rx.lang.clojure.core :as rx]))
 
 (def ^:private -ns- *ns*)
 (set! *warn-on-reflection* true)
@@ -26,13 +25,13 @@
   "
   [runner f & args]
   {:pre [(ifn? runner) (ifn? f)]}
-  (fn->o (fn [observer]
+  (rx/fn->o (fn [observer]
            (let [wrapped (-> (fn [o]
                                (apply f o args))
-                             base/wrap-on-completed
-                             base/wrap-on-error)
+                             rx/wrap-on-completed
+                             rx/wrap-on-error)
                  fu      (runner #(wrapped observer))]
-             (fn->subscription #(future-cancel fu))))))
+             (rx/fn->subscription #(future-cancel fu))))))
 
 (defmacro future-generator
   "Same as rx/generator macro except body is invoked in a separate thread.
@@ -60,12 +59,12 @@
   "
   [runner f & args]
   {:pre [(ifn? runner) (ifn? f)]}
-  (fn->o (fn [observer]
+  (rx/fn->o (fn [observer]
            (let [wrapped (-> #(rx/on-next % (apply f args))
-                             base/wrap-on-completed
-                             base/wrap-on-error)
+                             rx/wrap-on-completed
+                             rx/wrap-on-error)
                  fu      (runner #(wrapped observer))]
-             (fn->subscription #(future-cancel fu))))))
+             (rx/fn->subscription #(future-cancel fu))))))
 
 (defmacro future
   "Executes body in a separate thread and passes the single result to onNext.

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

@@ -1,6 +1,5 @@
 (ns rx.lang.clojure.realized
-  (:require [rx.lang.clojure.interop :as iop]
-            [rx.lang.clojure.base :as rx-base]))
+  (:require [rx.lang.clojure.interop :as iop]))
 
 (def ^:private -ns- *ns*)
 (set! *warn-on-reflection* true)
@@ -62,10 +61,12 @@
   single key maps, merging and then folding all the separate maps together. So code
   like this:
 
-    (rx/merge (rx-base/merge (->> (user-info-o user-id)
-                                  (rx/map (fn [u] {:user u})))
-                             (->> (user-likes-o user-id)
-                                  (rx/map (fn [u] {:likes u})))))
+    ; TODO update
+    (->> (rx/merge (->> (user-info-o user-id)
+                             (rx/map (fn [u] {:user u})))
+                        (->> (user-likes-o user-id)
+                             (rx/map (fn [u] {:likes u}))))
+         (rx/reduce merge {}))
 
   becomes:
 
@@ -86,8 +87,8 @@
                             .toList
                             (.map (iop/fn [list] {k (f list)})))))))]
 
-    (-> o
-        rx-base/merge ; funnel all the observables into a single sequence
+    (-> ^Iterable o
+        (rx.Observable/merge) ; funnel all the observables into a single sequence
         (.reduce {} (iop/fn* merge))))) ; do the map merge dance
 
 (defn ^rx.Observable realized-map*