Return async results as a stream

In an async query, success sequence and success partial rows are put
on the channel row by row. This means that the consumer can be unaware
of the batching that takes place between the driver and RethinkDB.

Author: danielcompton

src/rethinkdb/net.clj

@@ -150,6 +150,7 @@
 (defn add-token [conn query]
   (let [token (:token (swap! (:conn conn) update-in [:token] inc))
         query (assoc query :token token)]
+    (assert (nil? (get-in @conn [:pending token])))
     (swap! (:conn conn) assoc-in [:pending token] query)
     query))
 
@@ -176,6 +177,7 @@
   (let [{:keys [async?]} query
         {:keys [initial-query-chan]} @conn
         stream (s/stream)]
+    (log/debug "Init query" query)
     (async/go (async/>! initial-query-chan
                         (assoc query :result stream)))
     (if async?

test-resources/logback-test.xml

@@ -8,5 +8,5 @@
   <root level="ERROR">
     <appender-ref ref="STDOUT"/>
   </root>
-  <logger name="rethinkdb.net" level="INFO"/>
+  <logger name="rethinkdb.net" level="DEBUG"/>
 </configuration>

test/rethinkdb/async_test.clj

@@ -10,6 +10,11 @@
 (use-fixtures :each utils/setup-each)
 (use-fixtures :once utils/setup-once)
 
+(def pokemon [{:national_no 25 :name "Pikachu"}
+              {:national_no 26 :name "Raichu"}])
+(def changefeed-pokemon (map #(hash-map :new_val %) pokemon))
+
+
 (deftest always-return-async
   (let [root-logger ^Logger (LoggerFactory/getLogger "rethinkdb.net")
         level (.getLevel root-logger)]
@@ -22,31 +27,30 @@
     (.setLevel root-logger level)))
 
 (deftest async-results
-  (let [conn (r/connect :async? true :db utils/test-db)
-        pokemon [{:national_no 25 :name "Pikachu"}
-                 {:national_no 26 :name "Raichu"}]]
-    (are [expected query] (= (->> (r/run query conn)
-                                  (async/into [])
-                                  (async/<!!))
-                             expected)
-      ;; Insert (success atom)
-      [{:deleted   0
-        :errors    0
-        :inserted  2
-        :replaced  0
-        :skipped   0
-        :unchanged 0}]
-      (-> (r/table utils/test-table)
-          (r/insert pokemon))
-
-      ;; Success sequence
-      pokemon (-> (r/table utils/test-table))
-
-      ;; Success atom
-      [pokemon] (-> (r/table utils/test-table) (r/order-by :name))
-
-      ;; Changefeed
-      (map #(hash-map :new_val %) pokemon)
-      (-> (r/table utils/test-table)
-          (r/changes {:include-initial true})
-          (r/limit 2)))))
+  (with-open [conn (r/connect :async? true :db utils/test-db)]
+    (testing "Shape of async results"
+      (are [expected query] (= (->> (r/run query conn)
+                                    (async/into [])
+                                    (async/<!!))
+                               expected)
+        ;; Insert (success atom)
+        [{:deleted   0
+          :errors    0
+          :inserted  2
+          :replaced  0
+          :skipped   0
+          :unchanged 0}]
+        (-> (r/table utils/test-table)
+            (r/insert pokemon))
+
+        ;; Success sequence
+        pokemon (-> (r/table utils/test-table))
+
+        ;; Success atom
+        [pokemon] (-> (r/table utils/test-table) (r/order-by :name))
+
+        ;; Changefeed
+        changefeed-pokemon
+        (-> (r/table utils/test-table)
+            (r/changes {:include-initial true})
+            (r/limit 2))))))