a descriptive experiment making a clojure transducer

collate_transducer.clj

(defn collate
 "This is a transducer for collecting things that are connected in series. It
  will store one previous input. The previous input can then be compared to it's
  successor using COMPARES-FN and optionally integrated into a single output
  using COLLATE-FN.

  COMPARES-FN must accept 2 arguments, being 2 inputs (which are in order)
  COLLATE-FN must accept 2 arguments, being an accumulator and an input
  COLLATE-FN is a reducing function. A third arg can be given as the starting
  accumulator.
  "
  ([compares-fn collate-fn]
   (collate compares-fn collate-fn nil))
  ([compares-fn collate-fn accumulator]
   (fn [xf]
     ;; use volatile for performance since this is guaranteed to be on a single
     ;; thread
     (let [prev (volatile! nil)
           session (volatile! nil)]
       ;; a function that abides by the three rules of transducing
       ;; 1) call xf on init
       ;; 2) always return the accumulator (result) during a step (though may muck with inputs)
       ;; 3) call xf when complete
       (fn
         ;; init
         ([] (xf))
         ;; step
         ([result input]
          (let [prior @prev]
            (vreset! prev input)
            (if (nil? prior)
              ;; Here is where we setup an offset of 1 by storing the first thing
              ;; in prev. We need to do this in order to compare two things. We
              ;; do nothing for the first line, just return the
              ;; accumulator (result).
              result
              (if (compares-fn prior input)
                (do
                  (if @session
                    ;; add to session
                    (vreset! session (conj @session input))
                    ;; init session
                    (vreset! session (conj @session prior input)))
                  ;; do nothing. wait for end of session to emit the whole collated session.
                  result)
                (if @session
                  (let [sess @session]
                    (vreset! session nil)
                    ;; emit the whole collated session
                    (if accumulator
                      (xf result (reduce collate-fn accumulator sess))
                      (xf result (reduce collate-fn sess))))
                  ;; this is the normal operation; like a pass though. nothing happens
                  (xf result prior))))))
         ;; completion
         ([result]
          ;; we have reached the end of the list. we need to make up for our
          ;; offset we setup so we call xf with the last element which didn't get
          ;; called in the step function above, and then the actual completing
          ;; function
          (do
            (if @session
              ;; the last element satisfies compare-fn
              (let [sess @session]
                (vreset! session nil)
                ;; emit the whole collated session
                (if accumulator
                  (xf result (reduce collate-fn accumulator sess))
                  (xf result (reduce collate-fn sess))))
              ;; the last element does not satisfy compare-fn so nothing happens
              (xf result @prev))
            ;; required call to completing function
            (xf result))))))))

(comment

  ;; Example 1

  ;; add like numbers if they are connected in series
  (into [] (collate = +) [0 1 2 2 1 0 3 3 3 3 0])
  ;;=> [0 1 4 1 0 12 0]


  ;; Example 2

  ;; collect things with :b values that are connected in series
  (defn has-b [prev current]
    (and (:b prev)
         (:b current)))

  ;; reduce them into an :m value
  (defn put-b-on-m [acc el]
    (update acc :m conj (:b el) ))

  (into [] (collate has-b put-b-on-m {})
        [{:a 1} {:b nil} {:b 1} {:b 2} :z {:b 10 :a 5} :z {:b 3} {:b 4}])
  ;;                      vvvvvvvv      (skipped)         vvvvvvvv
  ;;=>  [{:a 1} {:b nil} {:m (1 2)} :z {:b 10, :a 5} :z {:m (3 4)}]

  )