                      pub sub alts!]])
   (:import (java.net ServerSocket)
            (java.io DataOutput)))
+
 ;; TODO: Do something about the conflict between keywords and strings
 ;;       for the keys (e.g. stream-name)
 

                           (conj (:all-channels old-state) c)]
                       (merge old-state
                              {:publication new-p
                               :all-channels new-channels})))))
        new-p))))


    ch))

(defmethod stream->ch "hot" [a-stream params]
  (let [ch (chan 1024)
        stream-name (get params :stream-name "__all__")]
    (if (= stream-name "__all__")
      (tap (:mult-channel (:global-channel a-stream)) ch)
      (sub (:p (publisher a-stream)) stream-name ch))
    ch))

(defrecord AsyncStreamState [projections publication pub-buffer active-streams
                             virtual-streams all-channels])

(defn pipeline [num fn ch]
