Initial import of indigena-rx

Imported code, fixed namespaces and adusted one test for changes since
0.14.1.

Author: daveray

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

@@ -0,0 +1,75 @@
+(ns rx.lang.clojure.base
+  "Generic, low-level rx helpers."
+  (:refer-clojure :exclude [merge])
+  (:require [rx.lang.clojure.interop :as iop])
+  (:import [rx Observable Observer Subscription]
+           [rx.observables BlockingObservable]
+           [rx.subscriptions Subscriptions]))
+
+(def ^:private -ns- *ns*)
+(set! *warn-on-reflection* true)
+
+(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 Exception 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
+  "
+  [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)))

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

@@ -0,0 +1,57 @@
+(ns rx.lang.clojure.blocking
+  "Blocking operators and functions. These should never be used in
+  production code except at the end of an async chain to convert from
+  rx land back to sync land. For example, to produce a servlet response."
+  (:refer-clojure :exclude [first into])
+  (:require [rx.lang.clojure.core :as rx])
+  (:import [rx Observable]
+           [rx.observables BlockingObservable]))
+
+(def ^:private -ns- *ns*)
+(set! *warn-on-reflection* true)
+
+(defn ^BlockingObservable ->blocking
+  "Convert an Observable to a BlockingObservable"
+  [^Observable o]
+  (.toBlockingObservable o))
+
+(defn first
+  "*Blocks* and waits for the first value emitted by the given observable.
+
+  If an error is produced it is thrown.
+
+  See:
+    clojure.core/first
+    rx/first
+  "
+  [observable]
+  (let [result (clojure.core/promise)]
+    (rx/subscribe (->> observable (rx/take 1))
+                #(clojure.core/deliver result [:value %])
+                #(clojure.core/deliver result [:error %])
+                #(clojure.core/deliver result nil))
+    (if-let [[type v] @result]
+      (case type
+        :value v
+        :error (throw v)))))
+
+(defn single
+  "*Blocks* and waits for the first value emitted by the given observable.
+
+   An error is thrown if more then one value is produced.
+  "
+  [observable]
+  (.single (->blocking observable)))
+
+(defn into
+  "*Blocks* and pours the elements emitted by the given observables into
+  to.
+
+  If an error is produced it is thrown.
+
+  See:
+    clojure.core/into
+    rx/into
+  "
+  [to from-observable]
+  (first (rx/into to from-observable)))

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

@@ -0,0 +1,97 @@
+(ns rx.lang.clojure.chunk
+  (:refer-clojure :exclude [chunk])
+  (:require [rx.lang.clojure.core :as rx]
+            [rx.lang.clojure.base :as rx-base]))
+
+(def ^:private -ns- *ns*)
+(set! *warn-on-reflection* true)
+
+(defn chunk
+  "Same as rx.Observable.merge(Observable<Observable<T>>) but the input Observables
+  are \"chunked\" so that at most chunk-size of them are \"in flight\" at any given
+  time.
+
+  The order of the input Observables is not preserved.
+
+  The main purpose here is to allow a large number of Hystrix observables to
+  be processed in a controlled way so that the Hystrix execution queues aren't
+  overwhelmed.
+
+  Example:
+
+    (->> users
+         (map #(-> (GetUserCommand. %) .toObservable))
+         (chunk 10))
+
+  See:
+    http://netflix.github.io/RxJava/javadoc/rx/Observable.html#merge(rx.Observable)
+    http://netflix.github.io/RxJava/javadoc/rx/Observable.html#mergeDelayError(rx.Observable)
+  "
+  ([chunk-size observable-source] (chunk chunk-size {} observable-source))
+  ([chunk-size options observable-source]
+  (let [new-state-atom #(atom {:in-flight #{}     ; observables currently in-flight
+                               :buffered  []      ; observables waiting to be emitted
+                               :complete  false   ; true if observable-source is complete
+                               :observer  % })    ; the observer
+        ps   #(do (printf "%s/%d/%d%n"
+                          (:complete %)
+                          (-> % :buffered count)
+                          (-> % :in-flight count))
+                (flush))
+
+        ; Given the current state, returns [action new-state]. action is the
+        ; next Observable or Throwable to emit, or :complete if we're done.
+        next-state (fn [{:keys [complete buffered in-flight] :as old}]
+                     (cond
+                       (empty? buffered)                [complete old]
+
+                       (< (count in-flight) chunk-size) (let [next-o (first buffered)]
+                                                          [next-o
+                                                           (-> old
+                                                               (update-in [:buffered] next)
+                                                               (update-in [:in-flight] conj next-o))])
+
+                       :else                            [nil old]))
+
+        ; Advance the state, performing side-effects as necessary
+        advance! (fn advance! [state-atom]
+                   (let [old-state        @state-atom
+                         [action new-state] (next-state old-state)]
+                     (if (compare-and-set! state-atom old-state new-state)
+                       (let [observer (:observer new-state)]
+                         (if (:debug options) (ps new-state))
+                         (cond
+                          (= :complete action)
+                            (rx/on-completed observer)
+
+                          (instance? Throwable action)
+                            (rx/on-error observer action)
+
+                          (instance? rx.Observable action)
+                            (rx/on-next observer
+                                        (.finallyDo ^rx.Observable action
+                                                    (reify rx.util.functions.Action0
+                                                              (call [this]
+                                                                (swap! state-atom update-in [:in-flight] disj action)
+                                                                (advance! state-atom)))))))
+                       (recur state-atom))))
+
+        subscribe (fn [state-atom]
+                    (rx/subscribe observable-source
+                                  (fn [o]
+                                    (swap! state-atom update-in [:buffered] conj o)
+                                    (advance! state-atom))
+
+                                  (fn [e]
+                                    (swap! state-atom assoc :complete e)
+                                    (advance! state-atom))
+
+                                  (fn []
+                                    (swap! state-atom assoc :complete :complete)
+                                    (advance! state-atom))))
+        observable (rx/fn->o (fn [observer]
+                               (subscribe (new-state-atom observer)))) ]
+    (if (:delay-error? options)
+      (rx.Observable/mergeDelayError observable)
+      (rx.Observable/merge observable)))))
+