       (if (nil? p)
         (let [c (chan 1)
               new-p {:channel c
-                     :p (pub c (fn [ev] (get ev "stream-name")))}]
+                     :p (pub c (fn [ev] (get ev "stream-name"
+                                             (get ev :stream-name))))}]
           (alter publications assoc async-stream new-p)
           new-p)
         p))))

  (log/info "cold-stream" (pr-str params))
  (let [ch (chan (buffer 1))
        full-s (data-from a-stream 
                          (get params "stream-name" "__all__")
                          (extract-date params))
        full-s (if (and (contains? params :limit)
                        (not (nil? (:limit params))))

  (process-event! [this msg]
    ;; Think about the order of store+send to taps
    (println "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! " (pr-str msg))
    (update-streams! this (get msg "stream-name"
                               (get msg :stream-name)))
    (go (>! (:channel (global-channel this)) msg)
        (>! (:channel (publisher this)) msg))
    (db/store db msg)
