             [clojure.tools.cli :as cli]
             [clj-time.local :as tl]
             [clojure.core.reducers :as r]
+            [clojure.core.async :as async]
             [clojurewerkz.elastisch.rest :as esr]
             [clojurewerkz.elastisch.rest.document :as esrd]
             [qbits.alia :as alia])

    (map my-deref channels)
    @data))

(defn c-get-data
  "Get data from C*."
  [f-par-fetch host paths tenant rollup period from to]
  (println "Connecting to Cassandra...")
  (let [cluster (alia/cluster {:contact-points [host]})
        session (alia/connect cluster keyspace)
        fetch! (fetchq session)]
    (println "Getting data form Cassandra...")
    (try
      (let [data (time (doall (f-par-fetch session fetch! paths tenant
                                           rollup period from to)))]

    (println "Period:" period)
    (newline)
    (let [paths (es-get-paths eshost path tenant)
          data (c-get-data par-fetch "par-fetch" chost paths tenant rollup period from to)
          ;;fdata (flatter data (fn [data] (r/reduce into [] data)) "r/reduce")
          ]
      (c-get-data par-fetch-async "par-fetch-async" chost paths tenant rollup period from to)
      ;;(flatter data (fn [data] (reduce into data)) "reduce")
      ;;(cleaner fdata (fn [data] (into [] (r/remove nil? data))) "r/remove")
      ;;(cleaner fdata (fn [data] (remove nil? data))  "remove")
