clojure.core ports

This page has all the original transducers in clojure.core implemented as conduits.

(ns my.lib
  (:refer-clojure :exclude [await))
  (:require [com.hypirion.conduit :refer :all])))

(defn mapping
  [f]
  (conduit
   (while true
     (yield (f (await))))))

(defn mapping-indexed
  [f]
  (conduit
   (loop [i 0]
     (yield (f i (await)))
     (recur (inc i)))))

(defn filtering
  [pred]
  (conduit
   (while true
     (let [val (await)]
       (when (pred val)
         (yield val))))))

(defn taking-while
  [pred]
  (conduit
   (loop []
     (let [val (await)]
       (when (pred val)
         (yield val)
         (recur))))))

(defn taking
  [n]
  (conduit
   (dotimes [_ n]
     (yield (await)))))

(defn taking-nth
  [n]
  (conduit
   (while true
     (yield (await))
     (dotimes [_ (- n 1)]
       (await)))))

(defn dropping
  [n]
  (conduit
   (dotimes [_ n]
     (await))
   (while true
     (yield (await)))))

(defn dropping-while
  [pred]
  (conduit
   (loop [v (await)]
     (if (pred v)
       (recur (await))
       (yield v)))
   (while true
     (yield (await)))))

(def catting
  (conduit
   (while true
     (doseq [val (await)]
       (yield val)))))

(defn mapcatting
  [f]
  (conduit
   (while true
     (doseq [val (f (await))]
       (yield val)))))

(def deduping
  (conduit
   (loop [old ::none]
     (let [new (await)]
       (when-not (= old new)
         (yield new))
       (recur new)))))

(defn replacing
  [smap]
  (conduit
   (while true
     (let [val (await)]
       (yield (get smap val val))))))

(defn keeping
  [f]
  (conduit
   (while true
     (let [v (f (await))]
       (when-not (nil? v)
         (yield v))))))

(defn keeping-indexed
  [f]
  (conduit
   (loop [i 0]
     (let [v (f i (await))]
       (when-not (nil? v)
         (yield v)))
     (recur (inc i)))))

(def distincting
  (conduit
   (loop [seen #{}]
     (let [val (await)]
       (if-not (contains? seen val)
         (do (yield val)
             (recur (conj seen val)))
         (recur seen))))))

(defn random-sampling
  [prob]
  (conduit
   (while true
     (let [val (await)]
       (when (< (rand) prob)
         (yield val))))))

(defn interposing
  [sep]
  (conduit
   (yield (await))
   (loop []
     (let [val (await)]
       (yield sep)
       (yield val)
       (recur)))))

(defn partitioning-all
  [n]
  (conduit
   (loop [vs [(await)]]
     (if (= n (count vs))
       (do (yield vs)
           (recur [(await)]))
       (if-let-await! v
         (recur (conj vs v))
         (yield vs))))))

(defn partitioning-by
  [f]
  (conduit
   (let [first-val (await)]
     (loop [vs [first-val]
            to-cmp (f first-val)]
       (if-let-await! v
         (let [new-to-cmp (f v)]
             (if (= to-cmp new-to-cmp)
               (recur (conj vs v) to-cmp)
               (do (yield vs)
                   (recur [v] new-to-cmp))))
         ;; no more values, so flush out vs and exit
         (yield vs))))))