Update blocking stuff

daveray · daveray · commit 42763685f936 · 2014-02-24T22:51:15.000-08:00
diff --git a/language-adaptors/rxjava-clojure/src/main/clojure/rx/lang/clojure/blocking.clj b/language-adaptors/rxjava-clojure/src/main/clojure/rx/lang/clojure/blocking.clj
@@ -1,47 +1,90 @@
 (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])
+  rx land back to sync land. For example, to produce a servlet response.
+
+  If you use these, you're a bad person.
+  "
+  (:refer-clojure :exclude [first into doseq last])
+  (:require [rx.lang.clojure.interop :as iop] [rx.lang.clojure.core :as rx])
   (:import [rx Observable]
            [rx.observables BlockingObservable]))
 
 (def ^:private -ns- *ns*)
 (set! *warn-on-reflection* true)
 
+(defmacro ^:private with-ex-unwrap
+  [& body]
+  `(try
+    ~@body
+    (catch RuntimeException e#
+      (throw (or
+               (and (identical? RuntimeException (class e#))
+                    (.getCause e#))
+               e#)))))
+
 (defn ^BlockingObservable ->blocking
-  "Convert an Observable to a BlockingObservable"
-  [^Observable o]
-  (.toBlockingObservable o))
+  "Convert an Observable to a BlockingObservable.
+
+  If o is already a BlockingObservable it's returned unchanged.
+  "
+  [o]
+  (if (instance? BlockingObservable o)
+    o
+    (.toBlockingObservable ^Observable o)))
+
+(defn o->seq
+  "Returns a lazy sequence of the items emitted by o
+
+  See:
+    rx.observables.BlockingObservable/getIterator
+    rx.lang.clojure.core/seq->o
+  "
+  [o]
+  (-> (->blocking o)
+      (.getIterator)
+      (iterator-seq)))
 
 (defn first
   "*Blocks* and waits for the first value emitted by the given observable.
 
+  If the Observable is empty, returns nil
+
   If an error is produced it is thrown.
 
   See:
     clojure.core/first
     rx/first
+    rx.observables.BlockingObservable/first
+  "
+  [observable]
+  (with-ex-unwrap
+    (.firstOrDefault (->blocking observable) nil)))
+
+(defn last
+  "*Blocks* and waits for the last value emitted by the given observable.
+
+  If the Observable is empty, returns nil
+
+  If an error is produced it is thrown.
+
+  See:
+    clojure.core/last
+    rx/last
+    rx.observable.BlockingObservable/last
   "
   [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)))))
+  (with-ex-unwrap
+    (.lastOrDefault (->blocking observable) nil)))
 
 (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)))
+  (with-ex-unwrap
+    (.single (->blocking observable))))
 
 (defn into
   "*Blocks* and pours the elements emitted by the given observables into
@@ -55,3 +98,40 @@
   "
   [to from-observable]
   (first (rx/into to from-observable)))
+
+(defn doseq*
+  "*Blocks* and executes (f x) for each x emitted by xs
+
+  Returns nil.
+
+  See:
+    doseq
+    clojure.core/doseq
+  "
+  [xs f]
+  (with-ex-unwrap
+    (-> (->blocking xs)
+        (.forEach (rx.lang.clojure.interop/action* f)))))
+
+(defmacro doseq
+  "Like clojure.core/doseq except iterates over an observable in a blocking manner.
+
+  Unlike clojure.core/doseq, only supports a single binding
+
+  Returns nil.
+
+  Example:
+
+    (rx-blocking/doseq [{:keys [name]} users-observable]
+      (println \"User:\" name))
+
+  See:
+    doseq*
+    clojure.core/doseq
+  "
+  [bindings & body]
+  (when (not= (count bindings) 2)
+    (throw (IllegalArgumentException. (str "sorry, rx/doseq only supports one binding"))))
+  (let [[k v] bindings]
+    `(doseq* ~v (fn [~k] ~@body))))
+
diff --git a/language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/blocking_test.clj b/language-adaptors/rxjava-clojure/src/test/clojure/rx/lang/clojure/blocking_test.clj
@@ -3,16 +3,63 @@
             [rx.lang.clojure.core :as rx]
             [clojure.test :refer [deftest testing is]]))
 
+(deftest test-->blocking
+  (testing "returns a BlockingObservable from an Observable"
+    (is (instance? rx.observables.BlockingObservable (b/->blocking (rx/return 0)))))
+  
+  (testing "is idempotent"
+    (is (instance? rx.observables.BlockingObservable (b/->blocking (b/->blocking (rx/return 0)))))))
+
+
+(deftest test-o->seq
+  (is (= [1 2 3] (b/o->seq (rx/seq->o [1 2 3])))))
+
 (deftest test-first
   (testing "returns first element of observable"
-    (is (= 1 (b/first (rx/return 1)))))
+    (is (= 1 (b/first (rx/seq->o [1 2 3])))))
+  (testing "returns nil for empty observable"
+    (is (nil? (b/first (rx/empty)))))
+  (testing "rethrows errors"
+    (is (thrown? java.io.FileNotFoundException
+                 (b/first (rx/throw (java.io.FileNotFoundException. "boo")))))))
+
+(deftest test-last
+  (testing "returns last element of observable"
+    (is (= 3 (b/last (rx/seq->o [1 2 3])))))
   (testing "returns nil for empty observable"
-    (is (nil? (b/first (rx/empty))))))
+    (is (nil? (b/last (rx/empty)))))
+  (testing "rethrows errors"
+    (is (thrown? java.io.FileNotFoundException
+                 (b/last (rx/throw (java.io.FileNotFoundException. "boo")))))))
 
 (deftest test-single
   (testing "returns one element"
     (is (= 1 (b/single (rx/return 1)))))
   (testing "throw if empty"
     (is (thrown? java.lang.IllegalArgumentException (b/single (rx/empty)))))
   (testing "throw if many"
-    (is (thrown? java.lang.IllegalArgumentException (b/single (rx/seq->o [1 2]))))))
+    (is (thrown? java.lang.IllegalArgumentException (b/single (rx/seq->o [1 2])))))
+  (testing "rethrows errors"
+    (is (thrown? java.io.FileNotFoundException
+                 (b/single (rx/throw (java.io.FileNotFoundException. "boo"))))))) 
+
+(deftest test-into
+  (is (= [1 2 3]
+         (b/into [1] (rx/seq->o [2 3]))))
+  (testing "rethrows errors"
+    (is (thrown? java.io.FileNotFoundException
+                 (b/into #{} (rx/throw (java.io.FileNotFoundException. "boo")))))))
+
+(deftest test-doseq
+  (is (= (range 3)
+         (let [capture (atom [])]
+           (b/doseq [{:keys [value]} (rx/seq->o (map #(hash-map :value %) (range 3)))]
+             (println value)
+             (swap! capture conj value))
+           @capture)))
+
+  (testing "rethrows errors"
+    (is (thrown? java.io.FileNotFoundException
+                 (b/doseq [i (rx/seq->o (range 3))]
+                   (throw (java.io.FileNotFoundException. "boo")))))))
+
