core.async retry

retry.clj

(ns com.example.util
  (:import [clojure.core.async.impl.channels ManyToManyChannel])
  (:require [clojure.core.async :as async]))


(defn retry
  [f & {:keys [out-ch retries delay]}]
  {:pre [(and (integer? retries) (pos? retries))
         (integer? delay)
         (instance? ManyToManyChannel out-ch)]}
  (let [stop-ch (async/chan 1)]
    (async/go-loop [n 0]
      (when (< n retries)
        (let [result (async/<! (async/thread
                                 (try
                                   (f)
                                   (catch Exception ex
                                     {::fail ex}))))]
          (if (::fail result)
            (async/alt!
              (async/timeout delay) (recur (inc n))
              stop-ch :stopped)
            (do
              (if (some? result) (async/>! out-ch result))
              (async/close! stop-ch)
              (async/close! out-ch))))))
    stop-ch))