clojure - 如何使用core.async channel 为Rx的`withLatestFrom`建模？

例如，给定一个带操作的通道，另一个带数据的通道，如何编写一个go块，该块将对数据通道上的最后一个值应用该操作？

(go-loop []
  (let [op (<! op-ch)
        data (<! data-ch)]
    (put! result-ch (op data))))

显然，这是行不通的，因为这将需要两个通道具有相同的频率。

（请参见http://rxmarbles.com/#withLatestFrom）

最佳答案

使用alts!，您可以完成所需的操作。

下面显示的with-latest-from实现了RxJS（我认为：P）在withLatestFrom中发现的相同行为。

(require '[clojure.core.async :as async])

(def op-ch (async/chan))
(def data-ch (async/chan))

(defn with-latest-from [chs f]
  (let [result-ch (async/chan)
        latest    (vec (repeat (count chs) nil))
        index     (into {} (map vector chs (range)))]
    (async/go-loop [latest latest]
      (let [[value ch] (async/alts! chs)
            latest     (assoc latest (index ch) value)]
        (when-not (some nil? latest)
          (async/put! result-ch (apply f latest)))
        (when value (recur latest))))
    result-ch))

(def result-ch (with-latest-from [op-ch data-ch] str))

(async/go-loop []
  (prn (async/<! result-ch))
  (recur))

(async/put! op-ch :+)
;= true
(async/put! data-ch 1)
;= true
; ":+1"
(async/put! data-ch 2)
;= true
; ":+2"
(async/put! op-ch :-)
;= true
; ":-2"