fun with accumlating window transducer

accumulate.clj

(defn accumulate
  "If an input satisfies ACC-PRED, accumulate it with ACC-FN (defaults to merge)
  All inputs will be passed on normally.
  All accumulated inputs will be merged (using ACC-FN) to following inputs until RESET-PRED.

  The input that satisfies RESET-PRED will receive the accumulated values and the input following it will receive none.

  The default ACC-FN, `merge`, assumes a stream of maps"
  ([acc-pred reset-pred]
   (accumulate acc-pred reset-pred merge))
  ([acc-pred reset-pred acc-fn]
   (accumulate acc-pred reset-pred merge merge))
  ([acc-pred reset-pred acc-fn disperse-fn]
   (fn [xf]
     (let [prev (volatile! nil)
           reset (volatile! false)]
       (fn
         ;; init
         ([] (xf))
         ;; step
         ([result input]
          ;; stored from previous run
          (if @reset
            (do
              (vreset! prev nil)
              (vreset! reset false)))
          (cond
            (acc-pred input)
            (do
              (vswap! prev acc-fn input)
              result)
            (reset-pred input)
            (do
              ;; close the window
              (vreset! reset true)
              ;; emit the trailer
              (xf result (disperse-fn input @prev)))
            :default
            ;; normal input
            (xf result (disperse-fn input @prev))))
         ;; terminate
         ([result]
          (if (and @prev (not @reset))
            (xf result (disperse-fn nil @prev)))
          (xf result)))))))


(defn assoc-if [m k v]
  (if (not-empty v)
    (assoc m k v)
    m))

(into [] (accumulate :hdr :trl merge #(assoc-if %1 :hdr %2) ) [{:a 1} {:hdr true :b 2} {:c 3} {:trl true :d 4} {:e 5}])
;;=> [{:a 1} {:c 3, :hdr {:hdr true, :b 2}} {:trl true, :d 4, :hdr {:hdr true, :b 2}} {:e 5}]

;; with last element a closing window input
(into [] (accumulate #(not (zero? (:n %))) ;;accumulate :n not zero
                     #(zero? (:n %))       ;; reset when :n zero
                     conj                  ;; create a list of :n's
                     #(assoc-if %1 :prev %2)) ;; stuff them into the last thing
      [{:n 0} {:n 1} {:n 2} {:n 3} {:n 0}])
;;=> [{:n 0} {:n 0, :prev ({:n 3} {:n 2} {:n 1})}]


;; with no closing window input
(into [] (accumulate #(not (zero? (:n %))) ;;accumulate :n not zero
                     #(zero? (:n %))       ;; reset when :n zero
                     conj                  ;; create a list of :n's
                     #(assoc-if %1 :prev %2)) ;; stuff them into the last thing
      [{:n 0} {:n 1} {:n 2} {:n 3}])
;;=> [{:n 0} {:prev ({:n 3} {:n 2} {:n 1})}]