diff --git a/README.md b/README.md
index b404fb2..4c6b1bc 100644
--- a/README.md
+++ b/README.md
@@ -1,14 +1,15 @@
 postgres.async
 ==============
 
+[![Clojars Project](https://img.shields.io/clojars/v/alaisi/postgres.async.svg)](https://clojars.org/alaisi/postgres.async)
+
 Asynchronous PostgreSQL client for Clojure.
 
 ## Download
 
 Add the following to your [Leiningen](http://github.com/technomancy/leiningen) `project.clj`:
 
-![latest postgres.async version](https://clojars.org/alaisi/postgres.async/latest-version.svg)
-
+[![latest postgres.async version](https://clojars.org/alaisi/postgres.async/latest-version.svg)](https://clojars.org/alaisi/postgres.async)
 
 ## Setting up a connection pool
 
@@ -103,7 +104,7 @@ See composition below for example.
 
 ## Composition
 
-Channel-returning functions can be composed with `dosql` macro that returns result of last form of first exception.
+Channel-returning functions can be composed with `dosql` macro that returns result of last form or first exception.
 
 ```clojure
 (<!! (go
@@ -138,7 +139,14 @@ Support for custom types can be added by extending `IPgParameter` protocol and `
     (.getBytes (str store) "UTF-8")))
 
 (defmethod from-pg-value com.github.pgasync.impl.Oid/HSTORE [oid ^bytes value]
-  (my-h-store/parse-string (String. value "UTF-8)))
+  (my-h-store/parse-string (String. value "UTF-8")))
+```
+
+`from-pg-value` can also be used for overriding "core" types. This is especially useful with temporal data types that are by default converted to `java.sql.Date`, `java.sql.Time`, `java.sql.Timestamp`.
+
+```clojure
+(defmethod from-pg-value com.github.pgasync.impl.Oid/DATE [oid ^bytes value]
+  (java.time.LocalDate/parse (String. value "UTF-8")))
 ```
 
 ## Dependencies
diff --git a/dev/user.clj b/dev/user.clj
index 789214a..76db850 100644
--- a/dev/user.clj
+++ b/dev/user.clj
@@ -6,7 +6,8 @@
   (pg/open-db {:hostname "localhost"
                :database "postgres"
                :username "postgres"
-               :password "postgres"}))
+               :password "postgres"
+               :validation-query "select 1"}))
 
 (defn reload []
   (repl/refresh))
diff --git a/project.clj b/project.clj
index d52a36b..f78cba8 100644
--- a/project.clj
+++ b/project.clj
@@ -1,17 +1,17 @@
-(defproject alaisi/postgres.async "0.6.0"
+(defproject alaisi/postgres.async "0.9.0-SNAPSHOT"
   :description "Asynchronous PostgreSQL Clojure client"
   :url "http://github.com/alaisi/postgres.async"
   :license {:name "Eclipse Public License"
             :url "http://www.eclipse.org/legal/epl-v10.html"}
   :scm {:name "git"
         :url "http://github.com/alaisi/postgres.async.git"}
-  :dependencies [[org.clojure/clojure "1.6.0"]
-                 [org.clojure/core.async "0.1.346.0-17112a-alpha"]
-                 [com.github.alaisi.pgasync/postgres-async-driver "0.6"]
-                 [cheshire "5.5.0" :scope "provided"]]
+  :dependencies [[org.clojure/clojure "1.8.0"]
+                 [org.clojure/core.async "0.2.374"]
+                 [com.github.alaisi.pgasync/postgres-async-driver "0.9"]
+                 [cheshire "5.6.1" :scope "provided"]]
   :lein-release {:deploy-via :clojars}
   :global-vars {*warn-on-reflection* true}
   :target-path "target/%s"
   :profiles {:dev {:source-paths ["dev"]
-                   :dependencies [[org.clojure/tools.namespace "0.2.4"]
-                                  [org.clojure/java.classpath "0.2.0"]]}})
+                   :dependencies [[org.clojure/tools.namespace "0.2.11"]
+                                  [org.clojure/java.classpath "0.2.3"]]}})
diff --git a/src/postgres/async.clj b/src/postgres/async.clj
index 098fb26..caa2c60 100644
--- a/src/postgres/async.clj
+++ b/src/postgres/async.clj
@@ -6,19 +6,28 @@
            [com.github.pgasync.impl.conversion DataConverter]))
 
 (defmulti from-pg-value (fn [oid value] oid))
+(defmethod from-pg-value :default [oid value] ::raw-value)
+
 (defprotocol IPgParameter
   (to-pg-value [value]))
 
 (defn- create-converter []
   (proxy [DataConverter] []
-    (toConvertable [oid value]
-      (from-pg-value oid value))
     (fromConvertable [value]
-      (to-pg-value value))))
+      (to-pg-value value))
+    (toObject [oid value]
+      (when value
+        (let [val (from-pg-value oid value)
+              this ^DataConverter this]
+          (if-not (= ::raw-value val)
+            val
+            (proxy-super toObject oid value)))))))
 
 (defn open-db
   "Creates a db connection pool"
-  [{:keys [hostname port username password database pool-size ssl] :as config}]
+  [{:keys [hostname port username password database
+           pool-size ssl validation-query pipeline]
+    :as config}]
   (doseq [param [:hostname :username :password :database]]
     (when (nil? (param config))
       (throw (IllegalArgumentException. (str param " is required")))))
@@ -29,7 +38,9 @@
       (.username username)
       (.password password)
       (.ssl (boolean ssl))
+      (.pipeline (boolean pipeline))
       (.poolSize (or pool-size 25))
+      (.validationQuery (or validation-query ""))
       (.dataConverter (create-converter))
       (.build)))
 
@@ -57,6 +68,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
diff --git a/src/postgres/async/impl.clj b/src/postgres/async/impl.clj
index cb68264..924b9bc 100644
--- a/src/postgres/async/impl.clj
+++ b/src/postgres/async/impl.clj
@@ -17,20 +17,36 @@
     (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))
 
+(defn- get-columns [^PgRow row]
+  (keys (.getColumns row)))
+
+(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 +84,5 @@
        " WHERE " (first where)
        (when returning
          (str " RETURNING " returning))))
+
+
diff --git a/test/postgres/async_test.clj b/test/postgres/async_test.clj
index df39264..cc240ee 100644
--- a/test/postgres/async_test.clj
+++ b/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")
@@ -25,6 +26,8 @@
                              :database (env "PG_DB" "postgres")
                              :username (env "PG_USER" "postgres")
                              :password (env "PG_PASSWORD" "postgres")
+                             :validation-query "select 1"
+                             :pipeline true
                              :pool-size 1})]
       (try
         (create-tables *db*)
@@ -37,7 +40,11 @@
 
   (testing "query returns rows as map"
     (let [rs (wait (query! *db* ["select 1 as x"]))]
-      (is (= 1 (get-in rs [0 :x]))))))
+      (is (= 1 (get-in rs [0 :x])))))
+
+  (testing "query returns oid bool as boolean"
+    (let [rs (wait (query! *db* ["select true as b"]))]
+      (is (= true (get-in rs [0 :b]))))))
 
 (deftest query-for-array
 
@@ -47,7 +54,40 @@
 
   (testing "nested arrays are converted to vectors"
     (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]))))))
+      (is (= [[1 2] [3 4] [5 nil]] (get-in rs [0 :a])))))
+
+  (testing "bool arrays are converted to boolean vectors"
+    (let [rs (wait (query! *db* ["select '{true,false}'::BOOL[] as b"]))]
+      (is (= [true false] (get-in rs [0 :b]))))))
+
+(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 query-for-overridden-types
+
+  (defmethod from-pg-value com.github.pgasync.impl.Oid/TIMESTAMP [oid ^bytes value]
+    (String. value "utf-8"))
+
+  (testing "timestamp can be returned as string"
+    (let [c (query-rows! *db* ["select $1::TIMESTAMP as ts"
+                               "2016-05-01T12:00:00.001"])]
+      (is (= {:ts "2016-05-01 12:00:00.001"} (<!! c)))))
+
+  (testing "unsupported type throws an exception"
+    (let [c (query-rows! *db* ["select '<x/>'::xml"])
+          ^Throwable ex (<!! c)]
+      (is (= "Unknown conversion source: XML" (.getMessage ex))))))
 
 (deftest inserts