     (go (doseq [m msgs] (>! channel m)))
     (pub channel (fn [_] true))))
 
-(defn db-response
+(defn- filter-map
+  [ev [k t]]
+  (cond 
+    (vector? t) ((first t) (k ev) (second t))
+    true (= (k ev) t)))
+
+(defn- filter-stream
+  [params]
+  (let [filter-map (reduce params-to-where {} (dissoc params :from :to :page :per_page))]
+    (fn [ev]
+      (every? (partial filter-test ev) filter-map))))
+
+(defn- db-response
   [ent params & [mime]]
   (resp-ok (mapv filter-user-data (query-from-params ent params)) mime))
 

       (apply str)
       parse-string))

(defn kafka-stream
  [zk type project]
  (let [conf (kafka-config zk) 
        topic (str "events_" type "_" project)

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