core.async flow limiter

gistfile1.clj

(ns dball.util.async
  (:require [clojure.core.async :refer [go-loop >! <! timeout close! chan dropping-buffer >!! alts!]]
            [clojure.tools.logging :as log])
  (:import [java.util.concurrent Executors ScheduledExecutorService TimeUnit]))

(defn ^ScheduledExecutorService scheduler
  []
  (Executors/newSingleThreadScheduledExecutor))

(defn ticker-chan
  [scheduler period-ms]
  (let [ticks (chan (dropping-buffer 1))
        task-fn (atom nil)
        tick-fn #(when-not (>!! ticks :tick)
                   (.cancel @task-fn))]
    (reset! task-fn (.scheduleAtFixedRate scheduler tick-fn period-ms period-ms
                                          TimeUnit/MILLISECONDS))
    ticks))

(defn flow-limiter
  ([from to min-period-ms]
     (flow-limiter from to min-period-ms true))
  ([from to min-period-ms close?]
     (let [scheduler (scheduler)]
       (go-loop [state :ready
                 ticks nil]
         (condp = state
           :ready
           (if-some [value (<! from)]
             (do
               (>! to value)
               (recur :pause (ticker-chan scheduler min-period-ms)))
             (recur :shutdown nil))

           :pause
           (let [[value task] (alts! [from ticks] :priority true)]
             (condp = task
               from
               (if (some? value)
                 (do
                   (<! ticks)
                   (>! to value)
                   (recur :pause ticks))
                 (do
                   (close! ticks)
                   (recur :shutdown nil)))

               ticks
               (do
                 (close! ticks)
                 (recur :ready nil))))

           :shutdown
           (do
             (when close? (close! to))
             (.shutdown scheduler)))))))