[v4] Add scan utils

src/taoensso/carmine_v4.clj

@@ -308,6 +308,49 @@
             ;; TODO Try publish error message?
             ))))))
 
+;;;;
+
+(defn reduce-scan
+  "For use with `scan`, `hscan`, `zscan`, etc.
+   Takes:
+     - (fn rf      [acc scan-result]) -> next accumulator
+     - (fn scan-fn [cursor]) -> next scan result"
+  ([rf          scan-fn] (reduce-scan rf nil scan-fn))
+  ([rf acc-init scan-fn]
+   (loop [cursor "0" acc acc-init]
+     (let [[next-cursor in] (scan-fn cursor)]
+       (if (= next-cursor "0")
+         (rf acc in)
+         (let [result (rf acc in)]
+           (if (reduced? result)
+             @result
+             (recur next-cursor result))))))))
+
+(comment
+  (reduce-scan (fn rf [acc in] (into acc in))
+    [] (fn scan-fn [cursor] (wcar {} (scan cursor)))))
+
+(defn reduce-hscan
+  "Like `reduce-scan` but:
+    - `rf` is (fn [acc k v]), as in `reduce-kv`.
+    - `rf` will never be called with the same key twice
+      (i.e. automatically de-duplicates elements)."
+  ([rf          scan-fn] (reduce-hscan rf nil scan-fn))
+  ([rf acc-init scan-fn]
+   (let [keys-seen_ (volatile! (transient #{}))]
+     (reduce-scan
+       (fn wrapped-rf [acc kvs]
+         (enc/reduce-kvs
+           (fn [acc k v]
+             (if (contains? @keys-seen_ k)
+               acc
+               (do
+                 (vswap! keys-seen_ conj! k)
+                 (enc/convey-reduced (rf acc k v)))))
+           acc
+           kvs))
+       acc-init scan-fn))))
+
 ;;;; Scratch
 
 ;; TODO For command docstrings