Add delay in clojure.core.async

Question

When using clojure.core.async, is there a way to have one channel wait for the first item to be put on it, then wait some small amount of time, then get all the items currently on the channel (that could have arrived while waiting) and get all of them without blocking?

I.e. is there a way to implement get-available-items:

(defn delayer [ch ch2]
  (go (loop []
        (when-let [v (<! ch)]
          (<! (timeout 500))
          (let [vs (get-available-items ch)
                items (cons v vs)]
            (>! ch2 items))
          (recur)))))

Basically, something like BlockingQueue.drain in Java.

Arthur Ulfeldt · Accepted Answer · 2015-02-04 23:09:14Z

6

There are plans to offer this feature with channels, though for now you can check for the presence of something on a channel with:

(alts!! [my-chan] :default :nothing-immediately-in-chan)

by iterating that you can drain a channel without blocking.

PS: extra thanks to tbaldridge and julianlevis on #clojure for helping with this one

edited Feb 4, 2015 at 23:09

answered Feb 4, 2015 at 17:45

Arthur Ulfeldt

91.7k30 gold badges205 silver badges285 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

DanLebrero · Accepted Answer · 2015-02-05 02:05:56Z

You can just alt on the same timeout channel until you run out of "waiting time", collecting any incoming values meanwhile.

These seems to work:

(require '[clojure.core.async :as a :refer [<! >! go chan]])

(defn delayer [in out]
  (a/go-loop []
    (when-let [v (<! in)]
      (loop [batch [v] timeout-ch (a/timeout 500)]
        (let [[v ch] (a/alts! [in timeout-ch])]
          (if (= in ch)
            (recur (conj batch v) timeout-ch)
            (>! out batch))))
      (recur))))

Notice that we create the timeout channel just once and we reuse it. A simple test to prove that it works:

(def out (chan))
(def in (chan))

(delayer in out)

; print batches as soon as available
(a/go-loop []
  (>pprint (str (java.util.Date.) (<! out)))
  (recur))

 ; put a value every 100 millis
(a/go-loop [i 100]
  (when-not (zero? i)
    (<! (a/timeout 100))
    (>! in i)
    (recur (dec i))))

Collectives™ on Stack Overflow

Add delay in clojure.core.async

2 Answers 2

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related