Add support for custom types

alaisi · alaisi · commit 3a56c04a705e · 2014-09-15T12:53:43.000+03:00
diff --git a/README.md b/README.md
@@ -51,5 +51,17 @@ Asynchronous PostgreSQL client for Clojure
 
 (close-db! db)
 ; nil
+
+;; Extension points for custom column types
+(require '[cheshire.core :as json])
+
+(defmethod from-pg-value Oid/JSON [oid value]
+  (json/parse-string value))
+
+(extend-protocol IPgParameter 
+  clojure.lang.IPersistentMap
+  (to-pg-value [value]
+    (.getBytes (json/generate-string value))))
+
 ```
 
diff --git a/src/clj_postgres_async/core.clj b/src/clj_postgres_async/core.clj
@@ -2,9 +2,17 @@
   (:require [clojure.string :as string]
             [clojure.core.async :refer [chan put! go <!] :as async])
   (:import [com.github.pgasync ConnectionPoolBuilder]
-           [com.github.pgasync.callback ErrorHandler ResultHandler
-            TransactionHandler TransactionCompletedHandler])
-  (:gen-class))
+           [com.github.pgasync.impl Oid]
+           [com.github.pgasync.impl.conversion DataConverter]
+           [java.util.function Consumer]))
+
+(defmacro ^:private consumer-fn [[param] body]
+  `(reify Consumer (accept [_# ~param]
+                     (~@body))))
+
+(defmulti from-pg-value (fn [oid value] oid))
+(defprotocol IPgParameter
+  (to-pg-value [value]))
 
 (defn open-db [{:keys [hostname port username password database pool-size]}]
   (-> (ConnectionPoolBuilder.)
@@ -14,6 +22,11 @@
       (.username username)
       (.password password)
       (.poolSize (or pool-size 25))
+      (.dataConverter (proxy [DataConverter] []
+                        (toConvertable [oid value]
+                          (from-pg-oid oid value))
+                        (fromConvertable [value]
+                          (to-pg-value value))))
       (.build)))
 
 (defn close-db! [db]
@@ -32,12 +45,11 @@
     {:updated (.updatedRows result) :rows (into [] rows)}))
 
 (defn execute! [db [sql & params] f]
-  (let [handler (reify
-                  ResultHandler (onResult [_ r]
-                                  (f [(result->map r) nil]))
-                  ErrorHandler (onError [_ t]
-                                 (f [nil t])))]
-    (.query db sql params handler handler)))
+  (.query db sql params
+          (consumer-fn [rs]
+                       (f [(result->map rs) nil]))
+          (consumer-fn [exception]
+                       (f [nil exception]))))
 
 (defn query! [db sql f]
   (execute! db sql (fn [[rs err]]
@@ -76,28 +88,25 @@
           f))
 
 (defn begin! [db f]
-  (let [handler (reify
-                  TransactionHandler (onBegin [_ tx]
-                                       (f [tx nil]))
-                  ErrorHandler (onError [_ t]
-                                 (f [nil t])))]
-    (.begin db handler handler)))
-
-(defn- complete-tx [tx completion-fn f]
-  (let [handler (reify
-                  TransactionCompletedHandler (onComplete [_]
-                                       (f [true nil]))
-                  ErrorHandler (onError [_ t]
-                                 (f [nil t])))]
-    (completion-fn handler)))
+  (.begin db
+          (consumer-fn [tx]
+                       (f [tx nil]))
+          (consumer-fn [exception]
+                       (f [nil exception]))))
 
 (defn commit! [tx f]
-  (complete-tx tx #(.commit tx %1 %1) f))
+  (.commit tx
+           #(f [true nil])
+           (consumer-fn [exception]
+                        (f [nil exception]))))
 
 (defn rollback! [tx f]
-  (complete-tx tx #(.rollback tx %1 %1) f))
+  (.rollback tx
+             #(f [true nil])
+             (consumer-fn [exception]
+                          (f [nil exception]))))
 
-(defmacro defasync [name args]
+(defmacro ^:private defasync [name args]
   `(defn ~name [~@args]
      (let [c# (chan 1)]
        (~(symbol (subs (str name) 1)) ~@args #(put! c# %))
diff --git a/test/clj_postgres_async/core_test.clj b/test/clj_postgres_async/core_test.clj
@@ -5,52 +5,51 @@
 
 (def ^:private ^:dynamic *db*)
 
-(defn- await [channel]
+(defn- wait [channel]
   (let [[r err] (<!! channel)]
     (if err
       (throw err)
       r)))
 
 (defn- create-tables [db]
-  (await (<execute! db ["drop table if exists clj_pg_test"]))
-  (await (<execute! db ["create table clj_pg_test (
+  (wait (<execute! db ["drop table if exists clj_pg_test"]))
+  (wait (<execute! db ["create table clj_pg_test (
                            id serial, t varchar(10))"])))
 
-(defn- env [name default]
-  (if-let [value (System/getenv name)]
-    value
-    default))
-
 (defn- db-fixture [f]
-  (binding [*db* (open-db {:hostname (env "PG_HOST" "localhost")
-                           :port     (env "PG_PORT" 5432)
-                           :database (env "PG_DB" "postgres")
-                           :username (env "PG_USER" "postgres")
-                           :password (env "PG_PASSWORD" "postgres")
-                           :pool-size 1})]
-    (try
-      (create-tables *db*)
-      (f)
-      (finally (close-db! *db*)))))
+  (letfn [(env [name default]
+            (or (System/getenv name) default))]
+    (binding [*db* (open-db {:hostname (env "PG_HOST" "localhost")
+                             :port     (env "PG_PORT" 5432)
+                             :database (env "PG_DB" "postgres")
+                             :username (env "PG_USER" "postgres")
+                             :password (env "PG_PASSWORD" "postgres")
+                             :pool-size 1})]
+      (try
+        (create-tables *db*)
+        (f)
+        (finally (close-db! *db*))))))
 
 (use-fixtures :each db-fixture)
 
 (deftest queries
-  (testing "<query! returns rows as map"
-    (let [rs (await (<query! *db* ["select 1 as x"]))]
+  (testing "query returns rows as map"
+    (let [rs (wait (<query! *db* ["select 1 as x"]))]
       (is (= 1 (get-in rs [0 :x]))))))
 
 (deftest inserts
   (testing "insert return row count"
-    (let [rs (await (<insert! *db* {:table "clj_pg_test"} {:t "x"}))]
+    (let [rs (wait (<insert! *db* {:table "clj_pg_test"} {:t "x"}))]
       (is (= 1 (:updated rs)))))
   (testing "insert with returning returns generated keys"
-    (let [rs (await (<insert! *db* {:table "clj_pg_test" :returning "id"} {:t "y"}))]
+    (let [rs (wait (<insert! *db* {:table "clj_pg_test" :returning "id"} {:t "y"}))]
       (is (get-in rs [:rows 0 :id])))))
 
 (deftest sql-macro
   (testing "dosql returns last form"
-    (is (= "123" (await (go (dosql
-                             [rs (<query! *db* ["select 123 as x"])
-                              rs (<query! *db* ["select $1::text as t" (:x (first rs))])]
+    (is (= "123" (wait (go (dosql
+                             [tx (<begin! *db*)
+                              rs (<query! tx ["select 123 as x"])
+                              rs (<query! tx ["select $1::text as t" (:x (first rs))])
+                              _  (<commit! tx)]
                              (:t (first rs)))))))))
