manifold-transform-mem-leak-exploration.clj

(require '[manifold.stream :as s])


(defn coerce-seq
  "if the page should happen to be an Exception, wrap
   it in a vector so that it can be concatenated to the
   value stream"
  [v]
  (if (sequential? v) v [v]))


(defn concat-seq-s
  "concat a stream of seqs onto a stream of plain values"
  [stream-s]
  (let [out (s/stream)]

    (s/connect-via
      stream-s
      #(s/put-all! out (coerce-seq %))
      out)

    out))


(def rand-v-stream (s/->source (repeatedly #(vector (rand-int 100) (rand-int 100) (rand-int 100)))))

(def oome-stream (s/transform
                   (mapcat coerce-seq)
                   rand-v-stream))

(def good-stream (concat-seq-s rand-v-stream))

@(s/reduce (fn [cnt _]
             (if (> cnt 10000000)
               (reduced cnt)
               (inc cnt)))
           0
           oome-stream)