javaadvent_core_async_aggregate.clj

(defn aggregate-field [fname channel npar]
  (let [as-value-fn       (extract-field-fn fname channel)
        fvalue-chan       (pipe channel (chan 1024 (map as-value-fn)))
        fvalue-aggregator (for [_ (range npar)]
                            (w-aggregate fvalue-chan))]
    (go
      (let [[sum count] (<! (async/reduce 
                             #(apply map + [%1 %2]) 
                             [0.0 0] 
                             (async/merge fvalue-aggregator)))]
        (/ sum count)))))