async-transduce.clj

(ns some.core
  (:refer-clojure :exclude [transduce])
  (:require [clojure.core.async :as async]))

;; -----------------------------------------------------------------------------

(defn transduce
  "An async transduce variant. Does not provide any special handling for steps
  in xform that might return channels, i.e. steps should return/pass non-channel
  values."
  ([xform f chan] (transduce xform f (f) chan))
  ([xform f init chan]
   (go (let [f (xform f)
             ret (async/reduce f init chan)]
         (f (<! ret))))))

;; (transduce xform (completing conj) (async/merge [ch1 ch2 ch3 ...]))