defqueue.clj

defqueue.clj

(require '[clojure.core.async :as async])
(require '[mount.core :as mount])

(defn work-queue [buffer max-retries parallelism]
  (let [chan (async/chan buffer)]
    (dotimes [_ parallelism]
      (async/go-loop []
        (when-some [i (async/<! chan)]
          (let [{:keys [f retries]} (if (map? i) i {:retries 0 :f i})]
            (try (f)
                 (catch Exception e
                   (if (> max-retries retries)
                     (async/put! chan {:f f :retries (inc retries)})
                     (.printStackTrace e)))))
          (recur))))
    chan))

(defmacro defqueue [symbol buffer max-retries parallelism]
  `(let [sym# ~symbol]
     (mount/defstate sym#
       :start (work-queue ~buffer ~max-retries ~parallelism)
       :stop (async/close! sym#))))

(defqueue downloader 100 2 4)