Hi again,
and some more golfing by Christophe:
(defn queue-process-uncontrolled
[input output stats]
(async/go
(loop [q clojure.lang.PersistentQueue/EMPTY]
(let [[val-to-q ch] (async/alts!
(if-let [v (peek q)]
[input [output v]]
[input]))]
(swap! stats update-stats-as-you-see-fit q)
(cond
; Read a value from input.
val-to-q (recur (conj q val-to-q))
; Input channel is closed. => drain queue.
(identical? ch input) (doseq [v q] (async/>! output v))
; Write happened.
:else (recur (pop q)))))
(defn queue-process-controlled
[input stats]
(let [output (async/chan)
process (queue-process-uncontrolled input output stats)]
(async/go
(<! process)
(async/close! output))
output))
Plus an improvement for the closing of the output channel.
queue-process-uncontrolled is not necessarily the master of the channel.
Meikel
--
--
You received this message because you are subscribed to the Google
Groups "Clojure" group.
To post to this group, send email to clojure@googlegroups.com
Note that posts from new members are moderated - please be patient with your
first post.
To unsubscribe from this group, send email to
clojure+unsubscr...@googlegroups.com
For more options, visit this group at
http://groups.google.com/group/clojure?hl=en
---
You received this message because you are subscribed to the Google Groups
"Clojure" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to clojure+unsubscr...@googlegroups.com.
For more options, visit https://groups.google.com/groups/opt_out.
