Scoping transducer for pipelining sub-structures without losing the container.

gistfile1.txt

(require '[clojure.core.async :as async :refer [chan <!! put!]])

(defn save-to [c save-f]
  (fn [rf] 
    (fn
      ([] (rf))
      ([acc] (rf acc))
      ([acc x]
       (put! c (save-f x))
       (rf acc x)))))

(defn restore-from [c restore-f]
  (fn [rf]
    (fn 
      ([] (rf))
      ([acc] (rf acc))
      ([acc x]
       (rf acc (restore-f x (<!! c)))))))


(defn scope [save putback xform]
  (let [c (chan)]
    (comp 
     (save-to c save)
     xform
     (restore-from c putback))))


(def m [{:a 1 :b 10} {:a 2 :b 20} {:a 3 :b 30}])

(def xt (scope #(select-keys % [:a]) 
               #(merge % %2)
               (comp (map :b) 
                     (map inc)
                     (map #(conj {} [:b %])))))
                             
(into [] xt m)

;; => [{:b 11, :a 1} {:b 21, :a 2} {:b 31, :a 3}]

(sequence xt m)

;; => ({:b 11, :a 1} {:b 21, :a 2} {:b 31, :a 3})