Try to implement chunks with core.async

example.clj

(ns cltrace.example
  (:require [clojure.core.async :refer [<! >! >!! chan go-loop]]))

(def queue (atom []))

(defn process-chan [in]
  (let [out (chan)]
    (go-loop [item (<! in)]
      (swap! queue conj item)
      (when (> (count @queue) 3)
        (let [records @queue
              [head rest] (split-at 3 records)]
          (if (compare-and-set! queue records (apply vector rest))
            (>! out head))))
      (recur (<! in)))
    out))

(defn bind-dummy-printer [output]
  "Dummy printer for a/chan."
  (go-loop [record (<! output)]
    (println "Chunk were received: " record)
    (recur (<! output))))

(defn example-main []
  "Setup example."
  (def input (chan))
  (let [output (process-chan input)]
    (bind-dummy-printer output)
    (map #(>!! input (str %)) (range 10))))