Merge pull request #3 from nervous-systems/channel-exceptions

Single-value channel results

Author: alaisi

src/postgres/async.clj

@@ -1,5 +1,6 @@
 (ns postgres.async
-  (:require [postgres.async.impl :refer [consumer-fn defasync] :as pg])
+  (:require [postgres.async.impl :refer [consumer-fn defasync] :as pg]
+            [clojure.core.async :refer [<!]])
   (:import [com.github.pgasync Db ConnectionPoolBuilder
             QueryExecutor TransactionExecutor Transaction]
            [com.github.pgasync.impl.conversion DataConverter]))
@@ -106,11 +107,12 @@
 (defasync <rollback! [tx])
 
 (defmacro dosql
-  "Takes values from channels returned by db functions and returns [nil exception]
-   on first error. Returns [result-of-body nil] on success."
+  "Takes values from channels returned by db functions and returns exception
+   on first error. Returns the result of evaluating the given forms on success"
   [bindings & forms]
-  (let [err (gensym "e")]
-    `(let [~@(pg/async-sql-bindings bindings err)]
-       (if ~err
-         [nil ~err]
-         [(do ~@forms) nil]))))
+  (if-let [[l r & bindings] (not-empty bindings)]
+    `(let [~l (<! ~r)]
+       (if (instance? Throwable ~l)
+         ~l
+         (dosql ~bindings ~@forms)))
+    `(do ~@forms)))

src/postgres/async/impl.clj

@@ -1,14 +1,14 @@
 (ns postgres.async.impl
   (:require [clojure.string :as string]
-            [clojure.core.async :refer [chan put! go <!]])
+            [clojure.core.async :refer [chan put! go]])
   (:import [java.util.function Consumer]
            [com.github.pgasync ResultSet]
            [com.github.pgasync.impl PgRow]))
 
 (defmacro defasync [name args]
   `(defn ~name [~@args]
      (let [c# (chan 1)]
-       (~(symbol (subs (str name) 1)) ~@args #(put! c# [%1 %2]))
+       (~(symbol (subs (str name) 1)) ~@args #(put! c# (or %2 %1)))
        c#)))
 
 (defmacro consumer-fn [[param] body]
@@ -60,14 +60,3 @@
        " WHERE " (first where)
        (when returning
          (str " RETURNING " returning))))
-
-(defn async-sql-bindings
-  "Converts bindings x (f) to [x err] (if [err] [nil err] (<! (f)))"
-  [bindings err]
-  (let [vars (map (fn [v]
-                    [v err])
-                  (take-nth 2 bindings))
-        fs   (map (fn [f]
-                    `(if ~err [nil ~err] (<! ~f)))
-                  (take-nth 2 (rest bindings)))]
-    (list* [err err] [nil nil] (interleave vars fs))))

test/postgres/async_test.clj

@@ -7,10 +7,10 @@
 (def table "clj_pg_test")
 
 (defn- wait [channel]
-  (let [[r err] (<!! channel)]
-    (if err
-      (throw err)
-      r)))
+  (let [r (<!! channel)]
+    (if (instance? Throwable r)
+      (throw r))
+    r))
 
 (defn- create-tables [db]
   (wait (<execute! db [(str "drop table if exists " table)]))
@@ -71,4 +71,17 @@
                              rs (<query! tx ["select 123 as x"])
                              rs (<query! tx ["select $1::text as t" (:x (first rs))])
                              _  (<commit! tx)]
-                            (:t (first rs)))))))))
+                            (:t (first rs))))))))
+  (testing "dosql short-circuits on errors"
+    (let [e (Exception. "Oops!")
+          executed (atom 0)]
+      (is (= (try
+               (wait (go (dosql
+                          [_ (<query! *db* ["select 123 as t"])
+                           _ (go e)
+                           _ (swap! executed inc)]
+                          "Just hanging out")))
+               (catch Exception caught
+                 {:caught caught}))
+             {:caught e}))
+      (is (= @executed 0)))))