   (log/info "Initialising hot-cold stream with params" (pr-str params))
   (let [date (extract-date params)
         ch (chan (buffer 1))
-        stream-name (get params "stream-name" "__all__")
+        stream-name (get params "stream-name"
+                         (get params :stream-name "__all__"))
         full-s (data-from a-stream
                           stream-name
                           (extract-date params))]

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

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