                                 (.printStackTrace e)
                                 e))
                   current-time (- (System/currentTimeMillis) start-ts)]
+              (>! ch new-value)
               (if (instance? Exception new-value)
-                (dosync
-                  (alter running-query
-                         merge {:last-event current-event
-                                :avg-time (next-avg
+                (do
+                  (close! ch)
+                  (dosync
+                   (alter virtual-streams dissoc projection-name)
+                   (alter running-query
+                          merge {:last-event current-event
+                                 :avg-time (next-avg
                                             (:avg-time @running-query)
                                             current-time
                                             (:processed @running-query))
-                                :processed (inc (:processed @running-query))
-                                :last-error new-value 
-                                :status :failed}))                
+                                 :processed (inc (:processed @running-query))
+                                 :last-error new-value 
+                                 :status :failed})))
                 (do
                   (dosync
-                    (alter running-query
-                           merge {:last-event current-event
-                                  :avg-time (next-avg
-                                              (:avg-time @running-query)
-                                              current-time
-                                              (:processed @running-query))
-                                  :current-value new-value
-                                  :processed (inc (:processed @running-query))}))
+                   (alter running-query
+                          merge {:last-event current-event
+                                 :avg-time (next-avg
+                                            (:avg-time @running-query)
+                                            current-time
+                                            (:processed @running-query))
+                                 :current-value new-value
+                                 :processed (inc (:processed @running-query))}))
                   (recur new-value (<! s))))))))))
   (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)
-    {:correct "true"}))
+                  ;; 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)
+                  {:correct "true"}))
 
 (defmethod stream "cold" [a-stream params]
   (log/info "cold-stream" (pr-str params))

  (init-stream-manager! [this])
  (update-streams! [this stream-name])
  (create-stream-endpoint! [this stream-name])
  (streams [this]))

(defmulti stream (fn [_ params]

(defonce publications (ref {}))
(defonce global-channels (ref {}))
(defonce active-streams (ref {}))
(defonce virtual-streams (ref {}))


;; Stream protocols and multimethods
