Throttle Functions with core.async

The number of possible executions of a function should be throttled. So after calling a function, any repeated call should be ignored within a time period. If there where calls in the meantime, the last one should be executed after the time period.

Here's my approach with core.async. The problem here is, that additional calls are summing up in the channel c. I'd need a channel with only one position inside, which will be overridden by put! everytime.

(defn throttle [f time]
  (let [c (chan 1)]
    (go-loop []
      (apply f (<! c))
      (<! (timeout time))
      (recur))
    (fn [& args]
      (put! c (if args args [])))))

usage:

(def throttled (throttle #(print %) 4000))
(doseq [x (range 10)]
   (throttled x))

; 0
;... after 4 seconds
; 9

Does anyone have an idea how to fix this?

Solution

(defn throttle [f time]
  (let [c (chan (sliding-buffer 1))]
    (go-loop []
      (apply f (<! c))
      (<! (timeout time))
      (recur))
    (fn [& args]
      (put! c (or args [])))))

To solve your channel question you can use a chan with a sliding buffer:

user> (require '[clojure.core.async :as async])
nil
user> (def c (async/chan (async/sliding-buffer 1)))
#'user/c
user> (async/>!! c 1)
true
user> (async/>!! c 2)
true
user> (async/>!! c 3)
true
user> (async/<!! c)
3

that way only the last value put into the channel will be computed at the next interval.