clojure core.async - consume until some timeout

async.clj

(defn consume-until-timeout
  [ms ch f init & args]
  (let [tmt (timeout ms)]
    (go (loop [ret init]
          (if-let [v (first (alts! [ch tmt]))]
            (recur (apply f ret [v]))
            ret)))))


(comment
  (let [c (chan)]
    (doseq [i (range 1 7)]
      (go (>! c (do
                  (Thread/sleep (* i 1000))
                  i))))
    (<!! (consume-until-timeout 3000 c conj []))))


(comment
  (let [c (chan)]
    (doseq [i (range 1 7)]
      (go (>! c (do
                  (Thread/sleep (* i 1000))
                  i))))
    (consume-until-timeout 6100 c prn [])))