Chunked transducer

chunked.clj

; there are some obvious micro optimizations but I left them out for clarity
; the relative ordering of read and writes with volatile and plain array should be thread-safe (if not, point it out)
(defn chunked
  ([] (chunked 32))
  ([n]
    (fn [f1]
      (let [xs (object-array n)
            nv (volatile! 0)
            flush (fn [acc]
                    (loop [i 0 acc acc]
                      (if (< i @nv) ; read of the volatile BEFORE read of the array
                        (let [acc (f1 acc (aget xs i))]
                          (if (reduced? acc)
                            acc
                            (recur (inc i) acc)))
                        acc)))]
        (fn 
          ([] (f1))
          ([acc] (f1 (flush acc)))
          ([acc x]
            (if (< @nv n)
              (do
                (aset xs @nv x)
                (vswap! nv inc) ; write of the volatile AFTER write of the array
                acc)
              (let [acc (flush acc)]
                (vreset! nv 0)
                acc))))))))