pipeline.clj

(defn update-stats
  "Take a collection of pages and update stats (fns)."
  [pages & [fns]]
  (let [fns (or fns [update-stat1 update-stat2 clean-page])
        xform (apply comp (map (partial map) fns)) ;;create transducer from
                                                   ;;collection of functions
        in-chan  (async/chan 1)
        out-chan (async/chan 1)]

    (async/onto-chan in-chan pages)
    (async/<!! (async/pipeline 4 out-chan xform in-chan))))