meikel-q.clj

(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))