core.async中的状态换能器

Question

我试图了解如何在core.async中制作状态传感器。例如，我如何制作一个传感器来统计通过一个通道来的元素的数量？例如，我希望将输入转换为一个计数，该计数取决于它之前的对象的数量。

从我读到的方法来看，使用volatile!来保持换能器内部的状态，但我仍然不知道如何把所有的东西放在一起。

Answer 1

您需要一个状态转换器，返回一个关闭在volatile!上的还原函数，跟踪计数。

(defn count-xf [rf]
  (let [ctr (volatile! 0)]
    (fn
      ([] (rf))
      ([result] (rf result))
      ([result _]                         ; we ignore the input as
       (rf result (vswap! ctr inc))))))   ; we just pass on the count

这可以使用核心函数completing进行简化。

(defn count-xf [rf]
  (let [ctr (volatile! 0)]
    (completing
     (fn [result _]
       (rf result (vswap! ctr inc))))))

例如:使用它

(let [ch (chan 1 count-xf)]
  (onto-chan ch (repeat 10 true))
  (<!! (clojure.core.async/into [] ch)))

;-> [1 2 3 4 5 6 7 8 9 10]

或者，您可以只使用map-indexed换能器，但这可能会降低您对换能器工作原理的了解。而且，对于这个特定的使用过程，它需要每步增加一些额外的开销。

(def count-xf (map-indexed (fn [i _] (inc i))))

注意到它的实现与上面的实现没有什么不同。

进一步参考：http://clojure.org/reference/transducers