iteration async

iteration.clj

;; needs a refer-clojure exclude on iteration

(defn iteration
  "returns a channel of items given step!, a function of some (opaque continuation data) k
   step! calls can get bufsize ahead of consumption (asynchronously)

   step! - fn of k/nil to chan yielding (opaque) 'ret'

   :bufsize - buffer this many step! calls, default 1
   :some? - fn of ret -> truthy, indicating there is a value
           will not call vsf/kf nor continue when false
           default some?
   :vsf - fn of ret -> vs, a collection of values produced by one step!, default c.c/vector
   :kf - fn of ret -> next-k or nil (will not continue), default identity
   :initk - the first value passed to step!, default nil

   it is presumed that step! with non-initk is unreproducible/non-idempotent
   if step! with initk is unreproducible, it is on the consumer to not consume twice"
  [step!
   & {:keys [vsf kf some? initk bufsize]
      :or {vsf vector
           kf identity
           some? some?
           initk nil
           bufsize 1}}]
  (let [bufch (chan bufsize)
        ch (chan 100 cat)]
    (pipe bufch ch)
    (go
     (loop [retc (step! initk)]
       (let [ret (<! retc)]
         (when-some [k (and (some? ret)
                            (>! bufch (vsf ret))
                            (kf ret))]
           (recur (step! k)))))
     (close! ch))
    ch))