explicit_state_transducers.cljc

(ns explicit-state-transducers)

(defn take-with-explicit-state
  "A stateless rewrite of the `c.c/take` transducer that requires its state to
  be provided as part of each input"
  [n]
  (fn [rf]
    (fn
      ([] (rf))
      ([result] (rf result))
      ([result {:keys [state value] :or {state n}}]
        (let [state' (dec state)
              result (if (pos? state)
                       (rf result {:state state' :value value})
                       result)]
          (if (not (pos? state'))
            (ensure-reduced result)
            result))))))

(defn drop-with-explicit-state
  "A stateless rewrite of the `c.c/drop` transducer that requires its state to
  be provided as part of each input"
  [n]
  (fn [rf]
    (fn
      ([] (rf))
      ([result] (rf result))
      ([result {:keys [state value] :or {state n}}]
       (if (pos? state)
         (rf result {:state (dec state)})
         (rf result {:state state :value value}))))))

(defn with-implicit-state
  "Takes an `init-state` function and a stateless transducer that must consume
  and emit maps with `:state` and `:value` keys. Returns a new stateful
  transducer whose state is managed using the functions returned by
  `init-state`. The `init-state` function is called when the transducer is
  initialized with a reducing function as part of a new transducing process.
  The function's return value must be a vector of three functions in the
  following order: `initialized?` should return a boolean indicating whether or
  not the state has yet been initialized by the transducer; `get-state` should
  return the state; and `set-state!` should take a single argument and reset
  the state to the given value."
  [init-state xform]
  (fn [rf]
    (let [[initialized? get-state set-state!] (init-state)
          rf' (fn
                ([] (rf))
                ([result] (rf result))
                ([result input]
                 (if (contains? input :state)
                   (set-state! (:state input)))
                 (if (contains? input :value)
                   (rf result (:value input))
                   result)))
          f (xform rf')]
      (fn
        ([] (f))
        ([result] (f result))
        ([result input]
          (f result (if (initialized?)
                      {:state (get-state) :value input}
                      {:value input})))))))

(defn with-hidden-state
  "Takes a stateless transducer that must consume and emit maps with `:state`
  and `:value` keys. Returns a new stateful transducer whose state is hidden."
  [xform]
  (let [init-state (fn []
                     (let [vol (volatile! {})]
                       [#(contains? @vol :state)
                        #(:state @vol)
                        #(vreset! vol {:state %})]))]
    (with-implicit-state init-state xform)))

(defn with-external-atom-state
  "Takes an atom and a stateless transducer that must consume and emit maps
  with `:state` and `:value` keys. Returns a new stateful transducer whose
  state is held in the given atom."
  [a xform]
  (let [init-state (fn []
                     [#(contains? @a :state)
                      #(:state @a)
                      #(reset! a {:state %})])]
    (with-implicit-state init-state xform)))

(comment
  (into [] (with-hidden-state (take-with-explicit-state 5)) (range))
  ; [0 1 2 3 4]
  (into [] (with-hidden-state (drop-with-explicit-state 5)) (range 10))
  ; [5 6 7 8 9]
  (def a (atom {}))
  (into [] (with-external-atom-state a (take-with-explicit-state 5)) [0 1])
  ; [0 1]
  @a
  ; {:state 3}
  (into [] (with-external-atom-state a (take-with-explicit-state 5)) [2 3 4 5 6 7 8 9])
  ; [2 3 4]
  @a
  ; {:state 0}
)