de-async: Channel that returns channels

de_async.clj

(require '[clojure.core.async :refer (go go-loop <! >! >!! chan close! alts!)])

(defn de-async 
  "From is a channel of channels.
  Returns a new channel that has the values of the channels.
  Closes when all channels are closed."
  [from]
  (let [to (chan)]
    (go-loop [cs [from]]
      (if-not (empty? cs)
        (let [[v ch] (alts! cs)]
          (cond (nil? v)    (recur (vec (filter #(not= % ch) cs)))
                (= ch from) (recur (conj cs v))
                :else       (do (>! to v)
                                (recur cs))))
        (close! to)))
    to))

(defn cin [] (doto (chan 2)
  (>!! (doto (chan 2) (>!! 1) (>!! 2) (close!)))
  (>!! (doto (chan 1) (>!! 3) (close!)))
  (close!)))

(defn print-all [c]
  (go-loop [v (<! c)]
    (if-not (nil? v)
      (do (println "got value" v)
          (recur (<! c)))
      (println "Done!"))))

(print-all (de-async (cin)))