asynchronous - Clojure RethinkDB 订阅 changefeed

Question

我正在使用 Clojure driver for RethinkDB .
我想从查询中获取更改提要。这是我到目前为止所拥有的:

(defn change-feed [conn]
  (loop [changes (future
                   (-> (r/db "mydb")
                       (r/table "mytable")
                       r/changes
                       (r/run conn)))]
    (println "date : " ((comp :name :newval) first @changes)) ;;prints nil
    (recur (rest changes))))

它在调用时阻塞在我的 REPL 中(这是正常的)。然后我使用 RethinkDB 接口(interface)添加数据。它打印 nil我收到以下错误:

IllegalArgumentException Don't know how to create ISeq from: clojure.core$future_call$reify__6736  clojure.lang.RT.seqFrom (RT.java:528)

我究竟做错了什么 ？我希望能够:

带走 future 的元素

也知道我一次可以带多少元素(如果有几件正在等待)

注意:我打算使用 manifold 最终操纵结果，因此使用它的任何解决方案都完全可以。

Answer 1

我认为你不需要 future block ，clj-rethinkdb 库返回的光标将阻塞，直到它准备好。

您也可以使用 doseq而不是一个可能更自然的循环。

我是 clj-rethinkdb 的维护者，我们已经在管道中进行了更改，以呈现 core.async 接口(interface)和新的 Cursor 抽象。想早点使用，请查看https://github.com/apa512/clj-rethinkdb/pull/55