clojure - core.async 中的有状态转换器

Question

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

据我所知，要走的路是使用 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))))

观察它的 implementation与上面的实现几乎没有区别。

进一步引用:http://clojure.org/reference/transducers