scratch.clj

(defn async-iteration [from-process
                       to-process
                       output-channel
                       & {:keys [values-selector some? init]
                          :or {values-selector vector
                               some? some?}}]
  (async/go
    (async/>! to-process init)
    (loop [value (async/<! from-process)]
      (if (some? value)
        (do
          (doseq [item (values-selector value)]
            (async/>! output-channel item))
          (async/>! value to-process))
        (async/close! output-channel)))))

(defn iteration
  [step!
   & {:keys [vsf kf some? initk bufsize xf]
      :or {vsf vector
           kf identity
           some? some?
           initk nil
           bufsize 1
           xf identity}}]
  (let [a (async/chan)
        b (async/chan)
        c (async/chan bufsize xf)]
    (async-iteration a b c :vsf vsf :some? some? :initk initk)
    (go
      (loop []
        (let [ret (<! (step! (async/<! b)))]
          (when-let [k (and (some? ret)
                            (kf ret))]
            (async/>! a k)
            (recur))))
      (close! ch))
    c))