meikel-q.clj

(defn queue-process
  [input output stats]
  (async/go
    ; Wait for the first input.
    (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.
          ; Is the assumption that the process owns output right? 
          (identical? ch input) (do
                                  (doseq [v q] (async/>! output v))
                                  (async/close! output))
          
          ; Write happened.
          :else (recur (pop q)))))