Fork me on GitHub

I was going to say that you could just do away with the timeout channels and just use a DelayQueue directly like the timeouts do


as long as you're targeting the JVM

Ben Sless02:02:38

I was thinking of trying a dynamic version of merge, This is the merge source

(defn merge
  "Takes a collection of source channels and returns a channel which
  contains all values taken from them. The returned channel will be
  unbuffered by default, or a buf-or-n can be supplied. The channel
  will close after all the source channels have closed."
  ([chs] (merge chs nil))
  ([chs buf-or-n]
     (let [out (chan buf-or-n)]
       (go-loop [cs (vec chs)]
         (if (pos? (count cs))
           (let [[v c] (alts! cs)]
             (if (nil? v)
               (recur (filterv #(not= c %) cs))
               (do (>! out v)
                   (recur cs))))
           (close! out)))
I can create a channel for each incoming element, deliver it to that channel after timeout, and always prune the closed channels as merge does. I worry that the performance of alts will be degrade for a large number of elements.