Issue #18: Add support for streaming query rows

Author: alaisi

src/postgres/async.clj

@@ -61,6 +61,14 @@
      (execute! db sql (fn [rs err]
                         (f (:rows rs) err)))))
 
+(defn query-rows!
+  "Executes an sql query with parameters and returns a channel where 0-n rows are emitted."
+  [^QueryExecutor db [sql & params]]
+  (let [c (chan)]
+    (-> (.queryRows db sql (into-array params))
+        (.subscribe (pg/row-observer c)))
+    c))
+
 (defn insert!
   "Executes an sql insert and returns update count and returned rows.
    Spec format is

src/postgres/async/impl.clj

@@ -17,20 +17,40 @@
     (apply f (concat args [callback]))
     channel))
 
-(defn column->value [^Object value]
+(defn- column->value [^Object value]
   (if (and value (-> value .getClass .isArray))
     (vec (map column->value value))
     value))
 
+;; TODO: make columns public in the Java driver
+(defn- get-columns [^PgRow row]
+  (-> (doto (.getDeclaredField PgRow "columns")
+        (.setAccessible true))
+      (.get row)
+      (keys)))
+
+(defn- row->map [^PgRow row ^Object rowmap ^String col]
+  (assoc rowmap
+         (keyword (.toLowerCase col))
+         (column->value (.get row col))))
+
 (defn result->map [^ResultSet result]
-  (let [columns (.getColumns result)
-        row->map (fn [^PgRow row rowmap ^String col]
-                   (assoc rowmap (keyword (.toLowerCase col))
-                          (column->value (.get row col))))]
+  (let [columns (.getColumns result)]
     {:updated (.updatedRows result)
      :rows (vec (map (fn [row]
-                           (reduce (partial row->map row) {} columns))
-                         result))}))
+                       (reduce (partial row->map row) {} columns))
+                     result))}))
+
+(defn ^rx.Observer row-observer [channel]
+  (reify rx.Observer
+    (onNext [_ row]
+      (put! channel (reduce (partial row->map row)
+                            {} (get-columns row))))
+    (onError [_ err]
+      (put! channel err)
+      (close! channel))
+    (onCompleted [_]
+      (close! channel))))
 
 (defn- list-columns [data]
   (if (map? data)
@@ -68,3 +88,5 @@
        " WHERE " (first where)
        (when returning
          (str " RETURNING " returning))))
+
+

test/postgres/async_test.clj

@@ -1,7 +1,8 @@
 (ns postgres.async-test
   (:require [clojure.test :refer :all]
             [clojure.core.async :refer [<!! go]]
-            [postgres.async :refer :all]))
+            [postgres.async :refer :all])
+  (:import [com.github.pgasync SqlException]))
 
 (def ^:dynamic *db*)
 (def table "clj_pg_test")
@@ -51,6 +52,20 @@
     (let [rs (wait (query! *db* ["select '{{1,2},{3,4},{5,NULL}}'::INT[][] as a"]))]
       (is (= [[1 2] [3 4] [5 nil]] (get-in rs [0 :a]))))))
 
+(deftest query-for-rows
+
+  (testing "each row is emitted to channel"
+    (let [c (query-rows! *db* ["select generate_series(1, $1::int4) as s", 3])]
+      (is (= {:s 1} (<!! c)))
+      (is (= {:s 2} (<!! c)))
+      (is (= {:s 3} (<!! c)))
+      (is (nil? (<!! c)))))
+
+  (testing "a single error is emitted on query failure")
+    (let [c (query-rows! *db* ["selectx"])]
+      (is (instance? SqlException (<!! c)))
+      (is (nil? (<!! c)))))
+
 (deftest inserts
 
   (testing "insert returns row count"