 (defn kafka-stream
   [zk type project]
   (let [conf (kafka-config zk) 
-        topic (str "events_" type "_" project)]
-    (messages (consumer conf) [topic])))
+        topic (str "events_" type "_" project)
+        msgs (messages (consumer conf) [topic]) 
+        channel (chan)]
+    (go (doseq [m msgs] (>! channel m)))
+    (pub channel (fn [_] true))))
 
 (defn db-response
   [ent params]

  (:require [clj-kafka.core :refer [with-resource]]
            [cheshire.core :refer [parse-string generate-string]]
            [clojure.core.match :refer [match]]
            [clojure.core.async :refer [go <! filter< map< dropping-buffer chan close! go-loop >!]]
            [zoo-live.web.resp :refer :all]
            [korma.core :refer :all]
            [clojure.string :as str]


(defn- streaming-response
  [msgs {:keys [params] :as req}]
  (let [in-chan (chan)
        out-chan (->> (sub msgs true in-chan)  
                      (map< kafka-json-string-to-map)
                      (map< (comp #(str % "\n") generate-string)))]
    (with-channel req channel
      (send! channel (resp-ok (<!! out-chan) stream-mime) false)
      (let [writer (go-loop [msg (<! out-chan)]
                            (send! channel (resp-ok msg stream-mime) false)
                            (recur (<! out-chan)))]
        (on-close channel (fn [status] 
                            (close! writer) 
                            (close! in-chan)
                            (close! out-chan)))))))

(defn handle-request
  [msgs db-ent type project]
