                                     partition-size
                                     reducers/cat
                                     reducers/append!
-                                    (reducers/map direct-proc-fn in-vec))]
+                                    (reducers/map (fn [in-obj] (direct-proc-fn (pre-in-fn in-obj))) in-vec))]
                         (out-fn obj))))
                 (.endsWith scenario "-simple-pmap")
                   (let [pmap-proc (create-simple-pmap-processor direct-proc-fn collection-size out-fn)]
                     (fn []
-                      (pmap-proc in-data)
-                      (.inc in-cntr)))
+                      (pmap-proc (pre-in-fn in-data))))
                 (.endsWith scenario "-simple-reducers-map")
                   (let [red-proc (create-simple-reducers-map-processor direct-proc-fn collection-size partition-size out-fn)]
                     (fn []
-                      (red-proc in-data)
-                      (.inc in-cntr)))
+                      (red-proc (pre-in-fn in-data))))
                 (.endsWith scenario "-async-pipeline")
                   (if
                     (and

    proc-fns))

(defn prepare-in-fn
  [options scenario in-data ^Counter in-cntr out-fn in-chan pipeline queue-size]
  (let [batch-delay (:batch-delay options)
        batch-size (:batch-size options)
        create-batched-in-fn (fn [in-fn]
                               (fn []
                                 (doseq [i (repeat batch-size 0)]
                                   (in-fn in-data)
                                   (.inc in-cntr))
                                 (sleep batch-delay)))
        create-flood-in-fn (fn [in-fn]
                             (fn []
                               (in-fn in-data)
                               (.inc in-cntr)))
        direct-proc-fn (create-direct-proc-fn scenario)
        collection-size (:collection-size options)
        partition-size (:partition-size options)
        in-fn (cond
                (.endsWith scenario "-direct")
                  (fn []
                    (out-fn (direct-proc-fn in-data)))
                (.endsWith scenario "-direct-pmap")
                  (let [d (repeat collection-size in-data)]
                    (doall d)
                    (fn []
                      (doseq [obj (pmap direct-proc-fn d)]
                        (out-fn obj))))
                (.endsWith scenario "-direct-reducers-map")
                  (let [in-vec (vec (repeat collection-size in-data))]

          (println "Starting async-pipeline-go input creation.")
          (async/go
            (loop []
              (async/>! in-chan (pre-in-fn in-data))
              (recur)))
          (sleep 130000))
        (do
