    [clojurewerkz.elastisch.query            :as q]
    [clojurewerkz.elastisch.rest.response    :as esrsp]
    [com.stuartsierra.component :as component]
-   [clojure.core.async :as async :refer [<! <!! >! >!! chan put! alts! timeout go go-loop]]))
+   [clojure.core.async :as async :refer [<! <!! >! >!! chan put! alts! timeout go go-loop close!]]))
 
 (def conn (esr/connect (:es-address conf)))
 (def native-conn (esn/connect [["127.0.0.1" 9300]] {"cluster.name" "elasticsearch_mn"}))

         ;; and start it running. For example, connect to a
         ;; database, create thread pools, or initialize shared
         ;; state.
         (let []
           (run-persistence-loop (:persistence channels))
           (run-rt-persistence-loop (:rt-persistence channels))
           (run-percolation-loop (:percolation channels) (:percolation-matches channels))
           (run-find-missing-loop (:tweet-missing channels) (:missing-tweet-found channels))
           (run-query-loop (:query channels) (:query-results channels))

           ;; Return an updated version of the component with
           ;; the run-time state assoc'd in.
           (assoc component :connection conn)))

  (stop [component]
        (log/info "Stopping Persistence Component")


        ;; Return the component, optionally modified. Remember that if you
        ;; dissoc one of a record's base fields, you get a plain map.
        (doseq [p-loop (vals loops)]
          (close! p-loop))

        (assoc component :loops nil)))

(defn new-persistence [conf]
  (map->Persistence {:conf conf}))
