Skip to content

Instantly share code, notes, and snippets.

@gszeliga
Created December 6, 2017 15:07
Show Gist options
  • Save gszeliga/a20a9c1fd6fbeef40cffa6ec039af974 to your computer and use it in GitHub Desktop.
Save gszeliga/a20a9c1fd6fbeef40cffa6ec039af974 to your computer and use it in GitHub Desktop.
(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)))))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment