transducers-stateful-map+filter.clj

(defn stateful-map
  "Returns a stateful transducer similar to `clojure.core/map-indexed` that
  transforms each item with `(f state <the item>)` where `state` is built by
  applying `state-building-fn` to the previous state (starting with `init`) and
  the transformed item.

  Ex., re-implementing map-indexed:
  ```clojure
  (sequence (stateful-map vector (fn build-state [idx _] (if idx (inc idx) 5))) \"abc\")
  ; => ([nil \\a] [5 \\b] [6 \\c])
  (sequence (stateful-map (fn stateful-map [seen n] (if (seen n) :duplicate n)) conj #{})
            [1 2 1 3 4 2 2 5])
  ; => (1 2 :duplicate 3 4 :duplicate :duplicate 5)
  ```

  See also [[stateful-filter]]."
  ([f state-building-fn] (stateful-map f state-building-fn nil))
  ([f state-building-fn init]
   (fn [rf]
     (let [state (volatile! init)]
       (fn
         ([] (rf))
         ([result] (vreset! state nil) (rf result))
         ([result input]
          (let [out (f @state input)]
            (vswap! state state-building-fn out)
            (rf result out))))))))

(defn stateful-filter
  "A transducer similar to `(filter pred)` but the `pred`-icate is invoked with
  a state and the current item, where state is built from previously filtered
  items (i.e from the previous state [starting from `init`] and the
  current item) by the `state-building-fn`.

  Ex.:
  ```clojure
  (sequence (stateful-filter not= (fn build-state [_ prev] prev)) [1 2 2 2 3 4 4 5])
  ; => (1 2 3 4 5)
  ;; Ensure an increasing seq:
  (sequence (stateful-filter < max -1) (shuffle (range 20)))
  ; => (7 9 10 12 16 17 19)

  See also [[stateful-map]].
  ```"
  ([pred state-building-fn] (stateful-filter pred state-building-fn nil))
  ([pred state-building-fn init]
   (fn [rf]
     (let [state (volatile! init)]
       (fn
         ([] (rf))
         ([result] (vreset! state nil) (rf result))
         ([result input]
          (if (pred @state input)
            (do
              (vswap! state state-building-fn input)
              (rf result input))
            result)))))))

Replied it on slack but I'll comment here too for posterity:
This can be generalized with a wrapper of sort:

(defn with-state
  ([trf f]
   (with-state trf f nil))
  ([trf f init]
   (with-state identity trf f init))
  ([txf trf f init]
   (let [state (volatile! init)
         tf (txf trf)]
     (fn [in]
       (let [old @state]
         (vswap! state tf in)
         (f old in))))))

(defn right [x y] y)
(sequence (stateful-filter not= right) [1 2 2 2 3 4 4 5])
(sequence (filter (with-state right not=)) [1 2 2 2 3 4 4 5])

where
tf - transition function
trf - transition reducing function
txf - transition transducer

This works differently for map since it takes the input rather than the output. It may be more semantically correct if you consider the stateless lifecycle function:
[state in] -> [state' out]

Great idea, thanks a lot!

As discovered by others, we could replace the volatile! with the thread-unsafe (which is OK since it will ever only be accessed by a single thread) but faster mutable var:

(let [b (clojure.lang.Box. 3)]
  ; ...
  (set! (. b val) 4))

Created a unified version
https://gist.github.com/bsless/c96d29df2037f728cf731f0848d6e1ab