using core.async to combine data into chunks with a timeout

chunker.clj

(ns noisesmith.chunker
  (:require [clojure.core.async :refer [alts! <! >! timeout go-loop]]))

(defn launch-chunks
  [chunk-size time-out rq-chan result-chan]
  (go-loop
   [acc []]
   (let [[res c] (alts! [(timeout time-out)
                          rq-chan])
         fresh (= c rq-chan)
         timed-out (and (not fresh) (seq acc))
         acc (if fresh
               (conj acc res)
               acc)]
     (recur
      (if (or (> (count acc) chunk-size)
              timed-out)
        (do (>! result-chan acc)
            [])
        acc)))))

repl session:

kingfisher.core=> (require '[clojure.core.async :as >])
nil
kingfisher.core=> (def chan-a (>/chan))
#'kingfisher.core/chan-a
kingfisher.core=> (def chan-b (>/chan))
#'kingfisher.core/chan-b
kingfisher.core=> (>/go-loop [] (let [x (>/<! chan-b)] (println (str "read " x))) (recur))
#object[clojure.core.async.impl.channels.ManyToManyChannel 0x6a6203db "clojure.core.async.impl.channels.ManyToManyChannel@6a6203db"]
kingfisher.core=> (>/>!! chan-b :foo)
read :foo
true
kingfisher.core=> (load-file "/tmp/chunker.clj")
#'noisesmith.chunker/launch-chunks
kingfisher.core=> (noisesmith.chunker/launch-chunks 3 10000 chan-a chan-b)
#object[clojure.core.async.impl.channels.ManyToManyChannel 0x4f3b5399 "clojure.core.async.impl.channels.ManyToManyChannel@4f3b5399"]
kingfisher.core=> (>/>!! chan-a 1)
true
kingfisher.core=> read [1]


kingfisher.core=> (dotimes [i 10] (>/>!! chan-a i))
read [0 1 2 3]
read [4 5 6 7]
nil
kingfisher.core=> read [8 9]