clojure transducers

transducers.clj

;; https://github.com/green-coder/transducer-exercises
(ns transducers)

;; We use this function instead of `into` for debugging.
;; The reason is that this avoids using transient
;; structures which do not `print` nicely.
(defn slow-into [to xf from]
  (transduce xf conj to from))

(def identity-transducer
  (fn [rf]
    (fn
      ;; 0-ary: identity (initialization? not used by clojure.core/transduce)
      ([] (rf))
      ;; 1-ary: finally
      ([x] (rf x))
      ;; 2-ary: process-element
      ([x y] (rf x y)))))

(defn debug
  ([] (debug 0))
  ([indent] (debug indent ">" "<"))
  ([in out] (debug 0 in out))
  ([indent in out]
   (let [spaces (str/join (repeat indent " "))
         in (str spaces in)
         out (str spaces out)]
     (fn [rf]
       (fn
         ([] (rf))
         ([x] (rf x))
         ([x y]
          (println in y)
          (let [result (rf x y)]
            (println out result)
            result)))))))

(comment
  (slow-into []
             (comp (debug)
                   (debug 2)
                   (debug 4 ">" "<")
                   (debug "      >" "      <")) ; 6-spaces prefix
             (range 3)))

(defn preserving-reduced [f]
  (fn [x y]
    (let [result (f x y)]
      (if (reduced? result)
        (reduced result)
        result))))

(defn beg [n]
  (fn [rf]
    (let [preserving-rf (preserving-reduced rf)]
      (fn
        ([] (rf))
        ([result] (rf result))
        ([result value]
         (reduce preserving-rf
                 result
                 (repeat n value)))))))

(comment
  (def beg-data (list :may :i :beg :your :pardon :?))
  (slow-into []
             (comp (debug 0)
                   (beg 2)
                   (debug 2)
                   (take 3)
                   (debug 4))
             beg-data))

(def cat-data [[1 2 :fish 3] [:heat 4] [5 :sleep 6] [7]])

(def my-cat
  (fn [rf]
    (let [preserving-rf (preserving-reduced rf)]
      (fn
        ([] (rf))
        ([result] (rf result))
        ([result value]
         (reduce #(cond (#{:fish :heat} %2) %1
                        (= :sleep %2) (reduced (reduced %1))
                        :else (preserving-rf %1 %2)) result value))))))

(comment
  (into [] my-cat cat-data)
  ;;=>
  [1 2 3 4 5]

  (into [] (comp (take 2)
                 my-cat)
        cat-data)
  ;;=>
  [1 2 3 4]

  (into [] (comp my-cat
                 (take 2))
        cat-data)
  ;;=>
  [1 2]

  (slow-into [] (comp (debug 0)
                      my-cat           ; try replacing it with `cat` and compare
                      (debug 2)
                      (take 2)
                      (debug 4))
             cat-data)
  ;;=>
  [1 2]
  ;; > [1 2 :fish 3]
  ;;   > 1
  ;;     > 1
  ;;     < [1]
  ;;   < [1]
  ;;   > 2
  ;;     > 2
  ;;     < [1 2]
  ;;   < #reduced[{:status :ready, :val [1 2]} 0x6197a6fe]
  ;; < #reduced[{:status :ready, :val [1 2]} 0x6197a6fe]
  )

(defn a-d-d
  "Implement a transducer that daydream during a number of elements. While in
  the daydream state, it buffers its input. When it stops daydreaming, it
  processes all of its buffer as a batch, then daydreams again."
  [n]
  (fn [rf]
    (let [preserving-rf (preserving-reduced rf)
          buffer (volatile! [])
          flush (fn [result]
                  (let [result' (reduce preserving-rf result @buffer)]
                    (vreset! buffer [])
                    result'))]
      (fn
        ([] (rf))
        ([result]
         ;; first flush the buffer, then finalize the inner transducer
         (rf (flush result)))
        ([result value]
         (vswap! buffer conj value)
         (if (< (count @buffer) n)
           result
           (flush result)))))))

(comment
  (into [] (a-d-d 3) (range 10))
  ;; =>
  [0 1 2 3 4 5 6 7 8 9]

  (slow-into [] (comp (debug 0)
                      (a-d-d 3)
                      (debug 2))
             (range 10))

  ;; check that it terminates early when reduced
  (slow-into [] (comp (debug 0)
                      (a-d-d 3)
                      (debug 2)
                      (take 5))
             (range 10)))

(defn my-paritition-all [n]
  (fn [rf]
    (let [buffer (volatile! [])
          flush (fn [result]
                  (if-let [b (seq @buffer)]
                    (let [result' (rf result @buffer)]
                      (vreset! buffer [])
                      result')
                    result))]
      (fn
        ([] (rf))
        ([result]
         ;; first flush the buffer, then finalize the inner transducer
         (rf (flush result)))
        ([result value]
         (vswap! buffer conj value)
         (if (< (count @buffer) n)
           result
           (flush result)))))))

(comment
  (into [] (my-paritition-all 3) (range 10))
  ;;=>
  [[0 1 2] [3 4 5] [6 7 8] [9]]

  (slow-into [] (comp (debug 0)
                      (my-paritition-all 3)
                      (debug 2)
                      (take 2))
             (range 10))
  ;;=>
  [[0 1 2] [3 4 5]])

(defn serieduce
  ([f]
   (fn [rf]
     (let [empty-sentinel (gensym)
           state (volatile! empty-sentinel)]
       (fn
         ([] (rf))
         ([result] (rf result))
         ([result value]
          (rf result (vswap! state
                             #(if (= empty-sentinel %1)
                                %2
                                (f %1 %2))
                             value)))))))
  ([f x]
   (fn [rf]
     (let [state (volatile! x)]
       (fn
         ([] (rf))
         ([result] (rf result))
         ([result value]
          (rf result (vswap! state f value))))))))

(comment
  (into [] (serieduce conj [1 2]) (range 3 6))
  ;;=>
  [[1 2 3] [1 2 3 4] [1 2 3 4 5]]

  (into [] (serieduce +) (range 5))
  ;;=>
  [0 1 3 6 10])