xfs.clj

(defn map-prev
  "maps f over values. f is a function of two arguments, the previous and current value, which is mapped over each value. previous value is initially nil"
  ([f]
   (let [vold (volatile! nil)]
     (map (fn [x]
            (let [old @vold]
              (vreset! vold (f old x)))))))
  ([f xs]
   (map f (cons nil xs) xs)))


(def ch1
  (chan 1 (map-prev (fn [prev x]
                      (if (some? prev)
                        (max prev x)
                        x)))))

(go-loop []
  (when-some [x (<! ch1)]
    (println "max:" x)
    (recur)))

(a/onto-chan!! ch1 [1 5 5 2 3 10 11 0 100])
;;max: 1
;;max: 5
;;max: 5
;;max: 5
;;max: 5
;;max: 10
;;max: 11
;;max: 11
;;max: 100




(defn assoc-xf
  "returns transducer which calls lookup on each map it encounters, subjects that
  transducer xf, and assoc's its result in the map as new-k"
  [lookup xf new-k]
  
  (let [next! (xf (fn [_ x] x))]
    (map (fn [m]
           (assoc m new-k (next! nil (lookup m)))))))


(def ch2
  (chan 1
        (comp
          (assoc-xf
            :x
            (map-prev (fn [prev x] (if (some? prev) (max prev x) x)))
            :max-x)
          (assoc-xf
            :x
            (map-prev (fn [prev x] (if (some? prev) (min prev x) x)))
            :min-x))))

(go-loop []
  (when-some [x (<! ch2)]
    (println "value:" x)
    (recur)))


(a/onto-chan!! ch2 [{:x 1} {:x 3} {:x 6} {:x 2} {:x -11} {:x 20}])

;;value: {:x 1, :max-x 1, :min-x 1}
;;value: {:x 3, :max-x 3, :min-x 1}
;;value: {:x 6, :max-x 6, :min-x 1}
;;value: {:x 2, :max-x 6, :min-x 1}
;;value: {:x -11, :max-x 6, :min-x -11}
;;value: {:x 20, :max-x 20, :min-x -11}