pipeline.clj

(defn pipeline [pipeline-config]
  {:pre [(even? (count pipeline-config))]}
  (let [x (->> (partition-all 2 pipeline-config)
               (map (fn [[buf-size fun]]
                      {:in (async/chan buf-size)
                       :buffer-size buf-size
                       :fun fun}))
               (partition-all 2 1)
               (map (fn [[step {from-out :in :as o}]]
                      (assoc step
                        :worker-count (or (and o (min (:buffer-size step)
                                                      (:buffer-size o)))
                                          (:buffer-size step))
                        :out (or from-out (async/chan)))))
               (mapv (fn [{:keys [worker-count in out fun] :as step}]
                       (assoc step
                         :workers
                         (let [n-chans (repeatedly
                                        worker-count
                                        (fn []
                                          (async/mapcat< (fn [x] (try (fun x)
                                                                    (catch Exception e
                                                                      (println e)
                                                                      [])))
                                                         in)))]
                           (async/pipe (async/merge n-chans) out))))))]
    {:in (:in (first x))
     :out (:out (last x))
     :network x}))