How to properly batch messages with core.async?

Question

I would like to batch messages on a core.async chan by count and timeout, (i.e. 10ms or 10 messages, whichever comes first). Tim Baldridge has a video on batching, but it uses deprecated functions in core.async and does not use transducers. I'm looking for something like the following...

(defn batch [in out max-time max-count]
  ...
 )

Oxidate

Answer 1

Transducers shouldn't really be a concern for a batching function – as a taker on the in channel, it will see values transformed by any transducers on that channel, and any takers listening on out will in turn see values transformed by that channel's transducer.

As for an implementation, the function below will take batches of max-count items from in, or however many arrive by max-time since the last batch was output, and output them to out, closing when the input channel closes, subject to the input channel's transducer (if any, and any takers listening on out will also have that channel's transducer applied as noted above):

(defn batch [in out max-time max-count]
  (let [lim-1 (dec max-count)]
    (async/go-loop [buf [] t (async/timeout max-time)]
      (let [[v p] (async/alts! [in t])]
        (cond
          (= p t)
          (do
            (async/>! out buf)
            (recur [] (async/timeout max-time)))

          (nil? v)
          (if (seq buf)
            (async/>! out buf))

          (== (count buf) lim-1)
          (do
            (async/>! out (conj buf v))
            (recur [] (async/timeout max-time)))

          :else
          (recur (conj buf v) t))))))

Expletive

Great piece of code, simple and correct. Used it for batching Redis PubSub messages (using out as a publisher). – Libra

Marvellous answer. – Sidesman

I was wondering whether clojure.core.async/take could be a good fit for this, but you basically need in any case to add the loop (and the "never ever block forever" timeout!) so at the end of the day, the implementation above still looks rock solid. – Ac

Where exactly in this code output chan gets closed when input channel is closed as stated in answer? – Secondrate

@OlimSaidov That's the (nil? v) case – Fraze

As far as I can see the (nil? v) case doesn't close out. I'm using a modified version that calls (close! out) in that branch. – Caffey