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))))))

E. g.使用它

(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