Transduction with a user-controllable state buffer which is flushed on transduction finalization. Useful for writing transducers which sometimes process items differently based on a look-ahead condition.

buffered_reduce.cljc

(defn buffered-reduce
  "Return a transducer wrapping `reduction-fn` which holds on to an internal
  buffer volatile (initialized with nil) which will be flushed (via `reduce`)
  when the transducer is closed.
  
  The `reduction-fn` should accept a combining function, an accumulating value,
  a current value, and a buffer volatile. This function should return a new
  accumulating value via 0, 1 or many calls to `(xf r new-value)`. It may also
  manipulate the contents of the volatile for its own use. If anything is found
  in the buffer when the reduction is complete, its items will be added to the
  reduction before closing."
  [reduction-fn]
  (fn [xf]
    (let [buffer (volatile! nil)]
      (fn
        ([] (xf))
        ([r] (let [remaining @buffer]
               (vreset! buffer nil)
               (xf (reduce xf r remaining))))
        ([r v]
         (reduction-fn xf r v buffer))))))