Example of exception-handling in Manifold with transducers

manifold-transducer-ex-handling.clj

(in-ns 'user)
(require '[manifold.stream :as s])

(defn flaky
  [x]
  (if (< (rand-int 10) 2)
    (throw (Exception. "flaky"))
    x))

(defn reliable-add-sentinel
  [x]
  (try
    (flaky x)
    (catch Exception _
      :got-an-exception)))

(defn stream-ex-test
  [xform]
  (let [s (s/stream 10 xform)
        results {:put-all       (s/put-all! s (range 100))
                 :consume-async (s/consume-async #(do (println %) true) s)}]
    (s/close! s)
    results))

(stream-ex-test (clojure.core/map flaky))
(stream-ex-test (clojure.core/map reliable-add-sentinel))

;; drop bad vals entirely
(stream-ex-test (comp (clojure.core/map reliable-add-sentinel)
                   (remove #(= :got-an-exception %))))