   (let [in-chan (chan)
         out-chan (->> (sub msgs true in-chan)  
                       (map< kafka-json-string-to-map)
+                      (filter< (filter-stream params))
                       (map< filter-user-data)
                       (map< (comp #(str % "\n") generate-string)))]
     (with-channel req channel

    (go (doseq [m msgs] (>! channel m)))
    (pub channel (fn [_] true))))

(defn db-response
  [ent params & [mime]]
  (resp-ok (mapv filter-user-data (query-from-params ent params)) mime))


  [config [type project]]
  (let [msgs (kafka-stream (:zookeeper config) type project)
        db-ent (ent config type project)]
    (GET (str "/" type "/" project) [:as req] (handle-request msgs db-ent type project))))
