Different ways of producing the same running diff with lazy-seqs, core.async or transducers.

runningdiff.clj

;; different ways of producing the same running diff with lazy-seqs, core.async or transducers.
;; the running diff injects a :diff key to each of the maps in the list. The :diff is the difference
;; between the current :value and the :value of the previously seen entry with the same :id.
;; it can be easily changed to process id-related items in a different way.

(def entries [{:id 1 :value 88}
              {:id 2 :value 99}
              {:id 1 :value 98}
              {:id 3 :value 5}
              {:id 2 :value 100}
              {:id 3 :value 10}
              {:id 1 :value 150}])

;; standard lazy seqs. Avoids atoms (or other state) by passing 
;; the previously seen items each iteration.
(letfn [(rdiff [xs seen]
          (lazy-seq
            ((fn [[{:keys [id value] :as x} :as xs] seen]
               (when-let [s (seq xs)]
                 (cons (assoc x :diff (- value (get seen id value)))
                       (rdiff (rest s) (assoc seen id value)))))
             xs seen)))]
  (rdiff entries {}))

;; core.async version is just a modification of the lazy seq version to pull
;; from a channel instead of (first xs).
(letfn [(rdiff [c seen]
          (lazy-seq
            ((fn [[{:keys [id value] :as x} :as xs] seen]
               (when x
                 (cons (assoc x :diff (- value (get seen id value)))
                       (rdiff c (assoc seen id value)))))
             (<!! c) seen)))]
  (rdiff (chan) {}))

;; The good thing of the xducer version is that it works idendepently from
;; the transport (as per xducers design). 
(def xdiff
  (fn [rf]
    (let [seen (volatile! {})]
      (fn
        ([] (rf))
        ([result] (rf result))
        ([result {:keys [id value] :as x}]
         (let [diff (- value (get @seen id value))]
           (vswap! seen assoc id value)
           (rf result (assoc x :diff diff))))))))

;; (sequence (xdiff) entries)
;; ({:id 1, :value 88, :diff 0} {:id 2, :value 99, :diff 0} {:id 1, :value 98, :diff 10} {:id 3, :value 5, :diff 0} {:id 2, :value 100, :diff 1} {:id 3, :value 10, :diff 5} {:id 1, :value 150, :diff 52})