如何耗尽通道的值然后返回结果(ClojureScript)？

Question

假设通道chan在队列中具有值"1"和"2".

目标:创建一个获取chan并返回向量的函数[1 2].请注意,如果此函数在返回其值之前必须阻塞一段时间,我完全没问题.

尝试:

(defn chan->vector
  [chan]  
  (let [a (atom true) v []]
    (while (not-nil? @a)
      (go
        (reset! a (<! chan))
        (into v @a)
        (reset! a (<! chan))
      )
    ) v
  )
)

结果:我的REPL冻结并最终吐出一个巨大的错误.我已经意识到这是因为(go ...)块是异步的,所以立即返回.因此,我的(while ...)循环中的原子永远不会被设置为nil,并且循环永远不会终止.

那么我该如何实现预期的结果呢？如果它是相关的,我正在使用ClojureScript和目标nodejs.

Answer 1

你应该使用alts!from core.async来完成这个任务(https://clojure.github.io/core.async/#clojure.core.async/alts!):

(def x (chan 10))

(go (>! x 1)
    (>! x 2)
    (>! x 3))

(defn read-all [from-chan]
  (<!! (go-loop [res []]
           (let [[v _] (alts! [from-chan] :default :complete)]
             (if (= v :complete)
               res
               (recur (conj res v)))))))

(read-all x) 
;; output: [1 2 3]

(read-all x)
;; output: []

(go (>! x 10)
    (>! x 20)
    (>! x 30)
    (>! x 40))

(read-all x)
;; output: [10 20 30 40]

里面的go-loop这个(a/alts! [from-chan] :default :complete)尝试读取信道的任何值,并且如果没有可用的一次它发出的默认值的值,所以你会看到,你应该打破循环和返回累积值.

更新:由于<!!cljs中不存在阻塞read(),您可以通过以下方式重写它:

(defn read-all [from-chan]
  (go-loop [res []]
    (let [[v _] (alts! [from-chan] :default :complete)]
      (if (= v :complete)
        res
        (recur (conj res v)))))))

所以它将返回通道,然后从那里读取一个值:

(go (let [res (<! (read-all x))]
      (println res)
      ;; do something else
      ))